<commit_before>// Copyright 2012 Cloudera Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

#include "codegen/llvm-codegen.h"
#include "exec/delimited-text-parser.h"
#include "exec/delimited-text-parser.inline.h"
#include "exec/hdfs-scan-node.h"
#include "exec/hdfs-text-scanner.h"
#include "exec/text-converter.h"
#include "exec/text-converter.inline.h"
#include "runtime/row-batch.h"
#include "runtime/runtime-state.h"
#include "util/cpu-info.h"
#include "util/debug-util.h"

using namespace boost;
using namespace impala;
using namespace llvm;
using namespace std;

const char* HdfsTextScanner::LLVM_CLASS_NAME = "class.impala::HdfsTextScanner";

HdfsTextScanner::HdfsTextScanner(HdfsScanNode* scan_node, RuntimeState* state) 
    : HdfsScanner(scan_node, state),
      byte_buffer_ptr_(NULL),
      byte_buffer_end_(NULL),
      byte_buffer_read_size_(0),
      boundary_mem_pool_(new MemPool()),
      boundary_row_(boundary_mem_pool_.get()),
      boundary_column_(boundary_mem_pool_.get()),
      slot_idx_(0),
      error_in_row_(false) {
}

HdfsTextScanner::~HdfsTextScanner() {
}

void HdfsTextScanner::IssueInitialRanges(HdfsScanNode* scan_node, 
    const vector<HdfsFileDesc*>& files) {
  // Text just issues all ranges at once
  for (int i = 0; i < files.size(); ++i) {
    scan_node->AddDiskIoRange(files[i]);
  }
}

Status HdfsTextScanner::ProcessSplit(ScannerContext* context) {
  // Reset state for new scan range
  InitNewRange(context);

  // Find the first tuple.  If tuple_found is false, it means we went through the entire
  // scan range without finding a single tuple.  The bytes will be picked up
  // by the scan range before.
  bool tuple_found;
  RETURN_IF_ERROR(FindFirstTuple(&tuple_found));

  if (tuple_found) {
    // Process the scan range
    int dummy_num_tuples;
    RETURN_IF_ERROR(ProcessRange(&dummy_num_tuples, false));

    // Finish up reading past the scan range
    RETURN_IF_ERROR(FinishScanRange());
  }

  // All done with this scan range.
  return Status::OK;
}

Status HdfsTextScanner::Close() {
  context_->AcquirePool(boundary_mem_pool_.get());
  // We must flush any pending batches in the row batch before telling the scan node
  // the range is complete. 
  context_->Close();
  scan_node_->RangeComplete(THdfsFileFormat::TEXT, THdfsCompression::NONE);
  
  COUNTER_UPDATE(scan_node_->memory_used_counter(),
      boundary_mem_pool_->peak_allocated_bytes());
  return Status::OK;
}

void HdfsTextScanner::InitNewRange(ScannerContext* context) {
  SetContext(context);
  stream_->set_read_past_buffer_size(NEXT_BLOCK_READ_SIZE);
  HdfsPartitionDescriptor* hdfs_partition = context_->partition_descriptor();
  
  char field_delim = hdfs_partition->field_delim();
  char collection_delim = hdfs_partition->collection_delim();
  if (scan_node_->materialized_slots().size() == 0) {
    field_delim = '\0';
    collection_delim = '\0';
  }
  
  delimited_text_parser_.reset(new DelimitedTextParser(scan_node_,
      hdfs_partition->line_delim(), field_delim, collection_delim,
      hdfs_partition->escape_char()));
  text_converter_.reset(new TextConverter(hdfs_partition->escape_char()));

  ResetScanner();
}

void HdfsTextScanner::ResetScanner() {
  error_in_row_ = false;
  
  // Note - this initialisation relies on the assumption that N partition keys will occupy
  // entries 0 through N-1 in column_idx_to_slot_idx. If this changes, we will need
  // another layer of indirection to map text-file column indexes onto the
  // column_idx_to_slot_idx table used below.
  slot_idx_ = 0;
  
  boundary_column_.Clear();
  boundary_row_.Clear();
  delimited_text_parser_->ParserReset();

  template_tuple_ = context_->template_tuple();
  partial_tuple_empty_ = true;
  byte_buffer_ptr_ = byte_buffer_end_ = NULL;

  partial_tuple_ = reinterpret_cast<Tuple*>(
      boundary_mem_pool_->Allocate(scan_node_->tuple_desc()->byte_size()));

  // Initialize codegen fn
  InitializeCodegenFn(
      context_->partition_descriptor(), THdfsFileFormat::TEXT, "HdfsTextScanner");
}

Status HdfsTextScanner::FinishScanRange() {
  // For text we always need to scan past the scan range to find the next delimiter
  while (true) {
    bool eosr;
    Status status = FillByteBuffer(&eosr, NEXT_BLOCK_READ_SIZE);<commit_msg>Fix text scanner limit bug.<commit_after>// Copyright 2012 Cloudera Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

#include "codegen/llvm-codegen.h"
#include "exec/delimited-text-parser.h"
#include "exec/delimited-text-parser.inline.h"
#include "exec/hdfs-scan-node.h"
#include "exec/hdfs-text-scanner.h"
#include "exec/text-converter.h"
#include "exec/text-converter.inline.h"
#include "runtime/row-batch.h"
#include "runtime/runtime-state.h"
#include "util/cpu-info.h"
#include "util/debug-util.h"

using namespace boost;
using namespace impala;
using namespace llvm;
using namespace std;

const char* HdfsTextScanner::LLVM_CLASS_NAME = "class.impala::HdfsTextScanner";

HdfsTextScanner::HdfsTextScanner(HdfsScanNode* scan_node, RuntimeState* state) 
    : HdfsScanner(scan_node, state),
      byte_buffer_ptr_(NULL),
      byte_buffer_end_(NULL),
      byte_buffer_read_size_(0),
      boundary_mem_pool_(new MemPool()),
      boundary_row_(boundary_mem_pool_.get()),
      boundary_column_(boundary_mem_pool_.get()),
      slot_idx_(0),
      error_in_row_(false) {
}

HdfsTextScanner::~HdfsTextScanner() {
}

void HdfsTextScanner::IssueInitialRanges(HdfsScanNode* scan_node, 
    const vector<HdfsFileDesc*>& files) {
  // Text just issues all ranges at once
  for (int i = 0; i < files.size(); ++i) {
    scan_node->AddDiskIoRange(files[i]);
  }
}

Status HdfsTextScanner::ProcessSplit(ScannerContext* context) {
  // Reset state for new scan range
  InitNewRange(context);

  // Find the first tuple.  If tuple_found is false, it means we went through the entire
  // scan range without finding a single tuple.  The bytes will be picked up
  // by the scan range before.
  bool tuple_found;
  RETURN_IF_ERROR(FindFirstTuple(&tuple_found));

  if (tuple_found) {
    // Process the scan range
    int dummy_num_tuples;
    RETURN_IF_ERROR(ProcessRange(&dummy_num_tuples, false));

    // Finish up reading past the scan range
    RETURN_IF_ERROR(FinishScanRange());
  }

  // All done with this scan range.
  return Status::OK;
}

Status HdfsTextScanner::Close() {
  context_->AcquirePool(boundary_mem_pool_.get());
  // We must flush any pending batches in the row batch before telling the scan node
  // the range is complete. 
  context_->Close();
  scan_node_->RangeComplete(THdfsFileFormat::TEXT, THdfsCompression::NONE);
  
  COUNTER_UPDATE(scan_node_->memory_used_counter(),
      boundary_mem_pool_->peak_allocated_bytes());
  return Status::OK;
}

void HdfsTextScanner::InitNewRange(ScannerContext* context) {
  SetContext(context);
  stream_->set_read_past_buffer_size(NEXT_BLOCK_READ_SIZE);
  HdfsPartitionDescriptor* hdfs_partition = context_->partition_descriptor();
  
  char field_delim = hdfs_partition->field_delim();
  char collection_delim = hdfs_partition->collection_delim();
  if (scan_node_->materialized_slots().size() == 0) {
    field_delim = '\0';
    collection_delim = '\0';
  }
  
  delimited_text_parser_.reset(new DelimitedTextParser(scan_node_,
      hdfs_partition->line_delim(), field_delim, collection_delim,
      hdfs_partition->escape_char()));
  text_converter_.reset(new TextConverter(hdfs_partition->escape_char()));

  ResetScanner();
}

void HdfsTextScanner::ResetScanner() {
  error_in_row_ = false;
  
  // Note - this initialisation relies on the assumption that N partition keys will occupy
  // entries 0 through N-1 in column_idx_to_slot_idx. If this changes, we will need
  // another layer of indirection to map text-file column indexes onto the
  // column_idx_to_slot_idx table used below.
  slot_idx_ = 0;
  
  boundary_column_.Clear();
  boundary_row_.Clear();
  delimited_text_parser_->ParserReset();

  template_tuple_ = context_->template_tuple();
  partial_tuple_empty_ = true;
  byte_buffer_ptr_ = byte_buffer_end_ = NULL;

  partial_tuple_ = reinterpret_cast<Tuple*>(
      boundary_mem_pool_->Allocate(scan_node_->tuple_desc()->byte_size()));

  // Initialize codegen fn
  InitializeCodegenFn(
      context_->partition_descriptor(), THdfsFileFormat::TEXT, "HdfsTextScanner");
}

Status HdfsTextScanner::FinishScanRange() {
  if (scan_node_->ReachedLimit()) return Status::OK;

  // For text we always need to scan past the scan range to find the next delimiter
  while (true) {
    bool eosr;
    Status status = FillByteBuffer(&eosr, NEXT_BLOCK_READ_SIZE);
{'fraction_non_alphanumeric': 0.10882230545151894, 'fraction_numerical': 0.0035372451102788183, 'mean_word_length': 4.531070195627158, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 48, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2340944', 'n_tokens_mistral': 3118, 'n_tokens_neox': 3025, 'n_words': 769}