Method GetNext

be/src/exec/hdfs-scan-node.cc:82–118

Status HdfsScanNode::GetNext(RuntimeState* state, RowBatch* row_batch, bool* eos) {
  SCOPED_TIMER(runtime_profile_->total_time_counter());
  ScopedGetNextEventAdder ea(this, eos);

  if (!initial_ranges_issued_.Load()) {
    // We do this in GetNext() to maximise the amount of work we can do while waiting for
    // runtime filters to show up. The scanner threads have already started (in Open()),
    // so we need to tell them there is work to do.
    // TODO: This is probably not worth splitting the organisational cost of splitting
    // initialisation across two places. Move to before the scanner threads start.
    Status status = IssueInitialScanRanges(state);
    if (!status.ok()) {
      // If the status returned is CANCELLED, it could be because the
      // reader_context_ was cancelled by a scanner thread which hit an error. In this
      // case, the scanner thread's error must take precedence. In other cases,
      // the non-ok status represents the error in ValidateScanRange() or describes
      // the unsupported compression formats. For such non-CANCELLED cases, the status
      // returned by IssueInitialScanRanges() takes precedence.
      unique_lock<timed_mutex> l(lock_);
      if (status.IsCancelled() && !status_.ok()) return status_;
      return status;
    }

    // Release the scanner threads
    discard_result(ranges_issued_barrier_.Notify());

    if (shared_state_->progress().done()) SetDone();
  }

  Status status = GetNextInternal(state, row_batch, eos);
  if (!status.ok() || *eos) {
    unique_lock<timed_mutex> l(lock_);
    lock_guard<SpinLock> l2(file_type_counts_lock_);
    StopAndFinalizeCounters();
  }
  return status;
}

Status HdfsScanNode::GetNextInternal(
    RuntimeState* state, RowBatch* row_batch, bool* eos) {