cloud-fan commented on code in PR #59764:
URL: https://github.com/apache/spark/pull/59764#discussion_r4373737135
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -4685,16 +4685,36 @@ object SQLConf {
.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 an " +
- "evaluation finds the ratio below this value, the partial aggregation
is bypassed for " +
- "the rest of the input. A larger value bypasses more aggressively.")
+ "so a ratio of 10 means the partial aggregation collapses ten rows
into one. When the " +
+ "periodic evaluation finds the ratio below this value, the partial
aggregation is " +
+ "bypassed for the rest of the input. A larger value bypasses more
aggressively. The " +
+ "evaluation when the map is about to spill uses " +
+
"'spark.sql.execution.aggregate.adaptivePartialAggregation.minCompactionToSpill'
instead " +
+ "when that is larger.")
.version("4.4.0")
.withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
.doubleConf
.checkValue(v => v >= 1.0 && v.isFinite,
"The minimum compaction ratio must be a finite value of at least 1.0.")
.createWithDefault(1.05)
+ val ADAPTIVE_PARTIAL_AGGREGATION_MIN_COMPACTION_TO_SPILL =
Review Comment:
Agreed. With the threshold reverted, this PR is back to the rename and the
timer fix it started as, and the threshold gets its own JIRA. As you say, that
PR will need to show why the two check points have to be tuned separately.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/aggregate/AdaptivePartialAggregationSuite.scala:
##########
@@ -1199,6 +1205,47 @@ class AdaptivePartialAggregationSuite extends QueryTest
with SharedSparkSession
}
}
+ test("the spill check passes through an epoch that compacts too little to
spill") {
+ // Keys `floor(2 * id / 3)` collapse every three rows into two keys, a
steady ratio of about
+ // 1.5. That is above `minCompaction` (1.05), so the periodic checks at 10
and 20 rows keep
+ // aggregating. The forced fallback fills the map when id 30 arrives,
after 30 rows over 20
+ // keys: keeping the aggregation would pay to spill the map and merge it
back for a ratio of
+ // only 1.5, so the default `minCompactionToSpill` (2.0) passes the rest
through instead, and
+ // the sorter is never created. Lowering it to 1.4 keeps the old behavior:
every epoch spills,
+ // and nothing is passed through.
+ Seq(None -> true, Some("1.4") -> false).foreach { case
(minCompactionToSpill, shouldBypass) =>
+ Seq(true, false).foreach { wholeStage =>
+ withSQLConf(
+ (Seq(
+ SQLConf.ADAPTIVE_PARTIAL_AGGREGATION_ENABLED.key -> "true",
+ SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> wholeStage.toString,
+ SQLConf.ENABLE_TWOLEVEL_AGG_MAP.key -> "false",
+ SQLConf.ADAPTIVE_PARTIAL_AGGREGATION_MIN_ROWS.key -> "10",
+ "spark.sql.TungstenAggregate.testFallbackStartsAt" -> "1, 30") ++
+ minCompactionToSpill.map(
+ SQLConf.ADAPTIVE_PARTIAL_AGGREGATION_MIN_COMPACTION_TO_SPILL.key
-> _) ++
+ fixedPlanConfs): _*) {
+ val df = () => spark.range(0, 400, 1, 1)
+ .select(($"id" * 2 / 3).cast("long").cast("string") as "k", $"id"
as "v")
+ .groupBy($"k")
+ .agg(sum($"v") as "s")
+ withClue(s"minCompactionToSpill=$minCompactionToSpill
wholeStage=$wholeStage") {
+ val c = runAndReadCounters(df)
+ if (shouldBypass) {
+ assert(c.skipped == 400 - 30,
+ "the spill check should pass through from the row that found
the map full")
+ assert(c.tasksFallBacked == 0,
+ "the spill check must replace the sort fallback, not trigger
it")
+ } else {
+ assert(c.skipped == 0, "an epoch above minCompactionToSpill must
keep aggregating")
+ assert(c.tasksFallBacked > 0, "an epoch that keeps aggregating
must spill")
+ }
+ }
+ }
+ }
+ }
+ }
+
test("a new in-memory map epoch after a spill can still bypass") {
Review Comment:
This is about the threshold change, which is reverted here, so I'll address
it in the new PR.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/aggregate/AdaptivePartialAggregationSuite.scala:
##########
@@ -1199,6 +1205,47 @@ class AdaptivePartialAggregationSuite extends QueryTest
with SharedSparkSession
}
}
+ test("the spill check passes through an epoch that compacts too little to
spill") {
+ // Keys `floor(2 * id / 3)` collapse every three rows into two keys, a
steady ratio of about
+ // 1.5. That is above `minCompaction` (1.05), so the periodic checks at 10
and 20 rows keep
+ // aggregating. The forced fallback fills the map when id 30 arrives,
after 30 rows over 20
+ // keys: keeping the aggregation would pay to spill the map and merge it
back for a ratio of
+ // only 1.5, so the default `minCompactionToSpill` (2.0) passes the rest
through instead, and
+ // the sorter is never created. Lowering it to 1.4 keeps the old behavior:
every epoch spills,
+ // and nothing is passed through.
+ Seq(None -> true, Some("1.4") -> false).foreach { case
(minCompactionToSpill, shouldBypass) =>
+ Seq(true, false).foreach { wholeStage =>
+ withSQLConf(
+ (Seq(
+ SQLConf.ADAPTIVE_PARTIAL_AGGREGATION_ENABLED.key -> "true",
+ SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> wholeStage.toString,
+ SQLConf.ENABLE_TWOLEVEL_AGG_MAP.key -> "false",
+ SQLConf.ADAPTIVE_PARTIAL_AGGREGATION_MIN_ROWS.key -> "10",
+ "spark.sql.TungstenAggregate.testFallbackStartsAt" -> "1, 30") ++
+ minCompactionToSpill.map(
+ SQLConf.ADAPTIVE_PARTIAL_AGGREGATION_MIN_COMPACTION_TO_SPILL.key
-> _) ++
+ fixedPlanConfs): _*) {
+ val df = () => spark.range(0, 400, 1, 1)
Review Comment:
This is about the threshold change, which is reverted here, so I'll address
it in the new PR.
##########
docs/sql-performance-tuning.md:
##########
@@ -251,6 +253,18 @@ start later.
</td>
<td>4.4.0</td>
</tr>
+ <tr>
+
<td><code>spark.sql.execution.aggregate.adaptivePartialAggregation.minCompactionToSpill</code></td>
+ <td>2.0</td>
Review Comment:
I'll do this in the new PR, with the cases you list.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -4685,16 +4685,36 @@ object SQLConf {
.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 an " +
- "evaluation finds the ratio below this value, the partial aggregation
is bypassed for " +
- "the rest of the input. A larger value bypasses more aggressively.")
+ "so a ratio of 10 means the partial aggregation collapses ten rows
into one. When the " +
+ "periodic evaluation finds the ratio below this value, the partial
aggregation is " +
+ "bypassed for the rest of the input. A larger value bypasses more
aggressively. The " +
+ "evaluation when the map is about to spill uses " +
+
"'spark.sql.execution.aggregate.adaptivePartialAggregation.minCompactionToSpill'
instead " +
+ "when that is larger.")
.version("4.4.0")
.withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
.doubleConf
.checkValue(v => v >= 1.0 && v.isFinite,
"The minimum compaction ratio must be a finite value of at least 1.0.")
.createWithDefault(1.05)
+ val ADAPTIVE_PARTIAL_AGGREGATION_MIN_COMPACTION_TO_SPILL =
+
buildConf("spark.sql.execution.aggregate.adaptivePartialAggregation.minCompactionToSpill")
+ .doc("The minimum compaction ratio required to keep the pre-shuffle
partial aggregation " +
+ "when its map is about to spill (see " +
+ s"'${ADAPTIVE_PARTIAL_AGGREGATION_ENABLED.key}'). Keeping the
aggregation past that " +
+ "point also pays to write the map to disk and merge it back later, so
it must collapse " +
+ "more rows into each key to be worthwhile than it must between spills.
When the ratio " +
+ "is below this value, the spill is skipped and the rest of the input
is passed " +
Review Comment:
The threshold's rationale is reverted here. The new PR rewrites these
sentences, including the "the spill is skipped entirely" wording that predates
it, and will qualify them for a task that has already spilled.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -4685,16 +4685,36 @@ object SQLConf {
.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 an " +
- "evaluation finds the ratio below this value, the partial aggregation
is bypassed for " +
- "the rest of the input. A larger value bypasses more aggressively.")
+ "so a ratio of 10 means the partial aggregation collapses ten rows
into one. When the " +
+ "periodic evaluation finds the ratio below this value, the partial
aggregation is " +
Review Comment:
This is about the threshold change, which is reverted here, so I'll address
it in the new PR.
##########
docs/sql-performance-tuning.md:
##########
@@ -200,6 +200,8 @@ queue behind it and flushed only after it drains, so a
duplicate of a key alread
merges after that key's accumulated rows, keeping an order-sensitive aggregate
such as
`first`/`last` consistent with a run that never bypasses. The ratio is
evaluated periodically, and
again right before the aggregation map would spill, in which case the spill is
skipped entirely.
+Keeping the aggregation past that point also pays to write the map to disk and
merge it back later,
+so the evaluation before a spill demands a higher ratio,
`minCompactionToSpill`.
Review Comment:
This is about the threshold change, which is reverted here, so I'll address
it in the new PR.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/aggregate/AdaptivePartialAggregationSuite.scala:
##########
@@ -1199,6 +1205,47 @@ class AdaptivePartialAggregationSuite extends QueryTest
with SharedSparkSession
}
}
+ test("the spill check passes through an epoch that compacts too little to
spill") {
+ // Keys `floor(2 * id / 3)` collapse every three rows into two keys, a
steady ratio of about
+ // 1.5. That is above `minCompaction` (1.05), so the periodic checks at 10
and 20 rows keep
Review Comment:
This is about the threshold change, which is reverted here, so I'll address
it in the new PR.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashAggregateExec.scala:
##########
@@ -772,15 +778,21 @@ case class HashAggregateExec(
// the maps must precede the held rows so the downstream Final merges a
group's map buffer
// before its pass-through buffers (a group can straddle the freeze, since
the maps only freeze
// once pass-through fires), and the copies were made when the row was
queued, so emitting them
- // here cannot corrupt the build's reusable rows.
+ // here cannot corrupt the build's reusable rows. The iterator setup sorts
and merges any
+ // spilled data, which the non-adaptive path times as part of the build,
so it is timed here
+ // too; it runs once per task.
+ val aggTime = metricTerm(ctx, "aggTime")
Review Comment:
Fixed. The line is back in its old place, so plans without the feature
generate the same code as before. The map setup registers its own reference to
the metric, but only on the adaptive path.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/aggregate/AdaptivePartialAggregationSuite.scala:
##########
@@ -102,7 +102,8 @@ class AdaptivePartialAggregationSuite extends QueryTest
with SharedSparkSession
*/
private def checkAdaptiveMatchesReference(
build: Int => DataFrame,
- expectBypass: Boolean = true): Unit = {
+ expectBypass: Boolean = true,
+ extraConfs: Seq[(String, String)] = Nil): Unit = {
Review Comment:
This is about the threshold change, which is reverted here, so I'll address
it in the new PR.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/TungstenAggregationIterator.scala:
##########
@@ -230,8 +230,8 @@ class TungstenAggregationIterator(
val spillCheckMinRows = math.max(minRows, 1L)
// The partial aggregation is ineffective when it does not collapse
`minCompaction` rows into
// one key. There is no fast map on this path, so the map's keys are all
the operator holds.
- def ineffective(): Boolean =
- processedRows < hashMap.getNumKeys().toDouble * adaptiveMinCompaction
+ def ineffective(minCompaction: Double): Boolean =
Review Comment:
This is about the threshold change, which is reverted here, so I'll address
it in the new PR.
--
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]
