| 393 | } |
| 394 | |
| 395 | void GroupingAggregator::CleanupHashTbl( |
| 396 | const vector<AggFnEvaluator*>& agg_fn_evals, HashTable::Iterator it) { |
| 397 | if (!needs_finalize_ && !needs_serialize_) return; |
| 398 | |
| 399 | // Iterate through the remaining rows in the hash table and call Serialize/Finalize on |
| 400 | // them in order to free any memory allocated by UDAs. |
| 401 | if (needs_finalize_) { |
| 402 | // Finalize() requires a dst tuple but we don't actually need the result, |
| 403 | // so allocate a single dummy tuple to avoid accumulating memory. |
| 404 | Tuple* dummy_dst = nullptr; |
| 405 | dummy_dst = Tuple::Create(output_tuple_desc_->byte_size(), tuple_pool_.get()); |
| 406 | while (!it.AtEnd()) { |
| 407 | Tuple* tuple = it.GetTuple<BucketType::MATCH_UNSET>(); |
| 408 | AggFnEvaluator::Finalize(agg_fn_evals, tuple, dummy_dst); |
| 409 | it.Next(); |
| 410 | // Free any expr result allocations to prevent them accumulating excessively. |
| 411 | expr_results_pool_->Clear(); |
| 412 | } |
| 413 | } else { |
| 414 | while (!it.AtEnd()) { |
| 415 | Tuple* tuple = it.GetTuple<BucketType::MATCH_UNSET>(); |
| 416 | AggFnEvaluator::Serialize(agg_fn_evals, tuple); |
| 417 | it.Next(); |
| 418 | // Free any expr result allocations to prevent them accumulating excessively. |
| 419 | expr_results_pool_->Clear(); |
| 420 | } |
| 421 | } |
| 422 | } |
| 423 | |
| 424 | Status GroupingAggregator::Reset(RuntimeState* state, RowBatch* row_batch) { |
| 425 | DCHECK(!is_streaming_preagg_) << "Cannot reset preaggregation"; |