| 462 | } |
| 463 | |
| 464 | void HashAggregation::maybeIncreasePartialAggregationMemoryUsage( |
| 465 | double aggregationPct) { |
| 466 | // If more than this many are unique at full memory, give up on partial agg. |
| 467 | BOLT_DCHECK(isPartialOutput_); |
| 468 | // If size is at max and there still is not enough reduction, abandon partial |
| 469 | // aggregation. |
| 470 | auto calShouldAbandon = [&]() { |
| 471 | return abandonPartialAggregationEarly(numOutputRows_) || |
| 472 | (aggregationPct > abandonPartialAggregationMinFinalPct_ && |
| 473 | maxPartialAggregationMemoryUsage_ >= |
| 474 | maxExtendedPartialAggregationMemoryUsage_); |
| 475 | }; |
| 476 | |
| 477 | bool shouldAbandon = calShouldAbandon(); |
| 478 | if (shouldAbandon && adaptiveAdjustment_) { |
| 479 | uint64_t totalRowCnt{0}, processedRowCnt{0}; |
| 480 | operatorCtx_->traverseOpToGetRowCount(totalRowCnt, processedRowCnt); |
| 481 | if (totalRowCnt > 0 && processedRowCnt > 0) { |
| 482 | // left unprocessed rows * (input->output expansion/filter ratio) * size |
| 483 | auto calculatedSkippedSize = avgRowSize_ * |
| 484 | (totalRowCnt - processedRowCnt) * totalInputRows_ / processedRowCnt; |
| 485 | if (calculatedSkippedSize >= skippedDataSizeThreshold_) { |
| 486 | *const_cast<int32_t*>(&abandonPartialAggregationMinPct_) = |
| 487 | std::max((int32_t)abandonPartialAggregationMinPct_, 95); |
| 488 | *const_cast<int32_t*>(&abandonPartialAggregationMinFinalPct_) = |
| 489 | std::max((int32_t)abandonPartialAggregationMinFinalPct_, 90); |
| 490 | *const_cast<int32_t*>(&partialAggregationSpillMaxPct_) = |
| 491 | std::max((int32_t)partialAggregationSpillMaxPct_, 85); |
| 492 | adaptiveAdjustment_ = false; |
| 493 | // recalculate abandon or not |
| 494 | shouldAbandon = calShouldAbandon(); |
| 495 | } |
| 496 | } |
| 497 | } |
| 498 | if (shouldAbandon) { |
| 499 | groupingSet_->abandonPartialAggregation(); |
| 500 | pool()->release(); |
| 501 | abandonedPartialAggregation_ = true; |
| 502 | LOG(INFO) << __FUNCTION__ << " numInputRows_ = " << numInputRows_ |
| 503 | << ", numOutputRows_ = " << numOutputRows_ |
| 504 | << ", aggregationPct = " << aggregationPct |
| 505 | << ", abandonPartialAggregationMinPct_ = " |
| 506 | << abandonPartialAggregationMinPct_ |
| 507 | << ", abandonPartialAggregationMinFinalPct_ = " |
| 508 | << abandonPartialAggregationMinFinalPct_ |
| 509 | << ", partialAggregationSpillMaxPct_ = " |
| 510 | << partialAggregationSpillMaxPct_ |
| 511 | << ", shouldAbandon = " << shouldAbandon; |
| 512 | return; |
| 513 | } |
| 514 | const int64_t extendedPartialAggregationMemoryUsage = std::min( |
| 515 | maxPartialAggregationMemoryUsage_ * 2, |
| 516 | maxExtendedPartialAggregationMemoryUsage_); |
| 517 | // Calculate the memory to reserve to bump up the aggregation buffer size. If |
| 518 | // the memory reservation below succeeds, it ensures the partial aggregator |
| 519 | // can allocate that much memory in next run. |
| 520 | const int64_t memoryToReserve = std::max<int64_t>( |
| 521 | 0, |
nothing calls this directly
no test coverage detected