Method ProcessRange

be/src/exec/sequence/hdfs-sequence-scanner.cc:345–421

}

Status HdfsSequenceScanner::ProcessRange(RowBatch* row_batch) {
  SeqFileHeader* seq_header = reinterpret_cast<SeqFileHeader*>(header_);
  // Block compressed is handled separately to minimize function calls.
  if (seq_header->is_compressed && !seq_header->is_row_compressed) {
    return ProcessBlockCompressedScanRange(row_batch);
  }

  // We count the time here since there is too much overhead to do
  // this on each record.
  SCOPED_TIMER(scan_node_->materialize_tuple_timer());
  int64_t num_rows_read = 0;

  const bool copy_strings = !seq_header->is_compressed && !string_slot_offsets_.empty();
  const bool has_materialized_slots = !scan_node_->materialized_slots().empty();
  while (!eos_) {
    DCHECK_GT(record_locations_.size(), 0);
    TupleRow* tuple_row_mem = row_batch->GetRow(row_batch->AddRow());

    // Get the next compressed or uncompressed record and parse it.
    RETURN_IF_ERROR(GetRecord(&record_locations_[0].record, &record_locations_[0].len));
    bool add_row = false;
    if (has_materialized_slots) {
      char* col_start;
      uint8_t* record_start = record_locations_[0].record;
      int num_tuples = 0;
      int num_fields = 0;
      char* row_end_loc;
      RETURN_IF_ERROR(delimited_text_parser_->ParseFieldLocations(
          1, record_locations_[0].len, reinterpret_cast<char**>(&record_start),
          &row_end_loc, field_locations_.data(), &num_tuples, &num_fields, &col_start));
      DCHECK_EQ(num_tuples, 1);

      uint8_t error_in_row = false;
      uint8_t errors[num_fields];
      memset(errors, 0, num_fields);
      MemPool* pool = row_batch->tuple_data_pool();
      add_row = WriteCompleteTuple(pool, field_locations_.data(),
          tuple_, tuple_row_mem, template_tuple_, &errors[0], &error_in_row);
      if (UNLIKELY(error_in_row)) {
        ReportTupleParseError(field_locations_.data(), errors);
        RETURN_IF_ERROR(parse_status_);
      }
      if (add_row && copy_strings) {
        if (UNLIKELY(!tuple_->CopyStrings("HdfsSequenceScanner::ProcessRange()",
              state_, string_slot_offsets_.data(), string_slot_offsets_.size(), pool,
              &parse_status_))) {
          return parse_status_;
        }
      }
    } else {
      add_row = WriteTemplateTuples(tuple_row_mem, 1) > 0;
    }
    num_rows_read++;
    if (add_row) RETURN_IF_ERROR(CommitRows(1, row_batch));

    // Sequence files don't end with syncs.
    if (stream_->eof()) {
      eos_ = true;