You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by do...@apache.org on 2026/08/04 14:09:45 UTC
(spark) branch master updated: Revert "[SPARK-58404][SQL] Support bypassing partial WindowGroupLimit"
This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun 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 c65076ba7b06 Revert "[SPARK-58404][SQL] Support bypassing partial WindowGroupLimit"
c65076ba7b06 is described below
commit c65076ba7b06dff2019ed6bf5b497e697578d03a
Author: Dongjoon Hyun <do...@apache.org>
AuthorDate: Tue Aug 4 07:09:29 2026 -0700
Revert "[SPARK-58404][SQL] Support bypassing partial WindowGroupLimit"
This reverts commit 91c03e36f0f4135759bfe6a823e911f5446b34df.
---
docs/sql-performance-tuning.md | 25 ---------
.../org/apache/spark/sql/internal/SQLConf.scala | 16 ------
.../spark/sql/execution/SparkStrategies.scala | 21 +------
.../spark/sql/DataFrameWindowFunctionsSuite.scala | 64 +---------------------
4 files changed, 4 insertions(+), 122 deletions(-)
diff --git a/docs/sql-performance-tuning.md b/docs/sql-performance-tuning.md
index 6acb36413d93..3ba3a6c749ca 100644
--- a/docs/sql-performance-tuning.md
+++ b/docs/sql-performance-tuning.md
@@ -168,31 +168,6 @@ SELECT /*+ REBALANCE_BY_SIZE('128m', c) */ * FROM t;
For more details please refer to the documentation of [Partitioning Hints](sql-ref-syntax-qry-select-hints.html#partitioning-hints).
-## Tuning Window Functions
-
-Top-k window queries (such as filtering on <code>row_number</code>, <code>rank</code> or
-<code>dense_rank</code>) plan a window group limit on both sides of the shuffle: a partial one
-before the shuffle to reduce the data feeding it, and a final one after. The following configuration
-tunes that behavior.
-
-<table class="spark-config">
- <thead><tr><th>Property Name</th><th>Default</th><th>Meaning</th><th>Since Version</th></tr></thead>
- <tr>
- <td><code>spark.sql.execution.bypassPartialWindowGroupLimit</code></td>
- <td>false</td>
- <td>
- When true, skips the pre-shuffle partial window group limit for partitioned top-k window
- queries and runs only a single window group limit after the shuffle. Bypassing the partial
- window group limit can improve performance when the pre-shuffle reduction ratio is low. The
- bypass only applies to windows with a non-empty partition spec; for an unpartitioned window
- the partial pass is always kept, since for a multi-partition input the shuffle funnels the
- whole input into a single reducer. When false, a partial window group limit runs before the
- shuffle and a final one runs after it.
- </td>
- <td>4.3.0</td>
- </tr>
-</table>
-
## Leveraging Statistics
Apache Spark's ability to choose the best execution plan among many possible options is determined in part by its estimates of how many rows will be output by every node in the execution plan (read, filter, join, etc.). Those estimates in turn are based on statistics that are made available to Spark in one of several ways:
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 d060d5f7a7f9..8a0d88346b63 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
@@ -4689,20 +4689,6 @@ object SQLConf {
"The threshold of window group limit must be -1, 0 or positive integer.")
.createWithDefault(1000)
- val BYPASS_PARTIAL_WINDOW_GROUP_LIMIT =
- buildConf("spark.sql.execution.bypassPartialWindowGroupLimit")
- .doc("When true, skips the pre-shuffle partial WindowGroupLimit and runs only a single " +
- "WindowGroupLimit after the shuffle. Bypassing the partial window group limit can " +
- "improve performance when the pre-shuffle reduction ratio is low. This only applies to " +
- "windows with a non-empty partition spec; an unpartitioned window always keeps the " +
- "partial pass, since for a multi-partition input the shuffle funnels the whole input " +
- "into a single reducer. When false (default), a partial WindowGroupLimit runs before " +
- "the shuffle and a final one runs after it.")
- .version("4.3.0")
- .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
- .booleanConf
- .createWithDefault(false)
-
val WINDOW_SEGMENT_TREE_ENABLED =
buildConf("spark.sql.window.segmentTree.enabled")
.withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
@@ -9136,8 +9122,6 @@ class SQLConf extends Serializable with Logging with SqlApiConf {
def windowGroupLimitThreshold: Int = getConf(WINDOW_GROUP_LIMIT_THRESHOLD)
- def bypassPartialWindowGroupLimit: Boolean = getConf(BYPASS_PARTIAL_WINDOW_GROUP_LIMIT)
-
def windowSegmentTreeEnabled: Boolean = getConf(WINDOW_SEGMENT_TREE_ENABLED)
def windowSegmentTreeMinPartitionRows: Int = getConf(WINDOW_SEGMENT_TREE_MIN_PARTITION_ROWS)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala
index 1d0917266ea8..996ecac61b72 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala
@@ -802,25 +802,10 @@ abstract class SparkStrategies extends QueryPlanner[SparkPlan] {
object WindowGroupLimit extends Strategy {
def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
case logical.WindowGroupLimit(partitionSpec, orderSpec, rankLikeFunction, limit, child) =>
- // When the partial window group limit is bypassed, skip the pre-shuffle partial
- // WindowGroupLimit and run only a single WindowGroupLimit after the shuffle. This can
- // improve performance when the pre-shuffle reduction ratio is low.
- //
- // The bypass is only gated on a non-empty partitionSpec. With an empty partitionSpec the
- // final WindowGroupLimit requires AllTuples, so for a multi-partition input the shuffle
- // funnels everything into a single reducer; the partial pass we would drop bounds each
- // input partition to `limit` rank groups before that shuffle, so keeping it is a sensible
- // default (an already-single-partition child needs no shuffle, so the partial cannot reduce
- // the shuffle input there and only adds another pass). This mirrors the grouping-keys
- // carve-out in bypassPartialAggregation (see AggUtils.planAggregateWithoutDistinct).
- val finalChild = if (conf.bypassPartialWindowGroupLimit && partitionSpec.nonEmpty) {
- planLater(child)
- } else {
- execution.window.WindowGroupLimitExec(partitionSpec,
- orderSpec, rankLikeFunction, limit, execution.window.Partial, planLater(child))
- }
+ val partialWindowGroupLimit = execution.window.WindowGroupLimitExec(partitionSpec,
+ orderSpec, rankLikeFunction, limit, execution.window.Partial, planLater(child))
val finalWindowGroupLimit = execution.window.WindowGroupLimitExec(partitionSpec, orderSpec,
- rankLikeFunction, limit, execution.window.Final, finalChild)
+ rankLikeFunction, limit, execution.window.Final, partialWindowGroupLimit)
finalWindowGroupLimit :: Nil
case _ => Nil
}
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 cf73633e74ed..c1d1f13915a6 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
@@ -27,7 +27,7 @@ import org.apache.spark.sql.catalyst.plans.physical.HashPartitioning
import org.apache.spark.sql.catalyst.trees.UnaryLike
import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
import org.apache.spark.sql.execution.exchange.{ENSURE_REQUIREMENTS, Exchange, ShuffleExchangeExec}
-import org.apache.spark.sql.execution.window.{WindowExec, WindowGroupLimitExec}
+import org.apache.spark.sql.execution.window.WindowExec
import org.apache.spark.sql.expressions.{Aggregator, MutableAggregationBuffer, UserDefinedAggregateFunction, Window}
import org.apache.spark.sql.functions._
import org.apache.spark.sql.internal.SQLConf
@@ -1682,68 +1682,6 @@ class DataFrameWindowFunctionsSuite extends SharedSparkSession
}
}
- test("SPARK-58404: bypass partial WindowGroupLimit") {
- val df = Seq(
- ("a", 0, "c"),
- ("a", 1, "x"),
- ("a", 2, "y"),
- ("b", 1, "h"),
- ("b", 1, "n"),
- ("c", 1, "z"),
- ("c", 2, "a")).toDF("key", "value", "order")
-
- def countWindowGroupLimits(df: DataFrame): Int = {
- collect(df.queryExecution.executedPlan) {
- case w: WindowGroupLimitExec => w
- }.size
- }
-
- val rankLikeFunctions = Seq(row_number(), rank(), dense_rank())
-
- // Partitioned window: a shuffle is required, so without the bypass both the partial and the
- // final WindowGroupLimit are present; with the bypass only the final one remains.
- val window = Window.partitionBy($"key").orderBy($"order")
- val expected = Seq(
- Row("a", 0, "c", 1),
- Row("b", 1, "h", 1),
- Row("c", 2, "a", 1))
-
- Seq(true, false).foreach { bypass =>
- withSQLConf(
- SQLConf.BYPASS_PARTIAL_WINDOW_GROUP_LIMIT.key -> bypass.toString,
- SQLConf.WINDOW_GROUP_LIMIT_THRESHOLD.key -> "100") {
- rankLikeFunctions.foreach { rankLikeFunction =>
- val result = df.withColumn("rn", rankLikeFunction.over(window)).where($"rn" === 1)
- checkAnswer(result, expected)
- assert(countWindowGroupLimits(result) === (if (bypass) 1 else 2))
- }
- }
- }
-
- // Unpartitioned window: the final WindowGroupLimit requires AllTuples, so the shuffle funnels
- // the whole input into a single reducer. The partial pass bounds each input partition to
- // `limit` rank groups before that shuffle, so the bypass is gated off for an empty partition
- // spec and both the partial and the final WindowGroupLimit remain even when the flag is on.
- // Note row_number() is excluded: with an empty partition spec and a limit below
- // topKSortFallbackThreshold, InferWindowGroupLimit rewrites it to a Limit + Sort (top-n)
- // rather than a WindowGroupLimit, so no WindowGroupLimit node is produced.
- val unpartitionedWindow = Window.orderBy($"order")
- val unpartitionedExpected = Seq(Row("c", 2, "a", 1))
-
- Seq(true, false).foreach { bypass =>
- withSQLConf(
- SQLConf.BYPASS_PARTIAL_WINDOW_GROUP_LIMIT.key -> bypass.toString,
- SQLConf.WINDOW_GROUP_LIMIT_THRESHOLD.key -> "100") {
- Seq(rank(), dense_rank()).foreach { rankLikeFunction =>
- val result =
- df.withColumn("rn", rankLikeFunction.over(unpartitionedWindow)).where($"rn" === 1)
- checkAnswer(result, unpartitionedExpected)
- assert(countWindowGroupLimits(result) === 2)
- }
- }
- }
- }
-
test("SPARK-42525: collapse two adjacent windows with the same partition/order in subquery") {
withTempView("t1") {
Seq((1, 1), (2, 2)).toDF("a", "b").createOrReplaceTempView("t1")
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org