This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 88b4c942d90c Revert "[SPARK-58404][SQL] Support bypassing partial
WindowGroupLimit"
88b4c942d90c is described below
commit 88b4c942d90ccb76dedd2e024564ec9c4419530e
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Tue Aug 4 07:09:49 2026 -0700
Revert "[SPARK-58404][SQL] Support bypassing partial WindowGroupLimit"
This reverts commit 8af2f3d6ccd8519bf393395bd6d29b89f7e91c5d.
---
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: [email protected]
For additional commands, e-mail: [email protected]
