cloud-fan commented on code in PR #58106:
URL: https://github.com/apache/spark/pull/58106#discussion_r3818968034
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/aggregate/AdaptivePartialAggregationSuite.scala:
##########
@@ -477,10 +496,93 @@ class AdaptivePartialAggregationSuite extends QueryTest
with SharedSparkSession
}
}
+ test("distinct with plain and filtered non-distinct aggregates") {
+ // One query carries all three shapes through the DISTINCT intermediate
phase
+ // (`PartialMerge ++ Partial`): the distinct aggregate (`count(DISTINCT
v)`), a plain
+ // non-distinct aggregate (`sum(v)`), and a filtered non-distinct aggregate
+ // (`avg(v) FILTER (...)`, whose `FILTER` is applied in the leading
`Partial` phase only).
+ // Fully distinct keys and values make neither partial phase reduce
anything, so both bypass in
+ // the same execution: asserting the 2-key phase (de-duplication, all
`Partial`) and the 1-key
+ // phase (distinct partial, `PartialMerge ++ Partial`) together proves the
two bypasses coexist,
+ // the plain and filtered non-distinct buffers pass through correctly, and
the results still
+ // match the feature-off reference.
+ withTempView("t") {
+ spark.range(0, 400, 1, 1)
+ .select($"id".cast("string") as "k", $"id" as "v")
+ .createOrReplaceTempView("t")
+ forEachCodegenAndMap() { clue =>
+ val df = () => spark.sql(
+ """SELECT k,
+ | count(DISTINCT v) AS cd,
+ | sum(v) AS s,
+ | avg(v) FILTER (WHERE v > 25) AS a_gt25
+ |FROM t GROUP BY k""".stripMargin)
+ withClue(clue) {
+ val byKeyCount = bypassRowsByGroupingKeyCount(df)
+ assert(byKeyCount.get(2).exists(_ > 0),
+ s"expected the de-duplication partial (grouping on k, v) to
bypass, got $byKeyCount")
+ assert(byKeyCount.get(1).exists(_ > 0),
+ s"expected the distinct partial (PartialMerge++Partial, grouping
on k) to bypass, " +
+ s"got $byKeyCount")
+ }
+ }
+ }
+ }
+
+ test("distinct with an order-sensitive non-distinct aggregate across
partitions") {
+ // A single-partition `range` fuses the whole four-phase DISTINCT plan
into one stage, so the
+ // split-topology path (the frozen map draining one row per queued row,
the queue flush, and
+ // the `shouldStop()` re-entry) never runs for a `PartialMerge` member,
nor does an
Review Comment:
**Nit:**
The aggregate member is not the thing that `runs` here; the split-topology
path is.
```suggestion
// the `shouldStop()` re-entry) never runs for a `PartialMerge` member,
including an
```
--
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]
