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 <[email protected]>
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 <[email protected]>
Signed-off-by: Xiduo You <[email protected]>
---
.../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: [email protected]
For additional commands, e-mail: [email protected]
