This is an automated email from the ASF dual-hosted git repository.
marin-ma 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 6dee1fa824 [VL] Fix empty cuDF broadcast join results by streaming the
build side (#12812)
6dee1fa824 is described below
commit 6dee1fa8244387eee3e2258885d72390d0f0be68
Author: Reema <[email protected]>
AuthorDate: Wed Aug 19 10:16:41 2026 +0200
[VL] Fix empty cuDF broadcast join results by streaming the build side
(#12812)
With cuDF enabled, broadcast hash joins run as CudfHashJoin, which builds
its own GPU hash table from the build-side value stream and cannot consume
the prebuilt CPU OpaqueHashTable created by the build-once-per-executor
optimization. Feeding it Iterator.empty silently produced empty join
results.
Skip the prebuilt-table path and stream the broadcast relation batches
when cuDF is enabled, the same path shuffle joins use on GPU. Skipping the
CPU cache build also keeps hybrid mode correct: a CPU-fallback join finds
no cached table and builds from the stream as usual.
Adds a CudfTest-tagged regression suite. The tag is excluded in CI, which
has no GPU test runners; run on a GPU host with -DtagsToExclude=None.
---
.github/workflows/velox_backend_ansi.yml | 8 +--
.github/workflows/velox_backend_enhanced.yml | 4 +-
.github/workflows/velox_backend_x86.yml | 12 ++--
backends-velox/pom.xml | 2 +-
.../execution/VeloxBroadcastBuildSideRDD.scala | 10 ++-
.../test/java/org/apache/gluten/tags/CudfTest.java | 29 ++++++++
.../gluten/execution/CudfBroadcastJoinSuite.scala | 79 ++++++++++++++++++++++
7 files changed, 130 insertions(+), 14 deletions(-)
diff --git a/.github/workflows/velox_backend_ansi.yml
b/.github/workflows/velox_backend_ansi.yml
index 5f4bb1d524..2ed44123ce 100644
--- a/.github/workflows/velox_backend_ansi.yml
+++ b/.github/workflows/velox_backend_ansi.yml
@@ -172,7 +172,7 @@ jobs:
java -version
$MVN_CMD clean test -Pspark-4.1 -Pscala-2.13 -Pjava-17
-Pbackends-velox \
-DargLine="-Dspark.test.home=/opt/shims/spark41/spark_home/
-Dspark.gluten.sql.ansiFallback.enabled=false" \
-
-DtagsToExclude=org.apache.spark.tags.ExtendedSQLTest,org.apache.gluten.tags.UDFTest,org.apache.gluten.tags.EnhancedFeaturesTest,org.apache.gluten.tags.SkipTest
\
+
-DtagsToExclude=org.apache.spark.tags.ExtendedSQLTest,org.apache.gluten.tags.UDFTest,org.apache.gluten.tags.EnhancedFeaturesTest,org.apache.gluten.tags.CudfTest,org.apache.gluten.tags.SkipTest
\
2>&1 | tee backends-velox-test-output.log
- name: "Parse test results"
if: always()
@@ -241,7 +241,7 @@ jobs:
$MVN_CMD clean test -Pspark-4.1 -Pscala-2.13 -Pjava-17
-Pbackends-velox -Pspark-ut \
-DwildcardSuites='org.apache.spark.' -Dtest=none
-DfailIfNoTests=false \
-DargLine="-Dspark.test.home=/opt/shims/spark41/spark_home/
-Dspark.gluten.sql.ansiFallback.enabled=false" \
-
-DtagsToExclude=org.apache.spark.tags.ExtendedSQLTest,org.apache.gluten.tags.UDFTest,org.apache.gluten.tags.EnhancedFeaturesTest,org.apache.gluten.tags.SkipTest
\
+
-DtagsToExclude=org.apache.spark.tags.ExtendedSQLTest,org.apache.gluten.tags.UDFTest,org.apache.gluten.tags.EnhancedFeaturesTest,org.apache.gluten.tags.CudfTest,org.apache.gluten.tags.SkipTest
\
2>&1 | tee spark-ut-test-output.log
- name: "Parse test results"
if: always()
@@ -318,7 +318,7 @@ jobs:
java -version
$MVN_CMD clean test -Pspark-4.0 -Pscala-2.13 -Pjava-17
-Pbackends-velox \
-DargLine="-Dspark.test.home=/opt/shims/spark40/spark_home/
-Dspark.gluten.sql.ansiFallback.enabled=false" \
-
-DtagsToExclude=org.apache.spark.tags.ExtendedSQLTest,org.apache.gluten.tags.UDFTest,org.apache.gluten.tags.EnhancedFeaturesTest,org.apache.gluten.tags.SkipTest
\
+
-DtagsToExclude=org.apache.spark.tags.ExtendedSQLTest,org.apache.gluten.tags.UDFTest,org.apache.gluten.tags.EnhancedFeaturesTest,org.apache.gluten.tags.CudfTest,org.apache.gluten.tags.SkipTest
\
2>&1 | tee backends-velox-spark40-test-output.log
- name: "Parse test results"
if: always()
@@ -387,7 +387,7 @@ jobs:
$MVN_CMD clean test -Pspark-4.0 -Pscala-2.13 -Pjava-17
-Pbackends-velox -Pspark-ut \
-DwildcardSuites='org.apache.spark.' -Dtest=none
-DfailIfNoTests=false \
-DargLine="-Dspark.test.home=/opt/shims/spark40/spark_home/
-Dspark.gluten.sql.ansiFallback.enabled=false" \
-
-DtagsToExclude=org.apache.spark.tags.ExtendedSQLTest,org.apache.gluten.tags.UDFTest,org.apache.gluten.tags.EnhancedFeaturesTest,org.apache.gluten.tags.SkipTest
\
+
-DtagsToExclude=org.apache.spark.tags.ExtendedSQLTest,org.apache.gluten.tags.UDFTest,org.apache.gluten.tags.EnhancedFeaturesTest,org.apache.gluten.tags.CudfTest,org.apache.gluten.tags.SkipTest
\
2>&1 | tee spark-ut-spark40-test-output.log
- name: "Parse test results"
if: always()
diff --git a/.github/workflows/velox_backend_enhanced.yml
b/.github/workflows/velox_backend_enhanced.yml
index ac0b1bb18c..e9493a3ee0 100644
--- a/.github/workflows/velox_backend_enhanced.yml
+++ b/.github/workflows/velox_backend_enhanced.yml
@@ -160,7 +160,7 @@ jobs:
export SPARK_HOME=/opt/shims/spark35/spark_home/
ls -l $SPARK_HOME
$MVN_CMD clean test -Pspark-3.5 -Pjava-17 -Pbackends-velox -Piceberg
-Pdelta -Phudi \
-
-DtagsToExclude=org.apache.spark.tags.ExtendedSQLTest,org.apache.spark.tags.SlowHiveTest,org.apache.gluten.tags.UDFTest,org.apache.gluten.tags.SkipTest
\
+
-DtagsToExclude=org.apache.spark.tags.ExtendedSQLTest,org.apache.spark.tags.SlowHiveTest,org.apache.gluten.tags.UDFTest,org.apache.gluten.tags.CudfTest,org.apache.gluten.tags.SkipTest
\
-DargLine="-Dspark.test.home=$SPARK_HOME"
- name: Upload test report
if: always()
@@ -264,7 +264,7 @@ jobs:
export JAVA_HOME=/usr/lib/jvm/java-17-openjdk
export PATH=$JAVA_HOME/bin:$PATH
java -version
-
TAGS_EXCLUDE="org.apache.spark.tags.ExtendedSQLTest,org.apache.spark.tags.SlowHiveTest,org.apache.gluten.tags.UDFTest,org.apache.gluten.tags.SkipTest"
+
TAGS_EXCLUDE="org.apache.spark.tags.ExtendedSQLTest,org.apache.spark.tags.SlowHiveTest,org.apache.gluten.tags.UDFTest,org.apache.gluten.tags.CudfTest,org.apache.gluten.tags.SkipTest"
if [ "${{ matrix.group }}" = "1" ]; then
$MVN_CMD clean test -Pspark-4.0 -Pscala-2.13 -Pjava-17
-Pbackends-velox -Piceberg -Pdelta \
-Pspark-ut
-DargLine="-Dspark.test.home=/opt/shims/spark40/spark_home/"
-DtagsToExclude="$TAGS_EXCLUDE" \
diff --git a/.github/workflows/velox_backend_x86.yml
b/.github/workflows/velox_backend_x86.yml
index 951f55cc9f..4a2c600efe 100644
--- a/.github/workflows/velox_backend_x86.yml
+++ b/.github/workflows/velox_backend_x86.yml
@@ -821,7 +821,7 @@ jobs:
java -version
$MVN_CMD clean test -Pspark-3.3 -Pjava-17 -Pbackends-velox -Piceberg
-Pdelta -Phudi -Ppaimon -Pspark-ut \
-DargLine="-Dspark.test.home=/opt/shims/spark33/spark_home/" \
-
-DtagsToExclude=org.apache.spark.tags.ExtendedSQLTest,org.apache.spark.tags.SlowHiveTest,org.apache.gluten.tags.UDFTest,org.apache.gluten.tags.EnhancedFeaturesTest,org.apache.gluten.tags.SkipTest
+
-DtagsToExclude=org.apache.spark.tags.ExtendedSQLTest,org.apache.spark.tags.SlowHiveTest,org.apache.gluten.tags.UDFTest,org.apache.gluten.tags.EnhancedFeaturesTest,org.apache.gluten.tags.CudfTest,org.apache.gluten.tags.SkipTest
- name: Upload test report
if: always()
uses: actions/upload-artifact@v7
@@ -926,7 +926,7 @@ jobs:
export PATH=$JAVA_HOME/bin:$PATH
java -version
export SPARK_HOME=/opt/shims/spark34/spark_home/
-
TAGS_EXCLUDE="org.apache.spark.tags.ExtendedSQLTest,org.apache.spark.tags.SlowHiveTest,org.apache.gluten.tags.UDFTest,org.apache.gluten.tags.EnhancedFeaturesTest,org.apache.gluten.tags.SkipTest"
+
TAGS_EXCLUDE="org.apache.spark.tags.ExtendedSQLTest,org.apache.spark.tags.SlowHiveTest,org.apache.gluten.tags.UDFTest,org.apache.gluten.tags.EnhancedFeaturesTest,org.apache.gluten.tags.CudfTest,org.apache.gluten.tags.SkipTest"
if [ "${{ matrix.group }}" = "1" ]; then
$MVN_CMD clean test -Pspark-3.4 -Pjava-17 -Pbackends-velox
-Piceberg -Piceberg-test -Pdelta -Phudi -Ppaimon -Pspark-ut \
-DtagsToExclude="$TAGS_EXCLUDE"
-DargLine="-Dspark.test.home=$SPARK_HOME" \
@@ -1050,7 +1050,7 @@ jobs:
export JAVA_HOME=/usr/lib/jvm/java-17-openjdk
export PATH=$JAVA_HOME/bin:$PATH
java -version
-
TAGS_EXCLUDE="org.apache.spark.tags.ExtendedSQLTest,org.apache.spark.tags.SlowHiveTest,org.apache.gluten.tags.UDFTest,org.apache.gluten.tags.EnhancedFeaturesTest,org.apache.gluten.tags.SkipTest"
+
TAGS_EXCLUDE="org.apache.spark.tags.ExtendedSQLTest,org.apache.spark.tags.SlowHiveTest,org.apache.gluten.tags.UDFTest,org.apache.gluten.tags.EnhancedFeaturesTest,org.apache.gluten.tags.CudfTest,org.apache.gluten.tags.SkipTest"
if [ "${{ matrix.group }}" = "1" ]; then
$MVN_CMD clean test -Pspark-3.5 -Pjava-17 -Pbackends-velox
-Piceberg -Pdelta -Phudi -Ppaimon -Pspark-ut \
-DargLine="-Dspark.test.home=/opt/shims/spark35/spark_home/"
-DtagsToExclude="$TAGS_EXCLUDE" \
@@ -1122,7 +1122,7 @@ jobs:
export JAVA_HOME=/usr/lib/jvm/java-17-openjdk
export PATH=$JAVA_HOME/bin:$PATH
java -version
-
TAGS_EXCLUDE="org.apache.spark.tags.ExtendedSQLTest,org.apache.spark.tags.SlowHiveTest,org.apache.gluten.tags.UDFTest,org.apache.gluten.tags.EnhancedFeaturesTest,org.apache.gluten.tags.SkipTest"
+
TAGS_EXCLUDE="org.apache.spark.tags.ExtendedSQLTest,org.apache.spark.tags.SlowHiveTest,org.apache.gluten.tags.UDFTest,org.apache.gluten.tags.EnhancedFeaturesTest,org.apache.gluten.tags.CudfTest,org.apache.gluten.tags.SkipTest"
if [ "${{ matrix.group }}" = "1" ]; then
$MVN_CMD clean test -Pspark-3.5 -Pscala-2.13 -Pjava-17
-Pbackends-velox -Piceberg -Pdelta -Pspark-ut \
-DargLine="-Dspark.test.home=/opt/shims/spark35-scala-2.13/spark_home/"
-DtagsToExclude="$TAGS_EXCLUDE" \
@@ -1426,7 +1426,7 @@ jobs:
export JAVA_HOME=/usr/lib/jvm/java-17-openjdk
export PATH=$JAVA_HOME/bin:$PATH
java -version
-
TAGS_EXCLUDE="org.apache.spark.tags.ExtendedSQLTest,org.apache.spark.tags.SlowHiveTest,org.apache.gluten.tags.UDFTest,org.apache.gluten.tags.EnhancedFeaturesTest,org.apache.gluten.tags.SkipTest"
+
TAGS_EXCLUDE="org.apache.spark.tags.ExtendedSQLTest,org.apache.spark.tags.SlowHiveTest,org.apache.gluten.tags.UDFTest,org.apache.gluten.tags.EnhancedFeaturesTest,org.apache.gluten.tags.CudfTest,org.apache.gluten.tags.SkipTest"
if [ "${{ matrix.group }}" = "1" ]; then
$MVN_CMD clean test -Pspark-4.0 -Pscala-2.13 -Pjava-17
-Pbackends-velox -Pdelta -Ppaimon -Pspark-ut \
-DargLine="-Dspark.test.home=/opt/shims/spark40/spark_home/"
-DtagsToExclude="$TAGS_EXCLUDE" \
@@ -1551,7 +1551,7 @@ jobs:
export JAVA_HOME=/usr/lib/jvm/java-17-openjdk
export PATH=$JAVA_HOME/bin:$PATH
java -version
-
TAGS_EXCLUDE="org.apache.spark.tags.ExtendedSQLTest,org.apache.spark.tags.SlowHiveTest,org.apache.gluten.tags.UDFTest,org.apache.gluten.tags.EnhancedFeaturesTest,org.apache.gluten.tags.SkipTest"
+
TAGS_EXCLUDE="org.apache.spark.tags.ExtendedSQLTest,org.apache.spark.tags.SlowHiveTest,org.apache.gluten.tags.UDFTest,org.apache.gluten.tags.EnhancedFeaturesTest,org.apache.gluten.tags.CudfTest,org.apache.gluten.tags.SkipTest"
if [ "${{ matrix.group }}" = "1" ]; then
# Group 1: streaming + gluten utils + top-level spark tests (~55
classes)
$MVN_CMD clean test -Pspark-4.1 -Pscala-2.13 -Pjava-17
-Pbackends-velox -Pspark-ut -Pdelta \
diff --git a/backends-velox/pom.xml b/backends-velox/pom.xml
index 8f833020bb..acdc199c17 100644
--- a/backends-velox/pom.xml
+++ b/backends-velox/pom.xml
@@ -374,7 +374,7 @@
<activeByDefault>true</activeByDefault>
</activation>
<properties>
-
<tagsToExclude>org.apache.gluten.tags.UDFTest,org.apache.gluten.tags.EnhancedFeaturesTest,org.apache.spark.tags.SkipTest</tagsToExclude>
+
<tagsToExclude>org.apache.gluten.tags.UDFTest,org.apache.gluten.tags.EnhancedFeaturesTest,org.apache.gluten.tags.CudfTest,org.apache.spark.tags.SkipTest</tagsToExclude>
</properties>
</profile>
<profile>
diff --git
a/backends-velox/src/main/scala/org/apache/gluten/execution/VeloxBroadcastBuildSideRDD.scala
b/backends-velox/src/main/scala/org/apache/gluten/execution/VeloxBroadcastBuildSideRDD.scala
index 2d4b157056..074eae17df 100644
---
a/backends-velox/src/main/scala/org/apache/gluten/execution/VeloxBroadcastBuildSideRDD.scala
+++
b/backends-velox/src/main/scala/org/apache/gluten/execution/VeloxBroadcastBuildSideRDD.scala
@@ -16,6 +16,7 @@
*/
package org.apache.gluten.execution
+import org.apache.gluten.config.GlutenConfig
import org.apache.gluten.iterator.Iterators
import org.apache.spark.{broadcast, SparkContext}
@@ -38,7 +39,14 @@ case class VeloxBroadcastBuildSideRDD(
case unsafe: UnsafeColumnarBuildSideRelation =>
unsafe.isOffload
}
- val output = if (isBNL || !offload) {
+ // With cuDF enabled the hash join runs as CudfHashJoin, which builds its
own GPU
+ // hash table from the build-side value stream and cannot consume the
prebuilt CPU
+ // OpaqueHashTable. Feeding Iterator.empty here silently produces empty
join
+ // results, so stream the broadcast batches instead (the same path shuffle
joins
+ // use on GPU). Skipping the CPU cache build also keeps hybrid mode
correct:
+ // VeloxBroadcastBuildSideCache.get finds no table, so the HashJoinNode
carries no
+ // reusable table and a CPU-fallback join builds from this stream as usual.
+ val output = if (isBNL || !offload || GlutenConfig.get.enableColumnarCudf)
{
val relation = broadcasted.value.asReadOnlyCopy()
Iterators
.wrap(relation.deserialized)
diff --git a/backends-velox/src/test/java/org/apache/gluten/tags/CudfTest.java
b/backends-velox/src/test/java/org/apache/gluten/tags/CudfTest.java
new file mode 100644
index 0000000000..5874be95e0
--- /dev/null
+++ b/backends-velox/src/test/java/org/apache/gluten/tags/CudfTest.java
@@ -0,0 +1,29 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.gluten.tags;
+
+import org.scalatest.TagAnnotation;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+@TagAnnotation
+@Retention(RetentionPolicy.RUNTIME)
+@Target({ElementType.METHOD, ElementType.TYPE})
+public @interface CudfTest {}
diff --git
a/backends-velox/src/test/scala/org/apache/gluten/execution/CudfBroadcastJoinSuite.scala
b/backends-velox/src/test/scala/org/apache/gluten/execution/CudfBroadcastJoinSuite.scala
new file mode 100644
index 0000000000..62a9a64e3b
--- /dev/null
+++
b/backends-velox/src/test/scala/org/apache/gluten/execution/CudfBroadcastJoinSuite.scala
@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.gluten.execution
+
+import org.apache.gluten.tags.CudfTest
+
+import org.apache.spark.SparkConf
+
+/**
+ * Regression tests for GLUTEN-12471: broadcast hash joins on the cuDF (GPU)
backend silently
+ * returned empty results because CudfHashJoin built its hash table from the
empty build-side
+ * iterator instead of the prebuilt CPU table.
+ *
+ * These tests require GPU hardware and a cuDF-enabled build, so they are
tagged with CudfTest and
+ * excluded from regular CI runs. Run on a GPU host:
+ *
+ * mvn test -Pbackends-velox -DtagsToExclude=None \
+ * -DwildcardSuites=org.apache.gluten.execution.CudfBroadcastJoinSuite
+ */
+@CudfTest
+class CudfBroadcastJoinSuite extends VeloxWholeStageTransformerSuite {
+
+ override protected val resourcePath: String = "/tpch-data-parquet"
+ override protected val fileFormat: String = "parquet"
+
+ override protected def sparkConf: SparkConf = {
+ super.sparkConf
+ .set("spark.gluten.sql.columnar.cudf", "true")
+ .set("spark.gluten.sql.columnar.backend.velox.cudf.allowCpuFallback",
"false")
+ // Force the broadcast path: the small side (orders filter) fits easily.
+ .set("spark.sql.autoBroadcastJoinThreshold", "10MB")
+ .set("spark.shuffle.manager",
"org.apache.spark.shuffle.sort.ColumnarShuffleManager")
+ .set("spark.memory.offHeap.enabled", "true")
+ .set("spark.memory.offHeap.size", "4g")
+ }
+
+ override def beforeAll(): Unit = {
+ super.beforeAll()
+ createTPCHNotNullTables()
+ }
+
+ test("GLUTEN-12471: cuDF broadcast hash join returns non-empty, correct
results") {
+ val query =
+ """
+ |SELECT l.l_orderkey, o.o_orderdate, l.l_extendedprice
+ |FROM lineitem l
+ |JOIN orders o ON l.l_orderkey = o.o_orderkey
+ |WHERE o.o_orderdate < date '1995-01-01'
+ |""".stripMargin
+
+ // runQueryAndCompare executes on Gluten AND vanilla Spark and compares
+ // results -- this is the core regression check: before the fix, the Gluten
+ // side returned 0 rows and the comparison failed.
+ runQueryAndCompare(query) {
+ df =>
+ val plan = df.queryExecution.executedPlan
+ // The join must actually be offloaded as a broadcast hash join, not
+ // demoted or fallen back.
+ val bhj = collect(plan) { case j: BroadcastHashJoinExecTransformer =>
j }
+ assert(bhj.nonEmpty, s"expected an offloaded broadcast hash join,
got:\n$plan")
+ assert(df.count() > 0, "broadcast join must not return empty results
(GLUTEN-12471)")
+ }
+ }
+
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
