ulysses-you commented on code in PR #57742:
URL: https://github.com/apache/spark/pull/57742#discussion_r3727560039
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashAggregateExec.scala:
##########
@@ -663,46 +851,123 @@ case class HashAggregateExec(
case _ => ("true", "", "")
}
- val findOrInsertRegularHashMap: String =
+ // The compaction ratio is measured at the operator level: all processed
rows against the keys
+ // held by both maps, so two-level-map routing does not change the
decision. The same predicate
+ // decides both check points -- periodically every `minRows` rows, and
right before the map
+ // would spill (in which case the spill is skipped entirely). `minRows =
0` disables the
+ // periodic check: the row count is only ever compared after being
incremented past 0, so it
+ // never matches and only the spill check remains.
+ val adaptiveIneffective = if (adaptivePartialAggEnabled) {
+ val totalKeys = if (isFastHashMapEnabled) {
+ s"($fastHashMapTerm.getNumKeys() + $hashMapTerm.getNumKeys())"
+ } else {
+ s"$hashMapTerm.getNumKeys()"
+ }
+ s"$processedRowsTerm < (double) $totalKeys * ${adaptiveMinCompaction}D"
+ } else {
+ ""
+ }
+
+ // After a spill the map starts a new in-memory epoch, so the counters
restart and the ratio of
+ // that epoch alone decides whether the remaining rows are passed through.
+ val adaptiveResetEpoch = if (adaptivePartialAggEnabled) {
s"""
- |// generate grouping key
- |${unsafeRowKeyCode.code}
- |int $unsafeRowKeyHash = ${unsafeRowKeyCode.value}.hashCode();
- |if ($checkFallbackForBytesToBytesMap) {
- | // try to get the buffer from hash map
- | $unsafeRowBuffer =
- | $hashMapTerm.getAggregationBufferFromUnsafeRow($unsafeRowKeys,
$unsafeRowKeyHash);
- |}
- |// Can't allocate buffer from the hash map. Spill the map and
fallback to sort-based
- |// aggregation after processing all input rows.
- |if ($unsafeRowBuffer == null) {
- | if ($sorterTerm == null) {
- | $sorterTerm = $hashMapTerm.destructAndCreateExternalSorter();
- | } else {
- |
$sorterTerm.merge($hashMapTerm.destructAndCreateExternalSorter());
- | }
- | $resetCounter
- | // the hash map had be spilled, it should have enough memory now,
- | // try to allocate buffer again.
- | $unsafeRowBuffer = $hashMapTerm.getAggregationBufferFromUnsafeRow(
- | $unsafeRowKeys, $unsafeRowKeyHash);
- | if ($unsafeRowBuffer == null) {
- | // failed to allocate the first page
- | throw QueryExecutionErrors.aggregateOutOfMemoryError();
- | }
- |}
+ |$processedRowsTerm = 0L;
Review Comment:
Addressed. You are right that the two sides of the ratio were on different
definitions: the processed-row
count restarts after a spill and the regular map's keys go with it, but the
fast map neither
spills nor clears, so its keys carried into the next epoch's denominator.
The fast-map key count
is now snapshotted at each spill and subtracted, putting both sides on the
same epoch. Once the
fast map fills it stops accepting keys, so the difference settles at zero
and the denominator is
the regular map's alone.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -4156,6 +4156,48 @@ object SQLConf {
.booleanConf
.createWithDefault(false)
+ val ADAPTIVE_PARTIAL_AGGREGATION_ENABLED =
+
buildConf("spark.sql.execution.aggregate.adaptivePartialAggregation.enabled")
+ .doc("When true, hash aggregation adaptively bypasses the pre-shuffle
partial aggregation " +
+ "at runtime when it observes that the partial aggregation is not
reducing the number of " +
+ "rows enough to be worthwhile. Once bypassed, the remaining input rows
are passed " +
+ "through as single-row partial aggregation buffers for the final
aggregation to merge, " +
+ "which avoids the cost of maintaining and spilling a large aggregation
map with little " +
+ "reduction benefit. This applies only to hash aggregation with
grouping keys.")
+ .version("4.4.0")
+ .withBindingPolicy(ConfigBindingPolicy.SESSION)
+ .booleanConf
+ .createWithDefault(true)
+
+ val ADAPTIVE_PARTIAL_AGGREGATION_MIN_ROWS =
+
buildConf("spark.sql.execution.aggregate.adaptivePartialAggregation.minRows")
+ .doc("The number of rows to process before adaptive partial aggregation
(see " +
+ s"'${ADAPTIVE_PARTIAL_AGGREGATION_ENABLED.key}') evaluates the
compaction ratio. The " +
+ "ratio is evaluated once this many rows have been processed since the
previous " +
+ "evaluation, so a decision is never made on too few rows. A value of 0
disables the " +
+ "periodic evaluation entirely, leaving only the check made when the
aggregation map is " +
+ "about to spill.")
+ .version("4.4.0")
+ .withBindingPolicy(ConfigBindingPolicy.SESSION)
+ .longConf
+ .checkValue(_ >= 0, "The minimum row count must not be negative.")
+ .createWithDefault(100000)
+
+ val ADAPTIVE_PARTIAL_AGGREGATION_MIN_COMPACTION =
+
buildConf("spark.sql.execution.aggregate.adaptivePartialAggregation.minCompaction")
+ .doc("The minimum compaction ratio required to keep the pre-shuffle
partial aggregation " +
+ s"(see '${ADAPTIVE_PARTIAL_AGGREGATION_ENABLED.key}'). The compaction
ratio is the " +
+ "number of processed rows divided by the number of keys held in the
aggregation maps, " +
+ "so a ratio of 10 means the partial aggregation collapses ten rows
into one. When the " +
+ s"ratio is below this value after
'${ADAPTIVE_PARTIAL_AGGREGATION_MIN_ROWS.key}' rows, " +
+ "or when the aggregation map is about to spill, the partial
aggregation is bypassed for " +
+ "the rest of the input. A larger value bypasses more aggressively.")
+ .version("4.4.0")
+ .withBindingPolicy(ConfigBindingPolicy.SESSION)
+ .doubleConf
+ .checkValue(_ >= 1.0, "The minimum compaction ratio must be at least
1.0.")
Review Comment:
Addressed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
