Method GetNextInternal

be/src/exec/json/hdfs-json-scanner.cc:141–193

}

Status HdfsJsonScanner::GetNextInternal(RowBatch* row_batch) {
  DCHECK(!eos_);
  DCHECK_GE(scanner_state_, OPENED);
  DCHECK_NE(scanner_state_, FINISHED);

  current_pool_ = row_batch->tuple_data_pool();

  if (scanner_state_ == OPENED) {
    // Find the first tuple. If scanner_state_ is not SCANNING, it means we went through
    // the entire scan range without finding a single tuple. The bytes will be picked up
    // by the previous scan range in the same file.
    RETURN_IF_ERROR(FindFirstTuple());
    if (scanner_state_ != SCANNING) {
      eos_ = true;
      scanner_state_ = FINISHED;
      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_);
  tuple_row_ = row_batch->GetRow(row_batch->AddRow());

  while (scanner_state_ == SCANNING) {
    num_tuples_materialized_ = 0;
    int num_tuples = 0;
    int max_tuples = row_batch->capacity() - row_batch->num_rows();

    RETURN_IF_ERROR(ParseWrapper(max_tuples, &num_tuples));
    COUNTER_ADD(scan_node_->rows_read_counter(), num_tuples);

    // Because the processes of parsing JSON, materializing tuples, and even reading data
    // are intertwined, it can be expensive to accurately time them individually.
    // Therefore, we using this method to measure the time it takes to materialize tuples,
    // please note that the value obtained will always be inflated because the time it
    // takes to parse JSON is also included.
    // TODO: find a better way.
    COUNTER_SET(scan_node_->materialize_tuple_timer(),
        parse_json_timer_->value() - get_buffer_timer_->value());

    RETURN_IF_ERROR(CommitRows(num_tuples_materialized_, row_batch));

    if (row_batch->AtCapacity() || scan_node_->ReachedLimitShared()) break;
  }

  if (scanner_state_ >= PAST_SCANNING || scan_node_->ReachedLimitShared()) {
    eos_ = true;
    scanner_state_ = FINISHED;
  }
  return Status::OK();
}

Status HdfsJsonScanner::FindFirstTuple() {
  DCHECK_EQ(scanner_state_, OPENED);