ulysses-you commented on code in PR #57986:
URL: https://github.com/apache/spark/pull/57986#discussion_r3795075763
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -1761,9 +1761,39 @@ object CollapseWindow extends Rule[LogicalPlan] {
s1.zip(s2).forall(e => e._1.semanticEquals(e._2))
}
+ /**
+ * Returns true if the given window expression can still be evaluated
correctly when the rows
+ * of the partition are reordered, so that it can be merged into another
window with a different
+ * (non-empty) order spec.
+ *
+ * The frame determines whether reordering is safe. 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 reordering changes only the
order in which the rows
+ * are seen, never which rows are in the frame. Since the order spec of the
window is empty,
+ * the query does not fix the row order, so evaluating its expressions under
any ordering
+ * yields a valid result, even though the value may differ for
order-dependent expressions
+ * such as `first`, `collect_list`, or floating-point `sum`/`avg`. 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 canEvaluateUnderAnyOrder(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 safe to evaluate under
any row order. In that
+ // case, they can be evaluated under the non-empty order spec of the
other window.
+ (specCompatible(w1.orderSpec, w2.orderSpec) ||
Review Comment:
Agreed the two directions are not equally safe, and that the child-empty
merge is the one that can drop `WindowGroupLimit`. I've added
`spark.sql.optimizer.collapseWindowWithEmptyOrderSpecInChild` (default false)
as the escape hatch you suggested, so top-k queries keep `WindowGroupLimit` by
default. I'm neutral on keeping that config, though — this interaction is
pre-existing rather than new: `CollapseWindow` never checks frames, so the same
loss already happens when merging two windows with the *same* order spec, e.g.
`row_number() OVER (ORDER BY c)` + `count() OVER (ORDER BY c ROWS BETWEEN
U..U)` under `WHERE rn <= 2`. The child-empty case is also narrow: the `count`
window below already forces a full `[k]` shuffle + `[k,a]` sort.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/CollapseWindowSuite.scala:
##########
@@ -168,4 +173,164 @@ class CollapseWindowSuite extends PlanTest {
comparePlans(optimized, correctAnswer)
}
+
+ test("collapse windows when one has an empty order spec " +
Review Comment:
Added an execution-level test in `DataFrameWindowFunctionsSuite`, and a test
in `InferWindowGroupLimitSuite` asserting the child-empty direction keeps
`WindowGroupLimit` by default and drops it with the config on.
##########
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:
Updated the description to cite `PushDownLocalSort.isOrderPreserving`.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -1776,13 +1806,19 @@ object CollapseWindow extends Rule[LogicalPlan] {
_.containsPattern(WINDOW), ruleId) {
case w1 @ Window(we1, _, _, w2 @ Window(we2, _, _, grandChild, _), _)
if windowsCompatible(w1, w2) =>
- w1.copy(windowExpressions = we2 ++ we1, child = grandChild)
+ w1.copy(
+ orderSpec = if (w1.orderSpec.nonEmpty) w1.orderSpec else w2.orderSpec,
Review Comment:
Added a comment noting the divergence is deliberate and why it's safe.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -1761,9 +1761,39 @@ object CollapseWindow extends Rule[LogicalPlan] {
s1.zip(s2).forall(e => e._1.semanticEquals(e._2))
}
+ /**
+ * Returns true if the given window expression can still be evaluated
correctly when the rows
+ * of the partition are reordered, so that it can be merged into another
window with a different
+ * (non-empty) order spec.
+ *
+ * The frame determines whether reordering is safe. 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 reordering changes only the
order in which the rows
+ * are seen, never which rows are in the frame. Since the order spec of the
window is empty,
+ * the query does not fix the row order, so evaluating its expressions under
any ordering
+ * yields a valid result, even though the value may differ for
order-dependent expressions
+ * such as `first`, `collect_list`, or floating-point `sum`/`avg`. 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 canEvaluateUnderAnyOrder(windowExpression: NamedExpression):
Boolean =
Review Comment:
Agreed a physical merge beside `PushDownLocalSort` would be architecturally
cleaner — it runs after `InferWindowGroupLimit` and needs no
order-insensitivity judgment. But that means moving the entire `CollapseWindow`
rule to the physical plan (rebuilding `windowFrameExpressionFactoryPairs`,
rewiring outputs), which is a much larger change. The WindowGroupLimit
interaction it avoids is, like Finding 1, a pre-existing property of the
logical rule (already present for same-order-spec merges), so I've kept the
logical rule with the config as a targeted escape hatch. Open to a follow-up
for the physical merge.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -1761,9 +1761,39 @@ object CollapseWindow extends Rule[LogicalPlan] {
s1.zip(s2).forall(e => e._1.semanticEquals(e._2))
}
+ /**
Review Comment:
Updated.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -1761,9 +1761,39 @@ object CollapseWindow extends Rule[LogicalPlan] {
s1.zip(s2).forall(e => e._1.semanticEquals(e._2))
}
+ /**
+ * Returns true if the given window expression can still be evaluated
correctly when the rows
+ * of the partition are reordered, so that it can be merged into another
window with a different
+ * (non-empty) order spec.
+ *
+ * The frame determines whether reordering is safe. 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 reordering changes only the
order in which the rows
+ * are seen, never which rows are in the frame. Since the order spec of the
window is empty,
+ * the query does not fix the row order, so evaluating its expressions under
any ordering
+ * yields a valid result, even though the value may differ for
order-dependent expressions
+ * such as `first`, `collect_list`, or floating-point `sum`/`avg`. 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 canEvaluateUnderAnyOrder(windowExpression: NamedExpression):
Boolean =
+ windowExpression match {
+ case Alias(WindowExpression(_, WindowSpecDefinition(_, _,
+ SpecifiedWindowFrame(_, UnboundedPreceding, UnboundedFollowing))),
_) => true
Review Comment:
Added a `RANGE BETWEEN U..U` plan test.
--
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]
