This is an automated email from the ASF dual-hosted git repository.
zzcclp pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gluten.git
The following commit(s) were added to refs/heads/main by this push:
new 91f1f43845 [GLUTEN-13234][CH] Avoid aggregate topk for identical
window keys (#13235)
91f1f43845 is described below
commit 91f1f438459bebbc477f7db500e18e6f151b49db
Author: lgbo <[email protected]>
AuthorDate: Mon Oct 12 14:14:14 2026 +0800
[GLUTEN-13234][CH] Avoid aggregate topk for identical window keys (#13235)
CH window top-k aggregate optimization can fail when the window partition
keys and order keys are exactly the same expressions.
## What changes are proposed in this pull request?
Fixes #13234.
For queries like:
```sql
ROW_NUMBER() OVER (
PARTITION BY l_returnflag
ORDER BY l_returnflag DESC
)
```
the partition key and order key are identical. This shape is not suitable
for the CH window-to-aggregate-topk optimization. When the aggregate top-k path
is forced, native execution may cast the generated `rn` column from a wrong
source column and fail with errors such as:
```text
Cannot convert NULL value to non-Nullable type: while converting source
column `rn#...` to destination column `rn#...`
```
or with a string-to-Int32 cast error when the misplaced source column
contains a non-null partition key value.
This patch updates the CH window-to-aggregate-topk rule to skip the
aggregate top-k rewrite when the partition keys and order keys are exactly the
same expressions. The query will keep the regular window/filter execution path
instead.
A regression test is added to `GlutenClickHouseTPCHSaltNullParquetSuite`.
The test forces the aggregate-topk optimization to be considered and verifies
that this same-key case does not produce `CHAggregateGroupLimitExecTransformer`.
## How was this patch tested?
- Added `GlutenClickHouseTPCHSaltNullParquetSuite` coverage for
`row_number` with identical partition/order keys.
- Locally reproduced the failure by forcing:
- `spark.sql.optimizer.windowGroupLimitThreshold=-1`
-
`spark.gluten.sql.columnar.backend.ch.runtime_settings.enable_window_group_limit_to_aggregate=true`
-
`spark.gluten.sql.columnar.backend.ch.runtime_settings.window.aggregate_topk_high_cardinality_threshold=2.0`
- Verified after the fix that the plan no longer contains
`CHAggregateGroupLimitExecTransformer` for the same-key case and the query
executes successfully.
- Locally built with:
```bash
./build/mvn -pl backends-clickhouse -am \
-DskipTests \
-Pbackends-clickhouse \
-Pspark-3.5 \
-Pscala-2.13 \
-Pceleborn \
-Pdelta \
-Djava.io.tmpdir=/tmp \
package
```
## Was this patch authored or co-authored using generative AI tooling?
Generated-by: DeepSeek Harness gpt-5.5
---
.../extension/ConvertWindowToAggregate.scala | 16 +++++++--
.../GlutenClickHouseTPCHSaltNullParquetSuite.scala | 39 ++++++++++++++++++++++
2 files changed, 53 insertions(+), 2 deletions(-)
diff --git
a/backends-clickhouse/src/main/scala/org/apache/gluten/extension/ConvertWindowToAggregate.scala
b/backends-clickhouse/src/main/scala/org/apache/gluten/extension/ConvertWindowToAggregate.scala
index d41778345c..3b7a2a828c 100644
---
a/backends-clickhouse/src/main/scala/org/apache/gluten/extension/ConvertWindowToAggregate.scala
+++
b/backends-clickhouse/src/main/scala/org/apache/gluten/extension/ConvertWindowToAggregate.scala
@@ -54,12 +54,14 @@ case class ConverRowNumbertWindowToAggregateRule(spark:
SparkSession)
if (
!isSupportedWindowFunction(windowExpressions) || !isTopKLimitFilter(
condition,
- windowExpressions(0))
+ windowExpressions(0)) ||
hasSamePartitionAndOrderKeys(partitionSpec, orderSpec)
) {
logDebug(
s"xxx Not Supported case for converting window to aggregate. is
topk limit: " +
s"${isTopKLimitFilter(condition, windowExpressions(0))}. is
supported window " +
- s"function: ${isSupportedWindowFunction(windowExpressions)}")
+ s"function: ${isSupportedWindowFunction(windowExpressions)}. " +
+ s"has same partition and order keys: " +
+ s"${hasSamePartitionAndOrderKeys(partitionSpec, orderSpec)}")
filter
} else {
val limit = getLimit(condition.asInstanceOf[BinaryComparison])
@@ -113,6 +115,16 @@ case class ConverRowNumbertWindowToAggregateRule(spark:
SparkSession)
}
}
+ private def hasSamePartitionAndOrderKeys(
+ partitionSpec: Seq[Expression],
+ orderSpec: Seq[SortOrder]): Boolean = {
+ partitionSpec.nonEmpty && partitionSpec.length == orderSpec.length &&
+ partitionSpec.zip(orderSpec).forall {
+ case (partitionExpr, sortOrder) =>
+ partitionExpr.semanticEquals(sortOrder.child)
+ }
+ }
+
// If the filter condition is a constant limit condition, return the limit
value.
def isTopKLimitFilter(condition: Expression, windowExpression:
NamedExpression): Boolean = {
def isWindowFunctionResult(
diff --git
a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/tpch/GlutenClickHouseTPCHSaltNullParquetSuite.scala
b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/tpch/GlutenClickHouseTPCHSaltNullParquetSuite.scala
index 6ba47dc901..06ba443c26 100644
---
a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/tpch/GlutenClickHouseTPCHSaltNullParquetSuite.scala
+++
b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/tpch/GlutenClickHouseTPCHSaltNullParquetSuite.scala
@@ -554,6 +554,45 @@ class GlutenClickHouseTPCHSaltNullParquetSuite
compareResultsAgainstVanillaSpark(sql, true, { _ => })
}
+ test("window row_number with same partition and order keys should not use
aggregate topk") {
+ val sql =
+ """
+ |WITH ranked AS (
+ | SELECT
+ | l_orderkey,
+ | l_returnflag,
+ | ROW_NUMBER() OVER (
+ | PARTITION BY l_returnflag
+ | ORDER BY l_returnflag DESC
+ | ) AS rn
+ | FROM lineitem
+ | WHERE l_orderkey > 1000
+ |)
+ |
+ |SELECT
+ | l_returnflag,
+ | l_orderkey,
+ | rn
+ |FROM ranked
+ |WHERE rn <= 3
+ |ORDER BY l_returnflag DESC
+ |""".stripMargin
+
+ withSQLConf(
+ ("spark.sql.optimizer.windowGroupLimitThreshold", "-1"),
+ (CHConfig.runtimeSettings("enable_window_group_limit_to_aggregate"),
"true"),
+
(CHConfig.runtimeSettings("window.aggregate_topk_high_cardinality_threshold"),
"2.0")
+ ) {
+ runQueryAndCompare(sql, compareResult = false) {
+ df =>
+ val aggregateTopK =
collectWithSubqueries(df.queryExecution.executedPlan) {
+ case e: CHAggregateGroupLimitExecTransformer => e
+ }
+ assert(aggregateTopK.isEmpty)
+ }
+ }
+ }
+
test("window sum 1") {
val sql =
"""
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
