This is an automated email from the ASF dual-hosted git repository.
JkSelf 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 8cbacfb03c [GUTEN-12406][VL] Fix broadcast build once with generated
build key alias (#12407)
8cbacfb03c is described below
commit 8cbacfb03cb60d1ece463938aee66df3ce757e25
Author: Wechar Yu <[email protected]>
AuthorDate: Fri Jul 3 12:55:11 2026 +0800
[GUTEN-12406][VL] Fix broadcast build once with generated build key alias
(#12407)
---
.../sql/execution/ColumnarBuildSideRelation.scala | 4 ++-
.../unsafe/UnsafeColumnarBuildSideRelation.scala | 4 ++-
.../gluten/execution/VeloxHashJoinSuite.scala | 32 ++++++++++++++++++++++
3 files changed, 38 insertions(+), 2 deletions(-)
diff --git
a/backends-velox/src/main/scala/org/apache/spark/sql/execution/ColumnarBuildSideRelation.scala
b/backends-velox/src/main/scala/org/apache/spark/sql/execution/ColumnarBuildSideRelation.scala
index 7042f34e93..285cbe7ef6 100644
---
a/backends-velox/src/main/scala/org/apache/spark/sql/execution/ColumnarBuildSideRelation.scala
+++
b/backends-velox/src/main/scala/org/apache/spark/sql/execution/ColumnarBuildSideRelation.scala
@@ -204,10 +204,12 @@ case class ColumnarBuildSideRelation(
)
}
+ val outputByExprId = newOutput.asScala.map(attr => attr.exprId ->
attr).toMap
val joinKeys = keys.asScala.map {
key =>
val attr = ConverterUtils.getAttrFromExpr(key)
- ConverterUtils.genColumnNameWithExprId(attr)
+ val outputAttr = outputByExprId.getOrElse(attr.exprId, attr)
+ ConverterUtils.genColumnNameWithExprId(outputAttr)
}.toArray
val hashJoinBuilder = HashJoinBuilder.create(runtime)
diff --git
a/backends-velox/src/main/scala/org/apache/spark/sql/execution/unsafe/UnsafeColumnarBuildSideRelation.scala
b/backends-velox/src/main/scala/org/apache/spark/sql/execution/unsafe/UnsafeColumnarBuildSideRelation.scala
index fa3dcc6934..57c53540ab 100644
---
a/backends-velox/src/main/scala/org/apache/spark/sql/execution/unsafe/UnsafeColumnarBuildSideRelation.scala
+++
b/backends-velox/src/main/scala/org/apache/spark/sql/execution/unsafe/UnsafeColumnarBuildSideRelation.scala
@@ -174,10 +174,12 @@ class UnsafeColumnarBuildSideRelation(
)
}
+ val outputByExprId = newOutput.asScala.map(attr => attr.exprId ->
attr).toMap
val joinKeys = keys.asScala.map {
key =>
val attr = ConverterUtils.getAttrFromExpr(key)
- ConverterUtils.genColumnNameWithExprId(attr)
+ val outputAttr = outputByExprId.getOrElse(attr.exprId, attr)
+ ConverterUtils.genColumnNameWithExprId(outputAttr)
}.toArray
val hashJoinBuilder = HashJoinBuilder.create(runtime)
diff --git
a/backends-velox/src/test/scala/org/apache/gluten/execution/VeloxHashJoinSuite.scala
b/backends-velox/src/test/scala/org/apache/gluten/execution/VeloxHashJoinSuite.scala
index c80950aa1f..518aad3427 100644
---
a/backends-velox/src/test/scala/org/apache/gluten/execution/VeloxHashJoinSuite.scala
+++
b/backends-velox/src/test/scala/org/apache/gluten/execution/VeloxHashJoinSuite.scala
@@ -357,6 +357,38 @@ class VeloxHashJoinSuite extends
VeloxWholeStageTransformerSuite {
}
}
+ test("Broadcast build once with generated build key alias") {
+ withSQLConf(
+ ("spark.sql.autoBroadcastJoinThreshold", "10MB"),
+ ("spark.sql.adaptive.enabled", "false"),
+ (VeloxConfig.VELOX_BROADCAST_BUILD_HASHTABLE_ONCE_PER_EXECUTOR.key,
"true")
+ ) {
+ createTPCHNotNullTables()
+ val query =
+ """
+ |SELECT /*+ BROADCAST(r) */ l.l_orderkey, l.l_partkey, r.key
+ |FROM lineitem l
+ |LEFT JOIN (
+ | SELECT p_partkey, key
+ | FROM (
+ | SELECT p_partkey, concat('{"Key":"', CAST(p_partkey AS STRING),
'"}') AS json_field
+ | FROM part
+ | ) p
+ | LATERAL VIEW json_tuple(json_field, 'Key') b AS key
+ |) r
+ |ON l.l_partkey = r.p_partkey
+ | AND CAST(l.l_partkey AS STRING) = r.key
+ |""".stripMargin
+
+ runQueryAndCompare(query) {
+ df =>
+ val plan = df.queryExecution.executedPlan
+ val broadcastJoins = plan.collect { case bhj:
BroadcastHashJoinExecTransformer => bhj }
+ assert(broadcastJoins.nonEmpty, "Should use broadcast hash join")
+ }
+ }
+ }
+
test("Reuse broadcast exchange with different hash table") {
withSQLConf(
("spark.sql.adaptive.enabled", "false")
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
