Method GetNext

be/src/exec/blocking-plan-root-sink.cc:114–164

}

Status BlockingPlanRootSink::GetNext(RuntimeState* state, QueryResultSet* results,
    int num_results, bool* eos, int64_t timeout_us) {
  // Used to track how long the consumer waits for RowBatches to be produced and
  // materialized.
  DCHECK_GE(timeout_us, 0);
  MonotonicStopWatch wait_timeout_timer;
  wait_timeout_timer.Start();

  unique_lock<mutex> l(lock_);

  // Set the shared QueryResultSet pointer 'results_' to the given 'results' object and
  // wake up the sender thread so it can add rows to 'results_'.
  results_ = results;
  num_rows_requested_ = num_results;
  sender_cv_.NotifyAll();

  // True if the consumer timed out waiting for the producer to send rows, false
  // otherwise.
  bool timed_out = false;

  // Wait while the sender is still producing rows and hasn't filled in the current
  // result set.
  while (sender_state_ == SenderState::ROWS_PENDING && results_ != nullptr
      && !state->is_cancelled() && !timed_out) {
    if (timeout_us == 0) {
      consumer_cv_.Wait(l);
    } else {
      // It is possible for the timeout to expire, and for the QueryResultSet to still
      // have some rows appended to it. This can happen if the producer acquires the lock,
      // the timeout expires, and then the producer appends rows to the QueryResultSet.
      // This does not affect correctness because the producer always sets 'results_' to
      // nullptr if it appends any rows to the QueryResultSet and it always appends either
      // an entire RowBatch, or as many rows as requested.
      int64_t wait_duration_us = max(static_cast<int64_t>(1),
          timeout_us - static_cast<int64_t>(
                           round(wait_timeout_timer.ElapsedTime() / NANOS_PER_MICRO)));
      if (!consumer_cv_.WaitFor(l, wait_duration_us)) {
        VLOG_QUERY << "Fetch timed out";
        timed_out = true;

        // If the consumer timed out, make sure results_ is set to nullptr because the
        // consumer will destroy the current QueryResultSet and create a new one for the
        // next fetch request.
        results_ = nullptr;
      }
    }
  }

  *eos = sender_state_ == SenderState::EOS;
  return state->GetQueryStatus();
}
}