| 57 | } |
| 58 | |
| 59 | Status StreamingAggregationNode::GetNext( |
| 60 | RuntimeState* state, RowBatch* row_batch, bool* eos) { |
| 61 | SCOPED_TIMER(runtime_profile_->total_time_counter()); |
| 62 | ScopedGetNextEventAdder ea(this, eos); |
| 63 | RETURN_IF_ERROR(ExecDebugAction(TExecNodePhase::GETNEXT, state)); |
| 64 | RETURN_IF_CANCELLED(state); |
| 65 | |
| 66 | if (!fast_limit_check_ && ReachedLimit()) { |
| 67 | *eos = true; |
| 68 | return Status::OK(); |
| 69 | } |
| 70 | |
| 71 | // With multiple Aggregators, each will only set a single tuple per row. We rely on the |
| 72 | // other tuples to be null to detect which Aggregator set which row. |
| 73 | if (aggs_.size() > 1) row_batch->ClearTuplePointers(); |
| 74 | |
| 75 | if (!child_eos_ || !child_batch_processed_) { |
| 76 | // For streaming preaggregations, we process rows from the child as we go. |
| 77 | RETURN_IF_ERROR(GetRowsStreaming(state, row_batch)); |
| 78 | *eos = false; |
| 79 | } else { |
| 80 | bool aggregator_eos = false; |
| 81 | RETURN_IF_ERROR( |
| 82 | aggs_[curr_output_agg_idx_]->GetNext(state, row_batch, &aggregator_eos)); |
| 83 | if (aggregator_eos) ++curr_output_agg_idx_; |
| 84 | *eos = curr_output_agg_idx_ >= aggs_.size(); |
| 85 | } |
| 86 | |
| 87 | IncrementNumRowsReturned(row_batch->num_rows()); |
| 88 | COUNTER_SET(rows_returned_counter_, rows_returned()); |
| 89 | return Status::OK(); |
| 90 | } |
| 91 | |
| 92 | Status StreamingAggregationNode::GetRowsStreaming( |
| 93 | RuntimeState* state, RowBatch* out_batch) { |
no test coverage detected