peter-toth commented on code in PR #57986:
URL: https://github.com/apache/spark/pull/57986#discussion_r3784891087
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -1761,9 +1761,38 @@ object CollapseWindow extends Rule[LogicalPlan] {
s1.zip(s2).forall(e => e._1.semanticEquals(e._2))
}
+ /**
+ * Returns true if the given window expression can be evaluated under any
ordering of the rows
+ * within a partition without changing the result, so that it can be merged
into another window
+ * with a different (non-empty) order spec.
+ *
+ * The frame determines whether the ordering matters. When the frame is the
whole partition
+ * (`UNBOUNDED PRECEDING` to `UNBOUNDED FOLLOWING`), it always covers all
the rows of the
+ * partition regardless of the ordering, so the ordering does not affect the
result: aggregates
+ * such as `count` or `sum` give the same value under any ordering, and
functions whose result
+ * does depend on the row order, such as `collect_list` or `first`, are
non-deterministic when
+ * the order spec is empty, so evaluating them under any ordering yields a
valid result. On the
+ * other hand, a bounded frame (e.g. `ROWS BETWEEN UNBOUNDED PRECEDING AND
CURRENT ROW`) is
+ * order-sensitive: which rows are in the frame depends on the ordering, so
even `count` or
+ * `sum` would change value, and such a window must not be merged.
+ */
+ private def orderInsensitive(windowExpression: NamedExpression): Boolean =
+ windowExpression match {
+ case Alias(WindowExpression(_, WindowSpecDefinition(_, _,
+ SpecifiedWindowFrame(_, UnboundedPreceding, UnboundedFollowing))),
_) => true
+ case _ => false
+ }
+
private def windowsCompatible(w1: Window, w2: Window): Boolean = {
specCompatible(w1.partitionSpec, w2.partitionSpec) &&
- specCompatible(w1.orderSpec, w2.orderSpec) &&
+ // The order specs can differ when one of them is empty, as long as the
window expressions
+ // of the window with the empty order spec are insensitive to the row
order. In that case,
+ // they can be evaluated under the non-empty order spec of the other
window.
+ (specCompatible(w1.orderSpec, w2.orderSpec) ||
+ (w1.orderSpec.isEmpty && w2.orderSpec.nonEmpty &&
+ w1.windowExpressions.forall(orderInsensitive)) ||
+ (w2.orderSpec.isEmpty && w1.orderSpec.nonEmpty &&
+ w2.windowExpressions.forall(orderInsensitive))) &&
Review Comment:
Late here, and I'm not re-opening this -- @ulysses-you's answer on FP
arithmetic reads fine to me. Two things I measured on this head that seem worth
having on record.
First, in @ulysses-you's favour: the exposure is narrower than it looks.
`PushDownLocalSort` (`spark.sql.execution.pushDownLocalSort`, default on since
4.3.0) already widens the lower local sort through an empty-order `WindowExec`,
so the base plan for the inner-window shape is already a single `Sort [c1 ASC,
c2 ASC]` feeding both windows. With default confs I get identical results
before and after this PR in both merge directions.
`PushDownLocalSort.isOrderPreserving` even has `case _: WindowExecBase => true`
with a comment making this PR's argument. That, rather than
`EliminateSorts.isOrderIrrelevantAggs`, is the precedent I'd cite in the
description -- as written, the description cites a helper that argues the other
way.
Second, a case the "use decimal type" answer doesn't cover. With
`spark.sql.execution.pushDownLocalSort=false`, a *deterministic* Scala UDAF
changes value:
```scala
case class FirstBuf(seen: Boolean, value: Int)
object FirstIntAgg extends Aggregator[Int, FirstBuf, Int] { /* returns the
first row it sees */ }
spark.udf.register("udaf_first", udaf(FirstIntAgg))
```
```sql
SELECT DISTINCT c1, cl, fst, udafFst FROM (
SELECT c1,
collect_list(c2) OVER (PARTITION BY c1) AS cl,
first(c2) OVER (PARTITION BY c1) AS fst,
udaf_first(c2) OVER (PARTITION BY c1) AS udafFst,
row_number() OVER (PARTITION BY c1 ORDER BY c2) AS rk
FROM t3) WHERE rk >= 1
```
```
base: [0, [8, 4, 2, 10, 6], 8, 8]
PR: [0, [2, 4, 6, 8, 10], 2, 2]
```
`isOrderIrrelevantAggs` has `case _: UserDefinedExpression => false` for
precisely this, and a user cannot switch a UDAF to decimal. Same for a pandas
`GROUPED_AGG` UDF, which is deterministic unless marked otherwise.
Worth noting the two merge directions are not equally exposed. When the
empty-order window is the *parent* -- the shape in the PR description and the
one the benchmark measures -- the merge cannot change anything:
`WindowExecBase.requiredChildOrdering` for an empty order spec is just
`[partitionSpec]`, which the ordered child window's `[partitionSpec ++
orderSpec]` output ordering already satisfies, so no sort is inserted and the
row order is identical. I measured that direction as unchanged even with
`pushDownLocalSort=false`. Everything above is the *child*-empty direction. So
restricting the relaxation to the parent-empty case removes this question
entirely -- and it also fixes a `WindowGroupLimit` regression I raised
separately as finding 1.
--
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]
