| 274 | } |
| 275 | |
| 276 | Status GroupingAggregator::GetRowsFromPartition( |
| 277 | RuntimeState* state, RowBatch* row_batch) { |
| 278 | DCHECK(!row_batch->AtCapacity()); |
| 279 | if (output_iterator_.AtEnd()) { |
| 280 | // Done with this partition, move onto the next one. |
| 281 | if (output_partition_ != nullptr) { |
| 282 | output_partition_->Close(false); |
| 283 | output_partition_ = nullptr; |
| 284 | // Try to save the large write page reservation (if it's used) after closing |
| 285 | // a partition. |
| 286 | if (!large_write_page_reservation_.is_closed() |
| 287 | && large_write_page_reservation_.GetReservation() == 0) { |
| 288 | TrySaveLargeWritePageReservation(); |
| 289 | } |
| 290 | } |
| 291 | if (aggregated_partitions_.empty() && spilled_partitions_.empty()) { |
| 292 | // No more partitions, all done. |
| 293 | partition_eos_ = true; |
| 294 | return Status::OK(); |
| 295 | } |
| 296 | // Process next partition. |
| 297 | RETURN_IF_ERROR(NextPartition()); |
| 298 | } |
| 299 | DCHECK(output_partition_ != nullptr); |
| 300 | |
| 301 | SCOPED_TIMER(get_results_timer_); |
| 302 | |
| 303 | // The output row batch may reference memory allocated by Serialize() or Finalize(), |
| 304 | // allocating that memory directly from the row batch's pool means we can safely return |
| 305 | // the batch. |
| 306 | vector<ScopedResultsPool> allocate_from_batch_pool = ScopedResultsPool::Create( |
| 307 | output_partition_->agg_fn_evals, row_batch->tuple_data_pool()); |
| 308 | int count = 0; |
| 309 | const int N = BitUtil::RoundUpToPowerOfTwo(state->batch_size()); |
| 310 | // Keeping returning rows from the current partition. |
| 311 | while (!output_iterator_.AtEnd() && !row_batch->AtCapacity()) { |
| 312 | // This loop can go on for a long time if the conjuncts are very selective. Do query |
| 313 | // maintenance every N iterations. |
| 314 | if ((count++ & (N - 1)) == 0) { |
| 315 | RETURN_IF_CANCELLED(state); |
| 316 | RETURN_IF_ERROR(QueryMaintenance(state)); |
| 317 | } |
| 318 | |
| 319 | int row_idx = row_batch->AddRow(); |
| 320 | TupleRow* row = row_batch->GetRow(row_idx); |
| 321 | Tuple* intermediate_tuple = output_iterator_.GetTuple<BucketType::MATCH_UNSET>(); |
| 322 | Tuple* output_tuple = GetOutputTuple(output_partition_->agg_fn_evals, |
| 323 | intermediate_tuple, row_batch->tuple_data_pool()); |
| 324 | output_iterator_.Next(); |
| 325 | row->SetTuple(agg_idx_, output_tuple); |
| 326 | DCHECK_EQ(conjunct_evals_.size(), conjuncts_.size()); |
| 327 | if (ExecNode::EvalConjuncts(conjunct_evals_.data(), conjuncts_.size(), row)) { |
| 328 | row_batch->CommitLastRow(); |
| 329 | ++num_rows_returned_; |
| 330 | if (ReachedLimit()) break; |
| 331 | } |
| 332 | } |
| 333 |
nothing calls this directly
no test coverage detected