| 119 | } |
| 120 | |
| 121 | Status AggregationNode::GetNext(RuntimeState* state, RowBatch* row_batch, bool* eos) { |
| 122 | SCOPED_TIMER(runtime_profile_->total_time_counter()); |
| 123 | ScopedGetNextEventAdder ea(this, eos); |
| 124 | RETURN_IF_ERROR(ExecDebugAction(TExecNodePhase::GETNEXT, state)); |
| 125 | RETURN_IF_CANCELLED(state); |
| 126 | |
| 127 | if (curr_output_agg_idx_ >= aggs_.size() || ReachedLimit()) { |
| 128 | *eos = true; |
| 129 | return Status::OK(); |
| 130 | } |
| 131 | |
| 132 | // With multiple Aggregators, each will only set a single tuple per row. We rely on the |
| 133 | // other tuples to be null to detect which Aggregator set which row. |
| 134 | if (aggs_.size() > 1) row_batch->ClearTuplePointers(); |
| 135 | |
| 136 | bool pagg_eos = false; |
| 137 | RETURN_IF_ERROR(aggs_[curr_output_agg_idx_]->GetNext(state, row_batch, &pagg_eos)); |
| 138 | if (pagg_eos) ++curr_output_agg_idx_; |
| 139 | |
| 140 | *eos = ReachedLimit() || (pagg_eos && curr_output_agg_idx_ >= aggs_.size()); |
| 141 | IncrementNumRowsReturned(row_batch->num_rows()); |
| 142 | COUNTER_SET(rows_returned_counter_, rows_returned()); |
| 143 | return Status::OK(); |
| 144 | } |
| 145 | |
| 146 | void AggregationNode::Close(RuntimeState* state) { |
| 147 | if (is_closed()) return; |
no test coverage detected