This is an automated email from the ASF dual-hosted git repository.
zhouyuan 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 3abf4c4df5 [GLUTEN-12566][CORE] Propagate keyGroupedPartitioning in
BatchScanExecTransformer (#12567)
3abf4c4df5 is described below
commit 3abf4c4df52b648c1669de11f249ba53e5dd91b6
Author: Yuming Wang <[email protected]>
AuthorDate: Thu Aug 13 00:15:21 2026 +0800
[GLUTEN-12566][CORE] Propagate keyGroupedPartitioning in
BatchScanExecTransformer (#12567)
* [GLUTEN-12566][CORE] Fix BatchScanExecTransformer to propagate
keyGroupedPartitioning
ScanTransformerFactory.createBatchScanTransformer() does not pass
keyGroupedPartitioning from the original BatchScanExec to the
BatchScanExecTransformer, leaving it as None even when the source
BatchScanExec has SPJ (Storage Partitioned Join) partitioning. This
causes Gluten's BatchScanExecTransformer to report UnknownPartitioning
instead of KeyGroupedPartitioning, defeating SPJ and inserting
redundant shuffles.
IcebergScanTransformer and PaimonScanTransformer already pass
keyGroupedPartitioning correctly; this fixes the same gap for regular
file scans (Parquet, ORC, etc.) created via ScanTransformerFactory.
Additionally fix two related issues in BatchScanExecTransformer that
were masked by keyGroupedPartitioning always being None:
1. doCanonicalize() does not normalize keyGroupedPartitioning. When
two semantically identical scans have different expression IDs in
their keyGroupedPartitioning expressions, the canonicalized plans
are not equal, preventing AQE exchange reuse.
2. hashCode() does not include keyGroupedPartitioning, violating the
equals/hashCode contract (equals compares it via spjParams, but
hashCode omits it).
Closes #12566.
* format
* Update BatchScanExecTransformer.scala
---
.../org/apache/gluten/execution/BatchScanExecTransformer.scala | 9 +++++++--
.../org/apache/gluten/execution/ScanTransformerFactory.scala | 2 ++
2 files changed, 9 insertions(+), 2 deletions(-)
diff --git
a/gluten-substrait/src/main/scala/org/apache/gluten/execution/BatchScanExecTransformer.scala
b/gluten-substrait/src/main/scala/org/apache/gluten/execution/BatchScanExecTransformer.scala
index 31fe9898e7..6d2854f30b 100644
---
a/gluten-substrait/src/main/scala/org/apache/gluten/execution/BatchScanExecTransformer.scala
+++
b/gluten-substrait/src/main/scala/org/apache/gluten/execution/BatchScanExecTransformer.scala
@@ -70,6 +70,8 @@ case class BatchScanExecTransformer(
runtimeFilters = QueryPlan.normalizePredicates(
runtimeFilters.filterNot(_ ==
DynamicPruningExpression(Literal.TrueLiteral)),
output),
+ keyGroupedPartitioning = keyGroupedPartitioning.map(
+ _.map(QueryPlan.normalizeExpressions(_, output))),
pushDownFilters = pushDownFilters.map(QueryPlan.normalizePredicates(_,
output))
)
}
@@ -216,12 +218,15 @@ abstract class BatchScanExecTransformerBase(
override def equals(other: Any): Boolean = other match {
case other: BatchScanExecTransformerBase =>
- this.pushDownFilters == other.pushDownFilters && super.equals(other)
+ this.keyGroupedPartitioning == other.keyGroupedPartitioning &&
+ this.pushDownFilters == other.pushDownFilters &&
+ super.equals(other)
case _ =>
false
}
- override def hashCode(): Int = Objects.hashCode(batch, runtimeFilters,
pushDownFilters)
+ override def hashCode(): Int =
+ Objects.hashCode(batch, runtimeFilters, keyGroupedPartitioning,
pushDownFilters)
/** Return a copy of this scan with a new output schema. */
def withOutput(newOutput: Seq[AttributeReference]):
BatchScanExecTransformerBase
diff --git
a/gluten-substrait/src/main/scala/org/apache/gluten/execution/ScanTransformerFactory.scala
b/gluten-substrait/src/main/scala/org/apache/gluten/execution/ScanTransformerFactory.scala
index 711f8c6908..b2839aa43b 100644
---
a/gluten-substrait/src/main/scala/org/apache/gluten/execution/ScanTransformerFactory.scala
+++
b/gluten-substrait/src/main/scala/org/apache/gluten/execution/ScanTransformerFactory.scala
@@ -44,6 +44,8 @@ object ScanTransformerFactory {
batchScanExec.output,
batchScanExec.scan,
batchScanExec.runtimeFilters,
+ keyGroupedPartitioning =
+ SparkShimLoader.getSparkShims.getKeyGroupedPartitioning(batchScanExec),
table =
SparkShimLoader.getSparkShims.getBatchScanExecTable(batchScanExec)
)
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
