| 90 | } |
| 91 | |
| 92 | Status StreamingAggregationNode::GetRowsStreaming( |
| 93 | RuntimeState* state, RowBatch* out_batch) { |
| 94 | if (child_batch_ == nullptr) { |
| 95 | child_batch_.reset( |
| 96 | new RowBatch(child(0)->row_desc(), state->batch_size(), mem_tracker())); |
| 97 | } |
| 98 | |
| 99 | int num_aggs = aggs_.size(); |
| 100 | // Create mini batches. |
| 101 | vector<unique_ptr<RowBatch>> mini_batches; |
| 102 | if (!replicate_input_ && num_aggs > 1) { |
| 103 | for (int i = 0; i < num_aggs; ++i) { |
| 104 | mini_batches.push_back(make_unique<RowBatch>( |
| 105 | child(0)->row_desc(), state->batch_size(), mem_tracker())); |
| 106 | } |
| 107 | } |
| 108 | |
| 109 | do { |
| 110 | DCHECK_EQ(out_batch->num_rows(), 0); |
| 111 | RETURN_IF_CANCELLED(state); |
| 112 | |
| 113 | if (child_batch_processed_) { |
| 114 | DCHECK_EQ(child_batch_->num_rows(), 0); |
| 115 | RETURN_IF_ERROR(child(0)->GetNext(state, child_batch_.get(), &child_eos_)); |
| 116 | child_batch_processed_ = false; |
| 117 | } |
| 118 | |
| 119 | if (num_aggs == 1) { |
| 120 | RETURN_IF_ERROR(aggs_[0]->AddBatchStreaming( |
| 121 | state, out_batch, child_batch_.get(), &child_batch_processed_)); |
| 122 | // We're not guaranteed to be able to stream the entirety of 'child_batch_' into |
| 123 | // 'out_batch' as AddBatchStreaming() will attach all var-len data to 'out_batch' |
| 124 | // and 'child_batch_' may have been referencing data that wasn't attached to it. |
| 125 | if (child_batch_processed_) { |
| 126 | child_batch_->Reset(); |
| 127 | } |
| 128 | if (fast_limit_check_) { |
| 129 | DCHECK(limit() > -1); |
| 130 | if (aggs_[0]->GetNumKeys() >= limit()) { |
| 131 | child_eos_ = true; |
| 132 | child_batch_processed_ = true; |
| 133 | child_batch_->Reset(); |
| 134 | runtime_profile_->AddInfoString("FastLimitCheckExceededRows", |
| 135 | SimpleItoa(aggs_[0]->GetNumKeys() - limit())); |
| 136 | VLOG_QUERY << Substitute("the number of rows ($0) returned from the streaming " |
| 137 | "aggregation node has exceeded the limit of $1",aggs_[0]->GetNumKeys(), |
| 138 | limit()); |
| 139 | break; |
| 140 | } |
| 141 | } |
| 142 | continue; |
| 143 | } |
| 144 | |
| 145 | if (replicate_input_) { |
| 146 | bool eos = false; |
| 147 | while (replicate_agg_idx_ < num_aggs) { |
| 148 | RETURN_IF_ERROR(aggs_[replicate_agg_idx_]->AddBatchStreaming( |
| 149 | state, out_batch, child_batch_.get(), &eos)); |
nothing calls this directly
no test coverage detected