Method GetNextInternal

be/src/exec/base-sequence-scanner.cc:158–215

}

Status BaseSequenceScanner::GetNextInternal(RowBatch* row_batch) {
  if (only_parsing_header_) {
    DCHECK(header_ == nullptr);
    eos_ = true;
    header_ = state_->obj_pool()->Add(AllocateFileHeader());
    Status status = ReadFileHeader();
    if (!status.ok()) {
      scan_node_->UpdateRemainingScanRangeSubmissions(-1);
      RETURN_IF_ERROR(state_->LogOrReturnError(status.msg()));
      // We need to complete the ranges for this file.
      CloseFileRanges(stream_->filename());
      return Status::OK();
    }
    // Header is parsed, set the metadata in the scan node and issue more ranges.
    scan_node_->SetFileMetadata(
        context_->partition_descriptor()->id(), stream_->filename(), header_);
    const HdfsFileDesc* desc = scan_node_->GetFileDesc(
        context_->partition_descriptor()->id(), stream_->filename());
    // Issue the scan range with priority since it would result in producing a RowBatch.
    status = scan_node_->AddDiskIoRanges(desc, EnqueueLocation::HEAD);
    scan_node_->UpdateRemainingScanRangeSubmissions(-1);
    return status;
  }
  if (eos_) return Status::OK();

  int64_t tuple_buffer_size;
  RETURN_IF_ERROR(
      row_batch->ResizeAndAllocateTupleBuffer(state_, &tuple_buffer_size, &tuple_mem_));
  tuple_ = reinterpret_cast<Tuple*>(tuple_mem_);
  DCHECK_GT(row_batch->capacity(), 0);

  Status status = ProcessRange(row_batch);
  if (!status.ok()) {
    // Log error from file format parsing.
    // TODO(IMPALA-5922): Include the file and offset in errors inside the scanners.
    if (!status.IsCancelled() &&
        !status.IsMemLimitExceeded() &&
        !status.IsInternalError() &&
        !status.IsDiskIoError() &&
        !status.IsThreadPoolError()) {
      state_->LogError(ErrorMsg(TErrorCode::SEQUENCE_SCANNER_PARSE_ERROR,
          stream_->filename(), stream_->file_offset(),
          (stream_->eof() ? "(EOF)" : "")));
    }

    // This checks for abort_on_error.
    RETURN_IF_ERROR(state_->LogOrReturnError(status.msg()));

    // Recover by skipping to the next sync.
    parse_status_ = Status::OK();
    int64_t error_offset = stream_->file_offset();
    status = SkipToSync(header_->sync, SYNC_HASH_SIZE);
    COUNTER_ADD(bytes_skipped_counter_, stream_->file_offset() - error_offset);
    RETURN_IF_ERROR(status);
    DCHECK(parse_status_.ok());
  }
  return Status::OK();
}