Method GetNext

be/src/exec/buffered-plan-root-sink.cc:163–266

}

Status BufferedPlanRootSink::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_);
    *eos = false;

    // Cap the maximum number of results fetched by this call to GetNext so that the
    // resulting QueryResultSet does not consume excessive amounts of memory.
    num_results = min(num_results, MAX_FETCH_SIZE);

    // Track the number of rows read from the queue and the number of rows to read.
    int num_rows_read = 0;
    // If 'num_results' <= 0 then by default fetch FETCH_NUM_BATCHES batches.
    const int num_rows_to_read =
        num_results <= 0 ? FETCH_NUM_BATCHES * state->batch_size() : num_results;

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

    // Read from the queue until the query is cancelled or the sink is closed, eos is
    // hit, all requested rows have been read, or the timeout has been hit.
    while (!IsCancelledOrClosed(state) && !*eos && num_rows_read < num_rows_to_read
        && !timed_out) {
      // Wait for the queue to have rows in it.
      while (!IsCancelledOrClosed(state) && IsQueueEmpty(state)
          && sender_state_ == SenderState::ROWS_PENDING && !timed_out) {
        if (timeout_us == 0) {
          rows_available_.Wait(l);
        } else {
          // Wait fetch_rows_timeout_us_ - row_batches_get_wait_timer_ microseconds for
          // rows to become available before returning to the client. Subtracting
          // wait_timeout_timer ensures the client only ever waits up to
          // fetch_rows_timeout_us_ microseconds before returning.
          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)));
          SCOPED_TIMER(row_batches_get_wait_timer_);
          timed_out = !rows_available_.WaitFor(l, wait_duration_us);
        }
      }

      // If the query was cancelled while the sink was waiting for rows to become
      // available, or if the query was cancelled before the current call to GetNext, set
      // eos and then return. The queue could be empty if the sink was closed while
      // waiting for rows to become available, or if the sink was closed before the
      // current call to GetNext.
      if (!IsCancelledOrClosed(state) && !IsQueueEmpty(state)) {
        // If current_batch_ is empty, then read directly from the queue.
        if (current_batch_row_ == 0) {
          // Debug action before GetBatch is called.
          RETURN_IF_ERROR(DebugAction(state->query_options(), "BPRS_BEFORE_GET_BATCH"));