peter-toth commented on code in PR #57602:
URL: https://github.com/apache/spark/pull/57602#discussion_r3685072984
##########
sql/core/src/test/scala/org/apache/spark/sql/DataFrameWindowFunctionsSuite.scala:
##########
@@ -1682,6 +1682,39 @@ class DataFrameWindowFunctionsSuite extends
SharedSparkSession
}
}
+ test("SPARK-58404: bypass partial WindowGroupLimit") {
+ val df = Seq(
+ ("a", 0, "c"),
+ ("a", 1, "x"),
+ ("a", 2, "y"),
+ ("b", 1, "h"),
+ ("b", 1, "n"),
+ ("c", 1, "z"),
+ ("c", 2, "a")).toDF("key", "value", "order")
+
+ val window = Window.partitionBy($"key").orderBy($"order")
+ val expected = Seq(
+ Row("a", 0, "c", 1),
+ Row("b", 1, "h", 1),
+ Row("c", 2, "a", 1))
+
+ Seq(true, false).foreach { bypass =>
+ withSQLConf(
+ SQLConf.BYPASS_PARTIAL_WINDOW_GROUP_LIMIT.key -> bypass.toString,
+ SQLConf.WINDOW_GROUP_LIMIT_THRESHOLD.key -> "100") {
+ val result = df.withColumn("rn",
row_number().over(window)).where($"rn" === 1)
+ checkAnswer(result, expected)
+
+ val limits = collect(result.queryExecution.executedPlan) {
+ case w: WindowGroupLimitExec => w
+ }
+ // When bypassed, only the final WindowGroupLimit remains; otherwise
both partial and
+ // final are present since a shuffle is required.
+ assert(limits.size === (if (bypass) 1 else 2))
+ }
+ }
+ }
Review Comment:
**Finding 1** (Blocking) **and finding 2** (Non-blocking), both on this
point -- I read the empty-`partitionSpec` case the other way round, so flagging
it here rather than opening a new thread.
The partial pass never reduces cardinality *across* partitions -- not here,
and not with a partition spec either. It's a `mapPartitions` filter, so all it
ever does is prune *within* one input partition. And that pruning is sound
because a row's global rank is always >= its rank inside a single input
partition: anything with global rank <= `limit` also has local rank <= `limit`,
so the partial can't drop a row the final needs. What it gives you is a hard
output bound of `limit` rank groups **per input partition**, independent of the
data.
That bound is why the empty-`partitionSpec` case is where the partial is
worth *most*, not least:
```
-- bypass off
WindowGroupLimitExec Final <- AllTuples
+- Sort [o]
+- Exchange SinglePartition <- carries ~ numPartitions * limit rank
groups
+- WindowGroupLimitExec Partial
+- Sort [o] <- one local sort per input partition,
in parallel
+- Scan t
-- bypass on
WindowGroupLimitExec Final
+- Sort [o] <- sorts the whole table, one task
+- Exchange SinglePartition <- carries the whole table
+- Scan t
```
So the flag here costs a full-table shuffle into one reducer plus a
single-threaded sort of everything, and buys back only the parallel local sorts
it skipped. Downside unbounded, upside bounded.
With a partition spec it's the opposite: the exchange is a hash exchange to
many reducers so post-shuffle work stays parallel, and the partial's bound is
`limit` rank groups per input partition **per key** -- many keys with few rows
each means it prunes nothing and is pure overhead. That is the "low reduction
ratio" workload the config is for, and it can only really happen when there is
a partition spec.
This is the same asymmetry `bypassPartialAggregation` ran into --
`AggUtils.scala:136-141`:
> The bypass is only beneficial when there are grouping keys
(`groupingExpressions.nonEmpty`): global aggregations (no GROUP BY) always
produce a single output row, so the pre-shuffle partial aggregation achieves
the maximum possible reduction ratio and should never be skipped. Bypassing a
global aggregation would shuffle all raw rows to a single partition with no
benefit, which is strictly worse than the normal Partial+Final path.
An empty `partitionSpec` is that same situation, so I'd mirror the guard in
`SparkStrategies.scala:808`:
```scala
val finalChild = if (conf.bypassPartialWindowGroupLimit &&
partitionSpec.nonEmpty) {
```
If leaving it ungated is a deliberate "the user asked for it, don't
second-guess" call, that's defensible, but then it's worth stating in the
config doc.
Either way, the comment added at
`DataFrameWindowFunctionsSuite.scala:1723-1727` needs rewording -- it currently
writes "the partial pass (which cannot reduce cardinality across partitions
here)" into the test as the justification. Separately in that same comment, the
`row_number()` note drops a condition: `InferWindowGroupLimit.scala:118-122`
only rewrites to `Limit` when `partitionSpec` is empty **and** `limit <
topKSortFallbackThreshold`. It holds for `limit = 1` here, so the test is fine,
but the exclusion isn't unconditional.
--
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]
