Method GetNext

be/src/exec/analytic-eval-node.cc:784–840

}

Status AnalyticEvalNode::GetNext(RuntimeState* state, RowBatch* row_batch, bool* eos) {
  SCOPED_TIMER(runtime_profile_->total_time_counter());
  ScopedGetNextEventAdder ea(this, eos);
  RETURN_IF_ERROR(ExecDebugAction(TExecNodePhase::GETNEXT, state));
  RETURN_IF_CANCELLED(state);
  RETURN_IF_ERROR(QueryMaintenance(state));
  VLOG_FILE << id() << " GetNext: " << DebugStateString();
  DCHECK(!input_stream_->is_closed()); // input_stream_ is closed if we already hit eos.

  if (ReachedLimit()) {
    // TODO: This transfer is simple and correct, but not necessarily efficient. We
    // should optimize the use/transfer of memory to better amortize allocations
    // over multiple Reset()/Open()/GetNext()* cycles.
    row_batch->tuple_data_pool()->AcquireData(prev_tuple_pool_.get(), false);
    row_batch->tuple_data_pool()->AcquireData(curr_tuple_pool_.get(), false);
    DCHECK(!input_stream_->is_closed());
    // Flush resources in case we are in a subplan and need to allocate more blocks
    // when the node is reopened.
    input_stream_->Close(row_batch, RowBatch::FlushMode::FLUSH_RESOURCES);
    *eos = true;
    return Status::OK();
  } else {
    *eos = false;
  }

  RETURN_IF_ERROR(ProcessChildBatches(state));

  bool output_eos = false;
  RETURN_IF_ERROR(GetNextOutputBatch(state, row_batch, &output_eos));
  if (input_eos_ && output_eos) {
    // Transfer the ownership of all row-backing resources on eos for simplicity.
    // TODO: This transfer is simple and correct, but not necessarily efficient. We
    // should optimize the use/transfer of memory to better amortize allocations
    // over multiple Reset()/Open()/GetNext()* cycles.
    row_batch->tuple_data_pool()->AcquireData(prev_tuple_pool_.get(), false);
    row_batch->tuple_data_pool()->AcquireData(curr_tuple_pool_.get(), false);
    // Flush resources in case we are in a subplan and need to allocate more blocks
    // when the node is reopened.
    input_stream_->Close(row_batch, RowBatch::FlushMode::FLUSH_RESOURCES);
    *eos = true;
  } else if (prev_pool_last_result_idx_ != -1 &&
      prev_pool_last_result_idx_ < input_stream_->rows_returned() &&
      prev_pool_last_window_idx_ < window_tuples_.front().first) {
    // Transfer resources to the output row batch if enough have accumulated and they're
    // no longer needed by output rows to be returned later.
    VLOG_FILE << id() << " Transfer prev pool to output batch, "
              << " pool size: " << prev_tuple_pool_->total_allocated_bytes()
              << " last result idx: " << prev_pool_last_result_idx_
              << " last window idx: " << prev_pool_last_window_idx_;
    row_batch->tuple_data_pool()->AcquireData(prev_tuple_pool_.get(), !*eos);
    prev_pool_last_result_idx_ = -1;
    prev_pool_last_window_idx_ = -1;
  }

  COUNTER_SET(rows_returned_counter_, rows_returned());
  return Status::OK();
}