You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ul...@apache.org on 2026/08/20 02:51:08 UTC
(spark) branch master updated: [SPARK-58757][SQL] Allow CollapseWindow to merge windows with an empty order spec
This is an automated email from the ASF dual-hosted git repository.
ulysses-you pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 8565f0d2a688 [SPARK-58757][SQL] Allow CollapseWindow to merge windows with an empty order spec
8565f0d2a688 is described below
commit 8565f0d2a6887e15f55c48b7a2d0e433f0beeaaa
Author: Xiduo You <ul...@gmail.com>
AuthorDate: Thu Aug 20 10:50:53 2026 +0800
[SPARK-58757][SQL] Allow CollapseWindow to merge windows with an empty order spec
### 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. 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 [...]
The merged window keeps the non-empty order spec of the other window.
When the empty-order window is the child (inner) window, merging it into an ordered parent can disable `InferWindowGroupLimit` for top-k queries, so that direction is gated by a new config `spark.sql.optimizer.collapseWindowWithEmptyOrderSpecInChild` (default false).
### Why are the changes needed?
For queries that mix an ordered window function with an unordered aggregate over the same partition, e.g.:
```sql
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?
Adds `spark.sql.optimizer.collapseWindowWithEmptyOrderSpecInChild` (default false) to control whether an empty-order child window is merged into an ordered parent.
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 and the config is enabled, 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 a [...]
### How was this patch tested?
Added tests to `CollapseWindowSuite` covering:
- collapse when the empty-order window has a whole-partition frame (`row_number` + `count`);
- collapse when the empty-order window has a `RANGE` whole-partition frame;
- 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;
- collapse when the empty-order window is the inner window with the config enabled, and no collapse by default.
Added a test to `InferWindowGroupLimitSuite` asserting the child-empty direction keeps `WindowGroupLimit` by default and drops it when the config is enabled.
Added an execution-level test to `DataFrameWindowFunctionsSuite` that runs the merged window and checks its result.
### Was this patch authored or co-authored using generative AI tooling?
Yes, developed with assistance from Claude Code.
Generated-by: Claude Code
Closes #57986 from ulysses-you/collapse-window-empty-order.
Authored-by: Xiduo You <ul...@gmail.com>
Signed-off-by: Xiduo You <ul...@apache.org>
---
.../spark/sql/catalyst/optimizer/Optimizer.scala | 53 ++++-
.../org/apache/spark/sql/internal/SQLConf.scala | 14 ++
.../catalyst/optimizer/CollapseWindowSuite.scala | 250 +++++++++++++++++++++
.../optimizer/InferWindowGroupLimitSuite.scala | 44 +++-
.../spark/sql/DataFrameWindowFunctionsSuite.scala | 22 ++
5 files changed, 378 insertions(+), 5 deletions(-)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala
index ea78b8be49be..77ffd7e29da0 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala
@@ -1754,6 +1754,10 @@ object OptimizeWindowFunctions extends Rule[LogicalPlan] {
* 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.
+ * - If the partition specs are the same and one of the order specs is empty, collapse into the
+ * parent when the window expressions of the empty-order window can be evaluated under any row
+ * order. The merged window keeps the non-empty order spec. Merging an empty-order child is
+ * gated by `spark.sql.optimizer.collapseWindowWithEmptyOrderSpecInChild`.
*/
object CollapseWindow extends Rule[LogicalPlan] {
private def specCompatible(s1: Seq[Expression], s2: Seq[Expression]): Boolean = {
@@ -1761,9 +1765,46 @@ 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. The
+ // operator then keeps the non-empty order spec while the merged-in expressions keep their
+ // own empty order spec; this divergence is safe because after analysis only
+ // OptimizeWindowFunctions reads an expression's own order spec, and it is a no-op without
+ // one. Merging an empty-order child into an ordered parent can disable
+ // InferWindowGroupLimit and LimitPushDownThroughWindow, so that direction is gated by
+ // conf.collapseWindowWithEmptyOrderSpecInChild.
+ (specCompatible(w1.orderSpec, w2.orderSpec) ||
+ (w1.orderSpec.isEmpty && w2.orderSpec.nonEmpty &&
+ w1.windowExpressions.forall(canEvaluateUnderAnyOrder)) ||
+ (conf.collapseWindowWithEmptyOrderSpecInChild && 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
@@ -1776,13 +1817,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,
+ 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))
}
}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
index c76adab2f8a2..6a90932f2c39 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
@@ -4813,6 +4813,17 @@ object SQLConf {
"The threshold of window group limit must be -1, 0 or positive integer.")
.createWithDefault(1000)
+ val COLLAPSE_WINDOW_WITH_EMPTY_ORDER_SPEC_IN_CHILD =
+ buildConf("spark.sql.optimizer.collapseWindowWithEmptyOrderSpecInChild")
+ .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
+ .doc("When true, the optimizer collapses two adjacent windows with the same partition " +
+ "spec into one when the window with the empty order spec is the child (inner) window. " +
+ "This saves a WindowExec pass but can disable the WindowGroupLimit and the LocalLimit " +
+ "push-down optimizations for top-k queries.")
+ .version("4.4.0")
+ .booleanConf
+ .createWithDefault(false)
+
val WINDOW_SEGMENT_TREE_ENABLED =
buildConf("spark.sql.window.segmentTree.enabled")
.withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
@@ -8853,6 +8864,9 @@ class SQLConf extends Serializable with Logging with SqlApiConf {
def coalesceShufflePartitionsEnabled: Boolean = getConf(COALESCE_PARTITIONS_ENABLED)
+ def collapseWindowWithEmptyOrderSpecInChild: Boolean =
+ getConf(COLLAPSE_WINDOW_WITH_EMPTY_ORDER_SPEC_IN_CHILD)
+
def minBatchesToRetain: Int = getConf(MIN_BATCHES_TO_RETAIN)
def maxVersionsToDeletePerMaintenance: Int = getConf(MAX_VERSIONS_TO_DELETE_PER_MAINTENANCE)
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/CollapseWindowSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/CollapseWindowSuite.scala
index 515203da7caf..fb2f0957d74d 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/CollapseWindowSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/CollapseWindowSuite.scala
@@ -19,9 +19,15 @@ 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, RangeFrame, 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
+import org.apache.spark.sql.internal.SQLConf
class CollapseWindowSuite extends PlanTest {
object Optimize extends RuleExecutor[LogicalPlan] {
@@ -168,4 +174,248 @@ class CollapseWindowSuite extends PlanTest {
comparePlans(optimized, correctAnswer)
}
+
+ test("collapse windows when one has an empty order spec " +
+ "(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 has a RANGE whole-partition frame") {
+ // `RANGE BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING` covers the whole partition just
+ // like `ROWS`, so it also collapses.
+ 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(RangeFrame, 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 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. This direction can disable InferWindowGroupLimit, so it is
+ // gated by `spark.sql.optimizer.collapseWindowWithEmptyOrderSpecInChild`.
+ 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 = withSQLConf(
+ SQLConf.COLLAPSE_WINDOW_WITH_EMPTY_ORDER_SPEC_IN_CHILD.key -> "true") {
+ Optimize.execute(analyzed)
+ }
+ assert(analyzed.output === optimized.output)
+
+ val correctAnswer = testRelation
+ .window(Seq(cnt, rk), partitionSpec1, orderSpec1)
+
+ comparePlans(optimized, correctAnswer)
+ }
+
+ test("don't collapse the inner empty-order window by default") {
+ // Merging an empty-order child into an ordered parent can disable InferWindowGroupLimit for
+ // top-k queries, so it is off by default.
+ 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 optimized = Optimize.execute(query.analyze)
+ val correctAnswer = query.analyze
+
+ 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). The empty-order
+ // window is the inner window here, so the config must be enabled.
+ 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 = withSQLConf(
+ SQLConf.COLLAPSE_WINDOW_WITH_EMPTY_ORDER_SPEC_IN_CHILD.key -> "true") {
+ 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)
+ }
+
+ test("collapse windows with a Project between them, empty order spec as parent") {
+ // The empty-order window is the parent here, so this merges by default without the config.
+ 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)
+ .select($"a", $"b", $"c", $"rk")
+ .window(Seq(cnt), partitionSpec1, Nil)
+ .select($"a", $"b", $"c", $"rk", $"cnt")
+
+ val optimized = Optimize.execute(query.analyze)
+ assert(query.analyze.output === optimized.output)
+
+ val correctAnswer = testRelation
+ .window(Seq(rk, cnt), partitionSpec1, orderSpec1)
+ .select(a, b, c, $"rk", $"cnt")
+ .analyze
+
+ comparePlans(optimized, correctAnswer)
+ }
}
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/InferWindowGroupLimitSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/InferWindowGroupLimitSuite.scala
index 5aa7a27f65fb..5e0f978300cc 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/InferWindowGroupLimitSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/InferWindowGroupLimitSuite.scala
@@ -20,9 +20,10 @@ package org.apache.spark.sql.catalyst.optimizer
import org.apache.spark.sql.Row
import org.apache.spark.sql.catalyst.dsl.expressions._
import org.apache.spark.sql.catalyst.dsl.plans._
-import org.apache.spark.sql.catalyst.expressions.{CurrentRow, DenseRank, Literal, NthValue, NTile, PercentRank, Rank, RowFrame, RowNumber, SpecifiedWindowFrame, UnboundedPreceding}
+import org.apache.spark.sql.catalyst.expressions.{CurrentRow, DenseRank, Literal, NthValue, NTile, PercentRank, Rank, RowFrame, RowNumber, SpecifiedWindowFrame, UnboundedFollowing, UnboundedPreceding}
+import org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression, Complete, Count}
import org.apache.spark.sql.catalyst.plans.PlanTest
-import org.apache.spark.sql.catalyst.plans.logical.{LocalRelation, LogicalPlan}
+import org.apache.spark.sql.catalyst.plans.logical.{LocalRelation, LogicalPlan, WindowGroupLimit}
import org.apache.spark.sql.catalyst.rules.RuleExecutor
import org.apache.spark.sql.internal.SQLConf
@@ -46,6 +47,17 @@ class InferWindowGroupLimitSuite extends PlanTest {
LimitPushDownThroughWindow) :: Nil
}
+ private object WithCollapseWindow extends RuleExecutor[LogicalPlan] {
+ val batches =
+ Batch("Insert WindowGroupLimit with CollapseWindow", FixedPoint(10),
+ CollapseWindow,
+ CollapseProject,
+ RemoveNoopOperators,
+ PushDownPredicates,
+ InferWindowGroupLimit,
+ LimitPushDownThroughWindow) :: Nil
+ }
+
private val testRelation = LocalRelation.fromExternalRows(
Seq("a".attr.int, "b".attr.int, "c".attr.int),
1.to(6).map(_ => Row(1, 2, 3)))
@@ -354,4 +366,32 @@ class InferWindowGroupLimitSuite extends PlanTest {
Optimize.execute(originalQuery.analyze),
WithoutOptimize.execute(originalQuery.analyze))
}
+
+ test("SPARK-58757: collapseWindowWithEmptyOrderSpecInChild keeps WindowGroupLimit by default") {
+ val cnt = windowExpr(
+ AggregateExpression(Count(c), Complete, isDistinct = false, None),
+ windowSpec(a :: Nil, Nil,
+ SpecifiedWindowFrame(RowFrame, UnboundedPreceding, UnboundedFollowing))).as("cnt")
+ val rn = windowExpr(
+ RowNumber(),
+ windowSpec(a :: Nil, c.desc :: Nil,
+ SpecifiedWindowFrame(RowFrame, UnboundedPreceding, CurrentRow))).as("rn")
+
+ def analyzed: LogicalPlan = testRelation
+ .window(Seq(cnt), a :: Nil, Nil)
+ .window(Seq(rn), a :: Nil, c.desc :: Nil)
+ .where($"rn" <= 2)
+ .analyze
+
+ // By default the config is off, so the empty-order child is not collapsed into the ordered
+ // parent and the WindowGroupLimit is preserved.
+ val defaultPlan = WithCollapseWindow.execute(analyzed)
+ assert(defaultPlan.collect { case _: WindowGroupLimit => 1 }.size == 1)
+
+ // With the config on, the empty-order child is collapsed, disabling the WindowGroupLimit.
+ withSQLConf(SQLConf.COLLAPSE_WINDOW_WITH_EMPTY_ORDER_SPEC_IN_CHILD.key -> "true") {
+ val enabledPlan = WithCollapseWindow.execute(analyzed)
+ assert(enabledPlan.collect { case _: WindowGroupLimit => 1 }.isEmpty)
+ }
+ }
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameWindowFunctionsSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameWindowFunctionsSuite.scala
index 136dc57fdfcf..3530ccbb6ecf 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameWindowFunctionsSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameWindowFunctionsSuite.scala
@@ -79,6 +79,28 @@ class DataFrameWindowFunctionsSuite extends SharedSparkSession
parameters = Map("wf_name" -> "row_number", "wf_expr" -> "row_number()"))
}
+ test("SPARK-58757: collapse window with an empty order spec into an ordered sibling") {
+ val df = Seq(
+ (0, 0), (0, 2), (0, 4),
+ (1, 1), (1, 3), (1, 5)).toDF("k", "v")
+ val ordered = Window.partitionBy("k").orderBy("v")
+ val unordered = Window.partitionBy("k")
+ val res = df.select(
+ $"k",
+ $"v",
+ row_number().over(ordered).as("rn"),
+ collect_list($"v").over(unordered).as("vs"),
+ first($"v").over(unordered).as("f"))
+ assert(res.queryExecution.optimizedPlan.collect { case w: LogicalWindow => w }.size === 1)
+ checkAnswer(res, Seq(
+ Row(0, 0, 1, Array(0, 2, 4), 0),
+ Row(0, 2, 2, Array(0, 2, 4), 0),
+ Row(0, 4, 3, Array(0, 2, 4), 0),
+ Row(1, 1, 1, Array(1, 3, 5), 1),
+ Row(1, 3, 2, Array(1, 3, 5), 1),
+ Row(1, 5, 3, Array(1, 3, 5), 1)))
+ }
+
test("corr, covar_pop, stddev_pop functions in specific window") {
withSQLConf(SQLConf.LEGACY_STATISTICAL_AGGREGATE.key -> "true",
SQLConf.ANSI_ENABLED.key -> "false") {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org