peter-toth commented on code in PR #57986:
URL: https://github.com/apache/spark/pull/57986#discussion_r3784879544
##########
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:
**Finding 1.** `InferWindowGroupLimit.isExpandingWindow` requires *every*
window expression to carry `(RowFrame, UnboundedPreceding, CurrentRow)`:
```scala
// InferWindowGroupLimit.scala:78
case Alias(WindowExpression(windowFunction, WindowSpecDefinition(_, _,
SpecifiedWindowFrame(RowFrame, UnboundedPreceding, CurrentRow))), _)
if !windowFunction.isInstanceOf[SizeBasedWindowFunction] => true
```
`canEvaluateUnderAnyOrder` admits exactly the opposite upper bound, so any
window this relaxation produces necessarily fails
`windowExpressions.forall(isExpandingWindow)` at
`InferWindowGroupLimit.scala:95` -- and that rule runs in `SparkOptimizer`'s
"Infer window group limit" batch, i.e. after this one.
The shape is reachable. `ExtractWindowExpressions` folds its `LinkedHashMap`
in insertion order (`Analyzer.scala:3677`), so the first window spec in the
select list becomes the innermost `Window`; writing the unordered aggregate
first makes the ordered window the parent, which is what the `Filter` sits on.
Measured on this head with a catalyst-only `RuleExecutor` over `Filter(rn <= 2,
Window(rn, [c], [a desc], Window(cnt, [c], Nil, t)))`:
```
CollapseWindow excluded: WindowGroupLimit [c#2], [a#0 DESC], row_number(), 2
CollapseWindow enabled: <no WindowGroupLimit>
```
Trading a per-partition early stop for one saved `WindowExec` pass is the
wrong way round on a large partition. Relaxing `isExpandingWindow` is not the
fix -- `WindowGroupLimit` drops rows *below* the window, which would change
`count(1) OVER (PARTITION BY k)`.
The cheap fix is to drop the child-empty branch and keep only the
parent-empty one:
```scala
(specCompatible(w1.orderSpec, w2.orderSpec) ||
(w1.orderSpec.isEmpty && w2.orderSpec.nonEmpty &&
w1.windowExpressions.forall(canEvaluateUnderAnyOrder))) &&
```
That direction cannot lose a `WindowGroupLimit`: the `Filter` then sits on
the empty-order window, which already fails `orderSpec.nonEmpty` on base, so
there is nothing to lose. It keeps the motivating example and the benchmark
(`row_number()` written before `count(1)` puts the empty-order window on top),
and it also settles the order-change question in @zml1206's thread -- see
finding 3. If you want the child-empty direction too, it needs a conf so a
regressed query has an escape hatch.
##########
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:
**Finding 6.** The object scaladoc just above (`Optimizer.scala:1753-1757`)
still states the old precondition:
```
* Collapse Adjacent Window Expression.
* - If the partition specs and order specs are the same and the window
expression are
* independent and are of the same window function type, collapse into the
parent.
```
Worth extending with the empty-order case -- that is the doc a reader hits
first.
##########
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:
**Finding 7.** The frame type is left free here, so `RANGE BETWEEN UNBOUNDED
PRECEDING AND UNBOUNDED FOLLOWING` merges as well. That is legal with an empty
order spec -- `WindowSpecDefinition.checkInputDataTypes` only rejects a
`RangeFrame` that is *not* unbounded (`windowExpressions.scala:74`) -- and it
is correct, because `WindowEvaluatorFactoryBase` maps `("AGGREGATE", _,
UnboundedPreceding, UnboundedFollowing, _)` to `UnboundedWindowFunctionFrame`
regardless of frame type. But both the PR description and the scaladoc describe
the case as `ROWS ...`, and no test covers `RANGE`. One extra case in the suite
and a word in the doc would pin it down.
##########
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:
**Finding 4.** After this the operator's `orderSpec` is non-empty while the
merged-in expressions still carry `WindowSpecDefinition(partitionSpec, Nil,
frame)`. That drops an invariant `ExtractWindowExpressions` sets up -- it keys
the grouping on the *expression's* own `(partitionSpec, orderSpec,
functionType)` (`Analyzer.scala:3667`), so until now the operator's spec always
matched its expressions'. The sibling rule keeps the analogous field in sync
rather than letting it drift:
```scala
// EliminateWindowPartitions.scala:36 -- rewrites the expression's spec, not
just the operator's
val newWsd = wsd.copy(partitionSpec = ps.filter(!_.foldable))
```
I traced the readers and nothing breaks today: only `WindowResolution`
(analyzer, already run) and `OptimizeWindowFunctions` (`Optimizer.scala:1745`,
which needs `orderSpec.nonEmpty` and so stays a no-op on the merged-in
expression) look at a window expression's own order spec. So this is a "say
why", not a "fix" -- a line in the rule comment noting the divergence is
deliberate would help, because it is surprising in `EXPLAIN`. Worth knowing the
flip side if you ever do sync them: `OptimizeWindowFunctions` would then
rewrite an empty-order `first` to `nth_value(_, 1)`, which is the same value
under the merged order.
##########
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:
**Finding 5.** (Alternative design, not a defect -- take it or leave it.)
The safety argument is much easier to make one layer down.
`PushDownLocalSort` (`spark.sql.execution.pushDownLocalSort`, default on since
4.3.0) already widens the lower local sort through an empty-order `WindowExec`
and drops the upper one; its `isOrderPreserving` has `case _: WindowExecBase =>
true`, with a comment making essentially the argument this scaladoc makes. I
confirmed on this head that the base plan for the child-empty shape is already
a single `Sort [c1 ASC, c2 ASC]` feeding both windows, and that base and merged
results are identical under the default conf.
So a physical rule beside it could merge two adjacent `WindowExec`s whenever
the lower one's `requiredChildOrdering` is already satisfied by what is
actually below it. That shape:
- needs no order-insensitivity judgement at all -- the orderings provably
coincide, so `canEvaluateUnderAnyOrder` and the whole frame check disappear;
- runs after `InferWindowGroupLimit`, so finding 1 does not arise (a
`WindowGroupLimitExec` between the two windows simply blocks it);
- gets the `Project`-between case for free, since `PushDownLocalSort`
already walks through a deterministic `ProjectExec`.
Counter-argument, and it is a real one: far more work than this two-line
change -- `windowFrameExpressionFactoryPairs` has to be rebuilt for the merged
operator and the output attributes rewired, and physical rules are harder to
follow. I mainly wanted the option on record.
##########
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:
**Finding 2.** All six new tests run `Optimize` = `CollapseWindow` +
`CollapseProject`, so they assert that the merge happens and nothing about what
the merged shape does to the rules that run after it -- which is where finding
1 lives. Two gaps:
1. A case that runs `InferWindowGroupLimit` on the merged plan.
`InferWindowGroupLimitSuite` already has the relation and the rule set to copy:
```scala
val batches = Batch("...", FixedPoint(10),
CollapseWindow, CollapseProject, RemoveNoopOperators, PushDownPredicates,
InferWindowGroupLimit) :: Nil
```
driven by `testRelation.window(Seq(cnt), Seq(c), Nil).window(Seq(rn),
Seq(c), Seq(a.desc)).where($"rn" <= 2)`, asserting a `WindowGroupLimit` is
still there.
2. An execution-level result test, e.g. in `DataFrameWindowFunctionsSuite`.
This rule now changes the row order an aggregate sees, and
`CollapseWindowSuite` is a `PlanTest`, so nothing in this PR ever runs a merged
window.
--
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]
