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 cf2bd18de7 [VL] Remap generated build key aliases in the driver-side
hash table build (#13332)
cf2bd18de7 is described below
commit cf2bd18de72644196a90ff77d64e7809f1d74f9d
Author: YangJie <[email protected]>
AuthorDate: Tue Oct 20 02:59:41 2026 -0400
[VL] Remap generated build key aliases in the driver-side hash table build
(#13332)
---
.../sql/execution/ColumnarBuildSideRelation.scala | 4 +-
.../unsafe/UnsafeColumnarBuildSideRelation.scala | 4 +-
.../gluten/execution/VeloxHashJoinSuite.scala | 61 ++++++++++++++++------
3 files changed, 50 insertions(+), 19 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 55616fc8a9..37203fdc9f 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
@@ -313,10 +313,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 e0f44d350f..47c2b13175 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
@@ -274,10 +274,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 0b86b036b6..25e5df6e98 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
@@ -531,6 +531,23 @@ class VeloxHashJoinSuite extends
VeloxWholeStageTransformerSuite {
}
}
+ // The LATERAL VIEW output `key` reaches the build side under a generated
alias.
+ private val generatedBuildKeyAliasQuery =
+ """
+ |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
+
test("Broadcast build once with generated build key alias") {
withSQLConf(
("spark.sql.autoBroadcastJoinThreshold", "10MB"),
@@ -538,23 +555,7 @@ class VeloxHashJoinSuite extends
VeloxWholeStageTransformerSuite {
(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) {
+ runQueryAndCompare(generatedBuildKeyAliasQuery) {
df =>
val plan = df.queryExecution.executedPlan
val broadcastJoins = plan.collect { case bhj:
BroadcastHashJoinExecTransformer => bhj }
@@ -563,6 +564,32 @@ class VeloxHashJoinSuite extends
VeloxWholeStageTransformerSuite {
}
}
+ Seq(false, true).foreach {
+ offHeap =>
+ test(s"Driver-side build with generated build key alias, off-heap
relation: $offHeap") {
+ withSQLConf(
+ ("spark.sql.autoBroadcastJoinThreshold", "10MB"),
+ ("spark.sql.adaptive.enabled", "false"),
+ (VeloxConfig.VELOX_DRIVER_SIDE_BROADCAST_HASH_TABLE_BUILD.key,
"true"),
+ (VeloxConfig.VELOX_BROADCAST_BUILD_RELATION_USE_OFFHEAP.key,
offHeap.toString)
+ ) {
+ createTPCHNotNullTables()
+ runQueryAndCompare(generatedBuildKeyAliasQuery) {
+ df =>
+ val relations = collectBroadcastRelations(df)
+ assert(relations.nonEmpty, "Should use a columnar broadcast")
+ // A failed driver-side build falls back to the executor-side
relation.
+ relations.foreach {
+ relation =>
+ assert(
+
relation.isInstanceOf[SerializedHashTableBroadcastRelation],
+ s"Driver-side build fell back, got
${relation.getClass.getName}")
+ }
+ }
+ }
+ }
+ }
+
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]
