Method InitNewRange

be/src/exec/sequence/hdfs-sequence-scanner.cc:92–117

}

Status HdfsSequenceScanner::InitNewRange() {
  DCHECK(header_ != nullptr);
  only_parsing_header_ = false;

  HdfsPartitionDescriptor* hdfs_partition = context_->partition_descriptor();

  text_converter_.reset(new TextConverter(hdfs_partition->escape_char(),
      scan_node_->hdfs_table()->null_column_value()));

  delimited_text_parser_.reset(new SequenceDelimitedTextParser(
      scan_node_->hdfs_table()->num_cols(), scan_node_->num_partition_keys(),
      scan_node_->is_materialized_col(), '\0', hdfs_partition->field_delim(),
      hdfs_partition->collection_delim(), hdfs_partition->escape_char()));

  num_buffered_records_in_compressed_block_ = 0;

  SeqFileHeader* seq_header = reinterpret_cast<SeqFileHeader*>(header_);
  if (seq_header->is_compressed) {
    RETURN_IF_ERROR(UpdateDecompressor(header_->codec));
  }

  // Initialize codegen fn
  RETURN_IF_ERROR(InitializeWriteTuplesFn(hdfs_partition,
      THdfsFileFormat::SEQUENCE_FILE, "HdfsSequenceScanner"));
  return Status::OK();
}

Status HdfsSequenceScanner::Open(ScannerContext* context) {
  RETURN_IF_ERROR(BaseSequenceScanner::Open(context));