Method GetNextOutputBatch

be/src/exec/analytic-eval-node.cc:728–765

}

Status AnalyticEvalNode::GetNextOutputBatch(
    RuntimeState* state, RowBatch* output_batch, bool* eos) {
  SCOPED_TIMER(evaluation_timer_);
  VLOG_FILE << id() << " GetNextOutputBatch: " << DebugStateString()
            << " tuple pool size:" << curr_tuple_pool_->total_allocated_bytes();
  if (input_stream_->rows_returned() == input_stream_->num_rows()) {
    *eos = true;
    return Status::OK();
  }

  const int num_child_tuples = child(0)->row_desc()->tuple_descriptors().size();
  RowBatch input_batch(child(0)->row_desc(), output_batch->capacity(), mem_tracker());
  int64_t stream_idx = input_stream_->rows_returned();
  RETURN_IF_ERROR(input_stream_->GetNext(&input_batch, eos));
  for (int i = 0; i < input_batch.num_rows(); ++i) {
    if (ReachedLimit()) break;
    DCHECK(!output_batch->AtCapacity());
    DCHECK(!result_tuples_.empty());
    VLOG_ROW << id() << " Output row idx=" << stream_idx << " " << DebugStateString(true);

    // CopyRow works as expected: input_batch tuples form a prefix of output_batch
    // tuples.
    TupleRow* dest = output_batch->GetRow(output_batch->AddRow());
    input_batch.CopyRow(input_batch.GetRow(i), dest);
    dest->SetTuple(num_child_tuples, result_tuples_.front().second);
    output_batch->CommitLastRow();
    IncrementNumRowsReturned(1);

    // Remove the head of result_tuples_ if all rows using that evaluated tuple
    // have been returned.
    DCHECK_LE(stream_idx, result_tuples_.front().first);
    if (stream_idx >= result_tuples_.front().first) result_tuples_.pop_front();
    ++stream_idx;
  }
  input_batch.TransferResourceOwnership(output_batch);
  if (ReachedLimit()) *eos = true;
  return Status::OK();
}

inline int64_t AnalyticEvalNode::NumOutputRowsReady() const {
  if (result_tuples_.empty()) return 0;