[SPARK-58757][SQL] Allow CollapseWindow to merge windows with an empty order spec

[ulysses-you]

What changes were proposed in this pull request?

Currently, CollapseWindow collapses two adjacent Window operators only when their partition specs and order specs are identical. This PR relaxes it to also merge two windows with the same partition spec when one of them has an empty order spec, as long as every window expression of the empty-order window is order-insensitive.

A window expression is treated as order-insensitive when its frame is the whole partition (ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING): such a frame always covers every row of the partition regardless of the ordering, so aggregates like count/sum/min/max give the same value under any ordering, and functions whose result does depend on the row order (e.g. collect_list, first) are non-deterministic when the order spec is empty, so evaluating them under any ordering yields a valid result. Windows with a bounded frame (e.g. ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) are order-sensitive and are never merged.

The merged window keeps the non-empty order spec of the other window.

Why are the changes needed?

For queries that mix an ordered window function with an unordered aggregate over the same partition, e.g.:

SELECT c2, c1,
       row_number() OVER (PARTITION BY c1 ORDER BY c2) AS rk,
       count(1)     OVER (PARTITION BY c1)
FROM t3

the current rule keeps two Window operators even though they share PARTITION BY c1. After this change they are collapsed into a single window operator, saving one WindowExec pass (the [c1, c2] sort is already shared in both plans). A local benchmark on 4M rows showed roughly 16% faster runtime in the non-spill case and 20% in the spill case.

Does this PR introduce any user-facing change?

The query result is unchanged for the common shape (the empty-order window written after the ordered one), where the merge does not change the input order of any window expression. When the empty-order window appears before an ordered sibling in the SELECT list, the merge evaluates its expressions under the sibling's order; for functions documented as non-deterministic without an order (first, last, collect_list) the value may differ, which is already allowed by their contract. FP sum/avg may also differ at the bit level, consistent with Spark's existing treatment of FP aggregation as order-sensitive (EliminateSorts.isOrderIrrelevantAggs).

How was this patch tested?

Added tests to CollapseWindowSuite covering:

• collapse when the empty-order window has a whole-partition frame (row_number + count), including the case where it is the inner window;
• collapse when the empty-order window has multiple window expressions;
• collapse when the empty-order window has first over the whole partition;
• the SPARK-34565 shape with a Project between the windows;
• no collapse when the empty-order window has a bounded frame.

Ran CollapseWindowSuite (13 tests) and TransposeWindowSuite (8 tests), all pass.

Was this patch authored or co-authored using generative AI tooling?

Yes, developed with assistance from Claude Code.

Generated-by: Claude Code

[uros-b]

LGTM, thank you @ulysses-you!

[zml1206]

This check seems too broad. A whole-partition frame guarantees that the same rows are included, but it does not guarantee that every aggregate produces the same result under a different input order.
In particular, floating-point Sum, Average, and CentralMomentAgg are order-sensitive because IEEE-754 arithmetic is not associative. For example:
(1e16 + -1e16) + 1.0 = 1.0
(1e16 + 1.0) + -1e16 = 0.0
Merging an unordered window into an ordered one changes the order in which rows are fed to these aggregates, so it can change a deterministic query result. EliminateSorts.isOrderIrrelevantAggs already handles this by excluding Float/Double variants of these aggregates.
Could we inspect the window function here instead of treating every whole-partition frame as order-insensitive, and add an execution-level regression test for floating-point sum?

[ulysses-you]

I do not think we should concern that, otherwise we can not optimize any ordering related plan, such as eliminate sort, a smj -> bhj/shj when the join is the child of a containing floating-point arithmetic Agg, also the result of sort-based agg is different with hash-based. So if people require an accurate result, they should use decimal type.

[zml1206]

Thanks, that makes sense.
Could we adjust the comment of orderInsensitive? The statement that sum gives the same value under any ordering is technically incorrect for floating-point inputs, and orderInsensitive is also slightly misleading for functions such as first and collect_list, whose values may change but remain valid under their nondeterministic contract.

[ulysses-you]

thank you, addressed!

[peter-toth]

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:

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))

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.

[ulysses-you]

thank you @uros-b !, also cc @cloud-fan @peter-toth

[peter-toth]

Thanks for the PR, @ulysses-you!

The relaxation reads right to me: a whole-partition frame covers the same rows under any order, and the merged window keeps the ordered sibling's spec. What I'd want fixed before merge is a downstream effect nobody has mentioned yet -- the merged window is no longer eligible for InferWindowGroupLimit, so a top-k-per-group query that also selects an unordered aggregate loses its WindowGroupLimit (measured: 1 node -> 0). Restricting the merge to the direction where the empty-order window is the parent fixes that, keeps your motivating example and benchmark, and settles @zml1206's thread at the same time. On that thread I have a measurement to add rather than a new objection.

Blocking

• 1. Merging makes the window ineligible for WindowGroupLimit: InferWindowGroupLimit needs every window expression to have a (RowFrame, UnboundedPreceding, CurrentRow) frame, which is exactly the frame this relaxation cannot produce, so ... count(1) OVER (PARTITION BY k), row_number() OVER (PARTITION BY k ORDER BY a) rn ... WHERE rn <= 5 loses its per-partition early stop. Only the child-empty direction is affected. [inline: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:1792]
• 2. Nothing tests the merged shape past this rule: all six new tests run CollapseWindow alone and are plan-only, so finding 1 is invisible to the suite and no test ever executes a merged window. [inline: sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/CollapseWindowSuite.scala:177]

Non-blocking

• 3. Order-sensitivity scope (@zml1206's thread -- not re-opening it): two things that thread didn't have. With spark.sql.execution.pushDownLocalSort=false a deterministic Scala UDAF changes value (8 -> 2) in the child-empty direction, and "use decimal type" can't fix a UDAF -- EliminateSorts.isOrderIrrelevantAggs has case _: UserDefinedExpression => false for exactly this. Conversely the exposure is narrower than it looks and the description cites the wrong precedent: PushDownLocalSort supports this change, isOrderIrrelevantAggs argues against it. Detail in a reply on that thread.
• 4. Operator and expression order specs now disagree: the merged Window.orderSpec is non-empty while its merged-in expressions keep an empty WindowSpecDefinition.orderSpec. I traced the readers and nothing breaks today, but EliminateWindowPartitions keeps the analogous partitionSpec in sync, so the divergence wants a comment. [inline: Optimizer.scala:1810]

Alternatives

• 5. Do the merge one layer down: PushDownLocalSort already arranges for both windows to see the same row order, so a physical merge of adjacent WindowExecs needs no order-insensitivity judgement at all, and it runs after InferWindowGroupLimit. Much more work; your call. [inline: Optimizer.scala:1780]

Minor

• 6. Stale rule doc: the CollapseWindow object scaladoc still says the rule needs identical order specs. [inline: Optimizer.scala:1764]
• 7. RANGE whole-partition frame: the pattern leaves the frame type free, so RANGE BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING merges too -- correct, but undocumented and untested. [inline: Optimizer.scala:1783]

[peter-toth]

Finding 1. InferWindowGroupLimit.isExpandingWindow requires every window expression to carry (RowFrame, UnboundedPreceding, CurrentRow):

// 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:

      (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.

[peter-toth]

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:

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.

[peter-toth]

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:

// 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.

[peter-toth]

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 WindowExecs 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.

[peter-toth]

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.

[peter-toth]

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.