ulysses-you commented on code in PR #57742:
URL: https://github.com/apache/spark/pull/57742#discussion_r3755858718


##########
sql/core/src/test/scala/org/apache/spark/sql/execution/aggregate/AdaptivePartialAggregationSuite.scala:
##########
@@ -0,0 +1,1136 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.aggregate
+
+import org.apache.spark.sql.{DataFrame, QueryTest, Row}
+import org.apache.spark.sql.catalyst.expressions.aggregate.Partial
+import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
+import org.apache.spark.sql.functions._
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+
+/**
+ * Tests for runtime adaptive partial aggregation
+ * (see [[SQLConf.ADAPTIVE_PARTIAL_AGGREGATION_ENABLED]]). When a partial 
aggregate is not reducing
+ * rows, the operator stops aggregating and streams the remaining rows through 
as single-row partial
+ * buffers for the Final aggregate to merge. It must never change results.
+ *
+ * The suite has two halves:
+ *   1. Correctness: output is identical to the reference (feature-off) run 
across the full matrix
+ *      of codegen on/off, two-level map on/off, and spill/no-spill, over a 
range of aggregate
+ *      shapes, key types, and `Expand`-bearing plans (ROLLUP / CUBE / 
GROUPING SETS /
+ *      multi-distinct).
+ *   2. Triggering: the `numBypassingRows` metric proves the bypass actually 
fires when (and only
+ *      when) it should -- high-cardinality input bypasses, low-cardinality 
input keeps aggregating,
+ *      the feature switch and eligibility rules are honored, and both check 
points work.
+ */
+class AdaptivePartialAggregationSuite extends QueryTest with SharedSparkSession
+  with AdaptiveSparkPlanHelper {
+
+  import testImplicits._
+
+  // A `testFallbackStartsAt` setting ("fastMapCounter, regularMapCounter") 
that makes the regular
+  // map fall back (spill) periodically, exercising the spill-check decision 
path in both the
+  // codegen and interpreted aggregation paths. Kept moderate so 
low-cardinality inputs (which are
+  // never bypassed and therefore really spill) do not open an unbounded 
number of spill readers.
+  private val forceSpillFallback = "4, 16"
+
+  // The upstream `CombineAdjacentAggregation` and `ReplaceHashWithSortAgg` 
rules would change the
+  // plan of these small single-partition queries away from a Partial+Final 
`HashAggregateExec`:
+  // the former merges the two adjacent phases (no shuffle in between) into a 
single `Complete`
+  // aggregate, and the latter converts a hash aggregate to a sort aggregate 
when the input is
+  // already sorted by the grouping key (a `Range` over an ascending `id` 
key). The adaptive
+  // feature lives in the partial hash aggregation, so both rules are disabled 
to keep that
+  // structure in the tests.
+  private val fixedPlanConfs = Seq(
+    SQLConf.COMBINE_ADJACENT_AGGREGATION_ENABLED.key -> "false",
+    SQLConf.REPLACE_HASH_WITH_SORT_AGG_ENABLED.key -> "false")
+
+  /**
+   * Runs `build` with adaptive partial aggregation disabled (the reference) 
and then across the
+   * full configuration matrix with it enabled, asserting every enabled run 
matches the reference.
+   *
+   * `build` takes the number of input partitions, which the matrix varies 
along with everything
+   * else, because the plan shape decides which parts of the feature run at 
all. When the two
+   * aggregates end up in one whole-stage -- no `Exchange` between them -- the 
partial aggregate's
+   * output feeds the Final's `doConsume` directly and never reaches
+   * `BufferedRowIterator.currentRows`, so `shouldStop()` stays false for the 
whole build and
+   * neither `needStopCheck` nor the resumed-build path is exercised. 
Splitting them puts the
+   * streamed rows through the output buffer and runs both.
+   *
+   * More than one input partition is necessary but not sufficient for that 
split: a `Range` keyed
+   * directly on `id` already reports an output partitioning that satisfies 
the Final aggregate's
+   * `ClusteredDistribution`, so `EnsureRequirements` inserts no `Exchange` 
however many partitions
+   * it has. Tests that want the split shape group on a derived key (a cast, 
say) so the input
+   * partitioning no longer satisfies the requirement.
+   *
+   * `expectBypass` ties the correctness guarantee to the triggering 
guarantee: beyond matching the
+   * reference, every cell must either actually stream rows through (when 
true) or keep
+   * aggregating (when false). Without it a test could silently stop 
exercising pass-through if the
+   * input stopped being bypassable, and only this assertion makes that fail 
loudly.
+   */
+  private def checkAdaptiveMatchesReference(
+      build: Int => DataFrame,
+      expectBypass: Boolean = true): Unit = {
+    for {
+      inputPartitions <- Seq(1, 2)
+      wholeStage <- Seq(true, false)
+      twoLevelMap <- Seq(true, false)
+      forceSpill <- Seq(true, false)
+    } {
+      // The reference is built with the same partitioning, so only the 
feature differs.
+      val reference = withSQLConf(
+        (SQLConf.ADAPTIVE_PARTIAL_AGGREGATION_ENABLED.key -> "false") +: 
fixedPlanConfs: _*) {
+        build(inputPartitions).collect().toSeq
+      }
+      val spillConf = if (forceSpill) {
+        Seq("spark.sql.TungstenAggregate.testFallbackStartsAt" -> 
forceSpillFallback)
+      } else {
+        Nil
+      }
+      withSQLConf(
+        (Seq(
+          SQLConf.ADAPTIVE_PARTIAL_AGGREGATION_ENABLED.key -> "true",
+          SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> wholeStage.toString,
+          SQLConf.ENABLE_TWOLEVEL_AGG_MAP.key -> twoLevelMap.toString,
+          // Small `minRows` so the periodic check runs on modest inputs.
+          SQLConf.ADAPTIVE_PARTIAL_AGGREGATION_MIN_ROWS.key -> "8") ++
+          spillConf ++ fixedPlanConfs): _*) {
+        val msg = s"inputPartitions=$inputPartitions wholeStage=$wholeStage " +
+          s"twoLevelMap=$twoLevelMap forceSpill=$forceSpill"
+        withClue(msg) {
+          // Collect once so the metrics are populated, then check whether the 
bypass fired for
+          // this cell. The metric lives on the `Partial`-mode 
`HashAggregateExec`, so that is the
+          // operator the assertion reads.
+          val df = build(inputPartitions)
+          df.collect()
+          val skipped = collect(df.queryExecution.executedPlan) {
+            case agg: HashAggregateExec if 
agg.aggregateExpressions.forall(_.mode == Partial) =>
+              agg.metrics.get("numBypassingRows").map(_.value).getOrElse(0L)
+          }.sum
+          if (expectBypass) {
+            assert(skipped > 0,
+              s"expected rows to bypass partial aggregation, got $skipped 
bypassed rows")
+          } else {
+            assert(skipped == 0,
+              s"expected no rows to bypass partial aggregation, got $skipped 
bypassed rows")
+          }
+          checkAnswer(df, reference)
+        }
+      }
+    }
+  }
+
+  /**
+   * The observable per-run counters we assert on, all read from the partial 
`HashAggregateExec` in
+   * a single execution so the metrics are not double-counted:
+   *   - `skipped`: our self-reported `numBypassingRows` metric.
+   *   - `partialOutputRows`: the partial aggregate's own `numOutputRows`. An 
independent,
+   *     pre-existing counter driven by the normal output path, so it is the 
ground truth for
+   *     whether rows were streamed through -- it equals the distinct key 
count when aggregation is
+   *     effective and climbs toward the input row count once the operator 
bypasses.
+   *   - `spillBytes`: the partial aggregate's `spillSize`. Reliable only when 
no fallback is
+   *     forced: on the interpreted path this is derived from the 
task-cumulative memory-spill
+   *     counter, so a forced fallback (or downstream shuffle-write spill) can 
inflate it.
+   *     Asserted only by the periodic check test, which forces no fallback; 
use
+   *     `tasksFallBacked` otherwise.
+   *   - `tasksFallBacked`: the partial aggregate's `numTasksFallBacked`, 
incremented only when the
+   *     regular map actually falls back into sort-based aggregation. When the 
spill check bypasses
+   *     at the spill boundary the sorter is never created, so this stays 0 -- 
direct, per-operator
+   *     evidence the bypass replaced the sort fallback.
+   */
+  private case class AggCounters(
+      skipped: Long,
+      partialOutputRows: Long,
+      spillBytes: Long,
+      tasksFallBacked: Long)
+
+  // Verifies `df` (an already-collected bypassing run) produces the same 
results as the feature-off
+  // reference. `build` is re-run for the reference so it gets a genuinely 
non-adaptive plan rather
+  // than reusing the bypassing run's cached one.
+  private def checkAgainstReference(df: DataFrame, build: () => DataFrame): 
Unit = {
+    val reference = withSQLConf(
+      SQLConf.ADAPTIVE_PARTIAL_AGGREGATION_ENABLED.key -> "false") {
+      build().collect().toSeq
+    }
+    checkAnswer(df, reference)
+  }
+
+  private def runAndReadCounters(build: () => DataFrame): AggCounters = {
+    // The triggering tests assert on metrics, so also verify the bypassing 
run produces the same
+    // results as the feature-off reference.
+    val df = build()
+    df.collect()
+    val partialAggs = collect(df.queryExecution.executedPlan) {
+      case agg: HashAggregateExec if agg.aggregateExpressions.forall(_.mode == 
Partial) => agg
+    }
+    // A partial aggregate is always present for the grouped queries these 
tests use.
+    assert(partialAggs.nonEmpty, "expected a partial HashAggregateExec in the 
plan")
+    val counters = AggCounters(
+      // The metric is only registered on aggregates the feature applies to; 
an aggregate without
+      // it bypassed nothing.
+      skipped = 
partialAggs.map(_.metrics.get("numBypassingRows").map(_.value).getOrElse(0L)).sum,
+      partialOutputRows = 
partialAggs.map(_.metrics("numOutputRows").value).sum,
+      spillBytes = partialAggs.map(_.metrics("spillSize").value).sum,
+      tasksFallBacked = 
partialAggs.map(_.metrics("numTasksFallBacked").value).sum)
+    checkAgainstReference(df, build)
+    counters
+  }
+
+  private def numBypassingRows(build: () => DataFrame): Long = 
runAndReadCounters(build).skipped
+
+  // Returns the bypassed-row count per Partial-mode `HashAggregateExec`, 
keyed by the number of
+  // grouping keys, and verifies the run matches the feature-off reference. A 
`count(DISTINCT ...)`
+  // group-by has two such Partial phases -- the de-duplication partial 
(grouping on key + distinct
+  // columns) and the distinct partial (grouping on the keys only) -- so their 
bypasses can be told
+  // apart by the grouping key count.
+  private def bypassRowsByGroupingKeyCount(build: () => DataFrame): Map[Int, 
Long] = {
+    val df = build()
+    df.collect()
+    val byKeyCount = collect(df.queryExecution.executedPlan) {
+      case agg: HashAggregateExec if agg.aggregateExpressions.forall(_.mode == 
Partial) =>
+        agg.groupingExpressions.length ->
+          agg.metrics.get("numBypassingRows").map(_.value).getOrElse(0L)
+    }.groupBy(_._1).map { case (n, pairs) => n -> pairs.map(_._2).sum }
+    checkAgainstReference(df, build)
+    byKeyCount
+  }
+
+  /**
+   * Runs `body` once per (wholeStage, twoLevelMap) combination with the 
feature enabled and a small
+   * `minRows`, threading a descriptive clue for failure messages.
+   *
+   * The fast (first-level) map is append-only and never spills, so only the 
regular (second-level)
+   * map can reach a spill boundary. With the default fast-map capacity (2^16) 
a small
+   * high-cardinality input would be fully absorbed by the fast map and never 
reach the regular map,
+   * so nothing could ever bypass. To make the triggering tests meaningful 
when the two-level map is
+   * on, we shrink the fast map via the first field of `testFallbackStartsAt` 
so rows fall through
+   * to the regular map. `regularFallback` optionally sets the second field to 
also force the
+   * regular map to spill (for the spill check); when 0 the regular map does 
not spill.
+   */
+  private def forEachCodegenAndMap(
+      minRows: Long = 8,
+      regularFallback: Int = 0,
+      minCompaction: Double = -1.0)(
+      body: String => Unit): Unit = {
+    for {
+      wholeStage <- Seq(true, false)
+      twoLevelMap <- Seq(true, false)
+    } {
+      // Shrink the fast map to 4 keys when it is on so rows reach the regular 
map. The second field
+      // controls regular-map spilling; 0 means "never" (a large sentinel).
+      val fallbackConf = if (twoLevelMap || regularFallback > 0) {
+        val fastCap = if (twoLevelMap) 4 else 1
+        val regular = if (regularFallback > 0) regularFallback else 
Int.MaxValue
+        Seq("spark.sql.TungstenAggregate.testFallbackStartsAt" -> s"$fastCap, 
$regular")
+      } else {
+        Nil
+      }
+      // A negative value means "leave the threshold at its default".
+      val thresholdConf = if (minCompaction >= 0.0) {
+        Seq(SQLConf.ADAPTIVE_PARTIAL_AGGREGATION_MIN_COMPACTION.key -> 
minCompaction.toString)
+      } else {
+        Nil
+      }
+      withSQLConf(
+        (Seq(
+          SQLConf.ADAPTIVE_PARTIAL_AGGREGATION_ENABLED.key -> "true",
+          SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> wholeStage.toString,
+          SQLConf.ENABLE_TWOLEVEL_AGG_MAP.key -> twoLevelMap.toString,
+          SQLConf.ADAPTIVE_PARTIAL_AGGREGATION_MIN_ROWS.key -> 
minRows.toString) ++
+          fallbackConf ++ thresholdConf ++ fixedPlanConfs): _*) {
+        body(s"wholeStage=$wholeStage twoLevelMap=$twoLevelMap")
+      }
+    }
+  }
+
+  /////////////////////////////////////////////////////////////////////////////
+  // Part 1: Correctness -- results identical to the feature-off reference.
+  /////////////////////////////////////////////////////////////////////////////
+
+  test("results unchanged for high-cardinality input that bypasses partial 
aggregation") {
+    // Every grouping key is distinct, so partial aggregation reduces nothing 
and should be
+    // bypassed by the periodic check.
+    checkAdaptiveMatchesReference { parts =>
+      spark.range(0, 200, 1, parts)
+        .select($"id".cast("string") as "k", ($"id" * 2) as "v")
+        .groupBy($"k")
+        .agg(sum($"v") as "s", count(lit(1)) as "c", max($"v") as "m")
+    }
+  }
+
+  test("results unchanged for low-cardinality input that keeps partial 
aggregation") {
+    // Few distinct keys, high reduction: partial aggregation is effective and 
should be kept, so
+    // the bypass metric must stay zero in every cell.
+    checkAdaptiveMatchesReference(
+      expectBypass = false,
+      build = { parts =>
+        spark.range(0, 600, 1, parts)
+          .select(($"id" % 5).cast("string") as "k", $"id" as "v")
+          .groupBy($"k")
+          .agg(sum($"v") as "s", count(lit(1)) as "c", min($"v") as "mn", 
max($"v") as "mx")
+      })
+  }
+
+  test("results unchanged for medium-cardinality input near the reduction 
threshold") {
+    // Roughly half the rows are distinct keys; exercises the boundary of the 
ratio checks. The
+    // overall compaction ratio (~2.0) is above the threshold, but the *first* 
periodic check still
+    // sees the leading distinct keys and fires, so the bypass must be 
observable too.
+    checkAdaptiveMatchesReference { parts =>
+      spark.range(0, 1000, 1, parts)
+        .select(($"id" % 500).cast("string") as "k", $"id" as "v")
+        .groupBy($"k")
+        .agg(sum($"v") as "s", count(lit(1)) as "c")
+    }
+  }
+
+  test("results unchanged with multiple grouping keys and string keys") {
+    checkAdaptiveMatchesReference { parts =>
+      spark.range(0, 500, 1, parts)
+        .select(
+          concat(lit("g"), ($"id" % 300).cast("string")) as "k1",
+          ($"id" % 7) as "k2",
+          $"id" as "v")
+        .groupBy($"k1", $"k2")
+        .agg(sum($"v") as "s", count(lit(1)) as "c")
+    }
+  }
+
+  test("results unchanged with nullable grouping keys") {
+    // Nulls are sparse enough (1 in 40) that the keys stay close to unique 
and the input really
+    // does bypass; a denser null key would lift the compaction ratio above 
the threshold and the
+    // test would never engage the feature.
+    checkAdaptiveMatchesReference { parts =>
+      spark.range(0, 400, 1, parts)
+        .select(
+          when($"id" % 40 === 0, lit(null)).otherwise($"id").cast("string") as 
"k",
+          $"id" as "v")
+        .groupBy($"k")
+        .agg(sum($"v") as "s", count(lit(1)) as "c")
+    }
+  }
+
+  test("results unchanged with average (multi-slot buffer) aggregate") {
+    // avg has a two-slot partial buffer (sum, count); pass-through buffers 
must carry all slots.
+    checkAdaptiveMatchesReference { parts =>
+      spark.range(0, 300, 1, parts)
+        .select($"id".cast("string") as "k", ($"id" + 1) as "v")
+        .groupBy($"k")
+        .agg(avg($"v") as "a", sum($"v") as "s")
+    }
+  }
+
+  test("results unchanged with a mix of many aggregate functions and buffer 
types") {
+    // Exercises a wide pass-through buffer spanning several aggregate buffer 
layouts at once:
+    // sum (decimal), avg (double), count, min/max, first/last, and stddev 
(imperative buffer).

Review Comment:
   Fixed. `approx_count_distinct` (`HyperLogLogPlusPlus`, an 
`ImperativeAggregate` with a mutable `LongType` buffer, `supportCodegen = 
false`) is now its own test, `results unchanged with an imperative-buffer 
aggregate`, so the `copyFrom(initialAggregationBuffer)` reset of state written 
by `initialize(buffer)` is exercised on the interpreted path, and it stays out 
of the codegen matrix so the mixed test keeps its codegen cells. The mix test's 
comment now says stddev is declarative.



##########
sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashAggregateExec.scala:
##########
@@ -845,29 +1158,56 @@ case class HashAggregateExec(
       }
     }
 
-    val declareRowBuffer: String = if (isFastHashMapEnabled) {
-      val fastRowType = if (isVectorizedHashMapEnabled) {
-        classOf[MutableColumnarRow].getName
+    val declareRowBuffer: String = {
+      val declareBuffers = if (isFastHashMapEnabled) {
+        val fastRowType = if (isVectorizedHashMapEnabled) {
+          classOf[MutableColumnarRow].getName
+        } else {
+          "UnsafeRow"
+        }
+        s"""
+           |UnsafeRow $unsafeRowBuffer = null;
+           |$fastRowType $fastRowBuffer = null;
+         """.stripMargin
+      } else {
+        s"UnsafeRow $unsafeRowBuffer = null;"
+      }
+      val declareBypassed = if (adaptivePartialAggEnabled) {
+        s"boolean $adaptiveRowBypassedTerm = false;"
       } else {
-        "UnsafeRow"
+        ""
       }
       s"""
-         |UnsafeRow $unsafeRowBuffer = null;
-         |$fastRowType $fastRowBuffer = null;
+         |$declareBuffers
+         |$declareBypassed
        """.stripMargin
-    } else {
-      s"UnsafeRow $unsafeRowBuffer = null;"
     }
 
     // We try to do hash map based in-memory aggregation first. If there is 
not enough memory (the
     // hash map will return null for new key), we spill the hash map to disk 
to free memory, then
     // continue to do in-memory aggregation and spilling until all the rows 
had been processed.
     // Finally, sort the spilled aggregate buffers by key, and merge them 
together for same key.
+    //
+    // With adaptive partial aggregation, once pass-through is active 
`updateRowInHashMap` fills the
+    // single-row buffer built above; we then emit `key ++ buffer` straight to 
the parent so the row
+    // skips both the fast map and the regular map.
+    val emitPassThroughRow = if (adaptivePartialAggEnabled) {
+      val numBypassingRows = metricTerm(ctx, "numBypassingRows")
+      s"""
+         |if ($adaptiveRowBypassedTerm) {
+         |  $numBypassingRows.add(1);
+         |  $outputFunc(${unsafeRowKeyCode.value}, $unsafeRowBuffer);

Review Comment:
   Fixed at this head, with your repro in as `fan-out below a split aggregate 
preserves the merge order`. A one-to-many child (`GenerateExec`/`ExpandExec`) 
has no `shouldStop()` in its fan-out loop, so with the drain deferred to the 
next re-entry the whole trigger batch was appended before the frozen maps 
drained: codegen gave `(400, 0)` against `(0, 400)` interpreted and `(0, 400)` 
feature-off. The drain now runs right after the trigger row whenever the child 
reports `needCopyResult`, which fan-out children always do; their output is a 
copy anyway, so the aliasing hazard that forced the deferred drain in the 
whole-stage root is gone, and the two paths and the reference all give `(0, 
400)`. We kept trigger-first rather than draining before the first bypassed 
row, because map-first would need a per-row copy on the generated path or a 
park-the-trigger re-entry state machine, which cannot resume a fan-out loop 
that keeps no position; the residual case, the first bypassed row colliding w
 ith a key in the map, is documented in the order test.



##########
sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashAggregateExec.scala:
##########
@@ -711,6 +995,54 @@ case class HashAggregateExec(
       } else {
         findOrInsertRegularHashMap
       }
+
+      // Every row is either accepted by an aggregation map or streamed 
through -- the fast map
+      // serves a row without it ever reaching the regular map, so both 
buffers are consulted to
+      // tell the two apart.
+      //
+      // An accepted row counts toward the compaction ratio, so the numerator 
matches the
+      // operator-level denominator. Counting inside the regular-map branch 
alone would drop the
+      // rows the fast map absorbed from the ratio and bypass an aggregation 
that is in fact
+      // reducing. A row no map holds is streamed once pass-through is active: 
both probes are
+      // skipped (guarded above), so neither buffer is set, and `rowBypassed` 
marks exactly those
+      // rows. The row that fails to insert at the spill boundary lands here 
too, while the row
+      // that merely flipped pass-through at the check point is already 
aggregated in the map that
+      // took it and must not be re-emitted.
+      val countOrPassThroughRow = if (adaptivePartialAggEnabled) {
+        val heldByAMap = if (isFastHashMapEnabled) {
+          s"($fastRowBuffer != null || $unsafeRowBuffer != null)"
+        } else {
+          s"($unsafeRowBuffer != null)"
+        }
+        // The grouping key was already projected in 
`findOrInsertRegularHashMap`
+        // (`unsafeRowKeyCode.code`), so `unsafeRowKeyCode.value` holds this 
row's key. Only build
+        // the single-row partial buffer here.
+        s"""
+           |if ($heldByAMap) {
+           |  if (!$adaptivePassThroughTerm) {
+           |    $processedRowsTerm += 1;
+           |    if ($processedRowsTerm == $adaptiveNextCheckRowTerm) {

Review Comment:
   Thanks, and agreed it is not a blocker. The docs now state this directly: 
"once triggered, pass-through is not reversed for the rest of the task", and "a 
distinct-heavy prefix can trip the pass-through early and keep it tripped even 
where the rest of the input would aggregate well", alongside the 
`minCompaction`/`minRows` tuning notes. On the opt-out, `enabled` (default 
`false`) is the documented switch in the config table; I trimmed the explicit 
sentence to keep the paragraph tight, say the word if you want it back.



##########
sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashAggregateExec.scala:
##########
@@ -141,6 +148,40 @@ case class HashAggregateExec(
     .map(_.asInstanceOf[DeclarativeAggregate])
   private val bufferSchema = 
DataTypeUtils.fromAttributes(aggregateBufferAttributes)
 
+  /**
+   * Whether adaptive partial aggregation applies to this operator. When it 
does, the aggregation
+   * may bypass partial aggregation at runtime and pass the remaining input 
rows through as
+   * single-row partial buffers (see 
[[SQLConf.ADAPTIVE_PARTIAL_AGGREGATION_ENABLED]]). It only
+   * applies to a pre-shuffle partial aggregation with grouping keys:
+   *   - `Partial` mode only: the downstream `Final` aggregation merges the 
passed-through
+   *     single-row buffers, so the output contract is unchanged. 
`Final`/`Complete` produce the
+   *     result themselves and have no such downstream. `PartialMerge` does 
have one and could be
+   *     supported by passing its incoming buffer through unchanged, but that 
is left for later.
+   *   - grouping keys present: a global aggregation produces a single output 
row, so partial
+   *     aggregation achieves the maximum reduction and must never be bypassed.
+   *   - DISTINCT aggregate functions are allowed: the intermediate 
`PartialMerge` phase of the
+   *     multi-phase distinct plan is not `Partial` mode (and requires a 
distribution), so it always
+   *     aggregates and de-duplicates, and the passed-through rows from the 
distinct `Partial` phase
+   *     therefore carry exactly one distinct value each.
+   */
+  private val adaptivePartialAggEnabled: Boolean = {
+    conf.adaptivePartialAggregationEnabled &&
+      groupingExpressions.nonEmpty &&
+      // Only the pre-shuffle partial aggregation has a downstream `Final` to 
merge passed-through
+      // single-row buffers. `requiredChildDistributionExpressions` is `None` 
exactly for that
+      // pre-shuffle phase and `Some` for the `Final`/`Complete` phase. This 
check is what keeps a
+      // group-by-only aggregate (no aggregate functions, so an empty 
`aggregateExpressions`) from
+      // being admitted vacuously: `aggregateExpressions.forall(_.mode == 
Partial)` alone is true
+      // for the empty list, which would wrongly make the `Final` phase 
eligible as well.
+      requiredChildDistributionExpressions.isEmpty &&

Review Comment:
   You are right, and that sentence is gone. The doc now keeps the two 
mechanisms separate: the intermediate phase of a mixed DISTINCT plan is 
excluded by the mode check (its modes are `PartialMerge ++ Partial`), while the 
group-by-only intermediate phase of a `count(DISTINCT v)`-only plan is excluded 
by `requiredChildDistributionExpressions`. What makes the eligible distinct 
`Partial` phase safe is now stated as what you found: its output partitioning 
`(k, v)` does not satisfy the `Final`'s `ClusteredDistribution(k)`, so another 
`Exchange` lands between them and there is still a `Final` to merge the 
passed-through buffers. The PR description says the same.



-- 
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]

Reply via email to