Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ulysses-you wants to merge 2 commits into
base: master
Choose a base branch
from
+204 −3
Open

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
334258c
[SPARK-58757][SQL] Allow CollapseWindow to merge windows with an empt…
ulysses-you Aug 13, 2026
a7ee6d5
[SPARK-58757][SQL] Allow CollapseWindow to merge windows with an empt…
ulysses-you Aug 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
42 changes: 39 additions & 3 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1761,9 +1761,39 @@ object CollapseWindow extends Rule[LogicalPlan] {
s1.zip(s2).forall(e => e._1.semanticEquals(e._2))
}

/**

@peter-toth peter-toth Aug 14, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

* 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 =

@peter-toth peter-toth Aug 14, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

windowExpression match {
case Alias(WindowExpression(_, WindowSpecDefinition(_, _,
SpecifiedWindowFrame(_, UnboundedPreceding, UnboundedFollowing))), _) => true

@peter-toth peter-toth Aug 14, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

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

@peter-toth peter-toth Aug 14, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

(w1.orderSpec.isEmpty && w2.orderSpec.nonEmpty &&
w1.windowExpressions.forall(canEvaluateUnderAnyOrder)) ||
(w2.orderSpec.isEmpty && w1.orderSpec.nonEmpty &&
w2.windowExpressions.forall(canEvaluateUnderAnyOrder))) &&
w1.references.intersect(w2.windowOutputSet).isEmpty &&
w1.windowExpressions.nonEmpty && w2.windowExpressions.nonEmpty &&
// This assumes Window contains the same type of window expressions. This is ensured
Expand All @@ -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,

@peter-toth peter-toth Aug 14, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

windowExpressions = we2 ++ we1,
child = grandChild)

case w1 @ Window(we1, _, _, Project(pl, w2 @ Window(we2, _, _, grandChild, _)), _)
if windowsCompatible(w1, w2) && w1.references.subsetOf(grandChild.outputSet) =>
Project(
pl ++ w1.windowOutputSet,
w1.copy(windowExpressions = we2 ++ we1, child = grandChild))
w1.copy(
orderSpec = if (w1.orderSpec.nonEmpty) w1.orderSpec else w2.orderSpec,
windowExpressions = we2 ++ we1,
child = grandChild))
}
}

Expand Down
165 changes: 165 additions & 0 deletions ...catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/CollapseWindowSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,11 @@ package org.apache.spark.sql.catalyst.optimizer

import org.apache.spark.sql.catalyst.dsl.expressions._
import org.apache.spark.sql.catalyst.dsl.plans._
import org.apache.spark.sql.catalyst.expressions.{
CurrentRow, RowFrame, RowNumber, SpecifiedWindowFrame,
UnboundedFollowing, UnboundedPreceding}
import org.apache.spark.sql.catalyst.expressions.aggregate.{
AggregateExpression, Complete, Count, First, Sum}
import org.apache.spark.sql.catalyst.plans.PlanTest
import org.apache.spark.sql.catalyst.plans.logical.{LocalRelation, LogicalPlan}
import org.apache.spark.sql.catalyst.rules.RuleExecutor
Expand Down Expand Up @@ -168,4 +173,164 @@ class CollapseWindowSuite extends PlanTest {

comparePlans(optimized, correctAnswer)
}

test("collapse windows when one has an empty order spec " +

@peter-toth peter-toth Aug 14, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

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

"(row_number + count over the whole partition)") {
val rk = windowExpr(
RowNumber(),
windowSpec(partitionSpec1, orderSpec1,
SpecifiedWindowFrame(RowFrame, UnboundedPreceding, CurrentRow))).as("rk")
val cnt = windowExpr(
AggregateExpression(Count(c), Complete, isDistinct = false, None),
windowSpec(partitionSpec1, Nil,
SpecifiedWindowFrame(RowFrame, UnboundedPreceding, UnboundedFollowing))).as("cnt")

val query = testRelation
.window(Seq(rk), partitionSpec1, orderSpec1)
.window(Seq(cnt), partitionSpec1, Nil)

val analyzed = query.analyze
val optimized = Optimize.execute(analyzed)
assert(analyzed.output === optimized.output)

val correctAnswer = testRelation
.window(Seq(rk, cnt), partitionSpec1, orderSpec1)

comparePlans(optimized, correctAnswer)
}

test("collapse windows when the empty-order window has multiple window expressions") {
// Every window expression of the empty-order window must be order-insensitive for the merge.
val rk = windowExpr(
RowNumber(),
windowSpec(partitionSpec1, orderSpec1,
SpecifiedWindowFrame(RowFrame, UnboundedPreceding, CurrentRow))).as("rk")
val cnt = windowExpr(
AggregateExpression(Count(c), Complete, isDistinct = false, None),
windowSpec(partitionSpec1, Nil,
SpecifiedWindowFrame(RowFrame, UnboundedPreceding, UnboundedFollowing))).as("cnt")
val sm = windowExpr(
AggregateExpression(Sum(b), Complete, isDistinct = false, None),
windowSpec(partitionSpec1, Nil,
SpecifiedWindowFrame(RowFrame, UnboundedPreceding, UnboundedFollowing))).as("sm")

val query = testRelation
.window(Seq(rk), partitionSpec1, orderSpec1)
.window(Seq(cnt, sm), partitionSpec1, Nil)

val analyzed = query.analyze
val optimized = Optimize.execute(analyzed)
assert(analyzed.output === optimized.output)

val correctAnswer = testRelation
.window(Seq(rk, cnt, sm), partitionSpec1, orderSpec1)

comparePlans(optimized, correctAnswer)
}

test("collapse windows when the empty-order window has first() over the whole partition") {
// `first` is non-deterministic when the order is not determined by the query, so evaluating it
// under the other window's order spec yields a valid result.
val rk = windowExpr(
RowNumber(),
windowSpec(partitionSpec1, orderSpec1,
SpecifiedWindowFrame(RowFrame, UnboundedPreceding, CurrentRow))).as("rk")
val fr = windowExpr(
First(a, ignoreNulls = true).toAggregateExpression(),
windowSpec(partitionSpec1, Nil,
SpecifiedWindowFrame(RowFrame, UnboundedPreceding, UnboundedFollowing))).as("fr")

val query = testRelation
.window(Seq(rk), partitionSpec1, orderSpec1)
.window(Seq(fr), partitionSpec1, Nil)

val analyzed = query.analyze
val optimized = Optimize.execute(analyzed)
assert(analyzed.output === optimized.output)

val correctAnswer = testRelation
.window(Seq(rk, fr), partitionSpec1, orderSpec1)

comparePlans(optimized, correctAnswer)
}

test("don't collapse windows when the empty-order window has a bounded frame") {
// The frame `ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW` is order-sensitive: which rows
// fall in the frame depends on the ordering, so the window cannot be evaluated under the other
// window's order spec.
val rk = windowExpr(
RowNumber(),
windowSpec(partitionSpec1, orderSpec1,
SpecifiedWindowFrame(RowFrame, UnboundedPreceding, CurrentRow))).as("rk")
val cnt = windowExpr(
AggregateExpression(Count(c), Complete, isDistinct = false, None),
windowSpec(partitionSpec1, Nil,
SpecifiedWindowFrame(RowFrame, UnboundedPreceding, CurrentRow))).as("cnt")

val query = testRelation
.window(Seq(rk), partitionSpec1, orderSpec1)
.window(Seq(cnt), partitionSpec1, Nil)

val optimized = Optimize.execute(query.analyze)
val correctAnswer = query.analyze

comparePlans(optimized, correctAnswer)
}

test("collapse windows when the empty-order window is the inner window") {
// The empty-order window can also be the child of the ordered window. In that case its
// expressions are evaluated under the ordered window's order spec, which is valid because all
// of them are order-insensitive.
val rk = windowExpr(
RowNumber(),
windowSpec(partitionSpec1, orderSpec1,
SpecifiedWindowFrame(RowFrame, UnboundedPreceding, CurrentRow))).as("rk")
val cnt = windowExpr(
AggregateExpression(Count(c), Complete, isDistinct = false, None),
windowSpec(partitionSpec1, Nil,
SpecifiedWindowFrame(RowFrame, UnboundedPreceding, UnboundedFollowing))).as("cnt")

val query = testRelation
.window(Seq(cnt), partitionSpec1, Nil)
.window(Seq(rk), partitionSpec1, orderSpec1)

val analyzed = query.analyze
val optimized = Optimize.execute(analyzed)
assert(analyzed.output === optimized.output)

val correctAnswer = testRelation
.window(Seq(cnt, rk), partitionSpec1, orderSpec1)

comparePlans(optimized, correctAnswer)
}

test("collapse windows with a Project between them when one has an empty order spec") {
// The same merge applies when a Project sits between the two windows and only passes through
// columns that are available below the inner window (SPARK-34565 shape).
val rk = windowExpr(
RowNumber(),
windowSpec(partitionSpec1, orderSpec1,
SpecifiedWindowFrame(RowFrame, UnboundedPreceding, CurrentRow))).as("rk")
val cnt = windowExpr(
AggregateExpression(Count(c), Complete, isDistinct = false, None),
windowSpec(partitionSpec1, Nil,
SpecifiedWindowFrame(RowFrame, UnboundedPreceding, UnboundedFollowing))).as("cnt")

val query = testRelation
.window(Seq(cnt), partitionSpec1, Nil)
.select($"a", $"b", $"c", $"cnt")
.window(Seq(rk), partitionSpec1, orderSpec1)
.select($"a", $"b", $"c", $"cnt", $"rk")

val analyzed = query.analyze
val optimized = Optimize.execute(analyzed)
assert(analyzed.output === optimized.output)

val correctAnswer = testRelation
.window(Seq(cnt, rk), partitionSpec1, orderSpec1)
.select(a, b, c, $"cnt", $"rk")
.analyze

comparePlans(optimized, correctAnswer)
}
}