[VL] Deserialize broadcast build side per the consuming stage's cuDF tag

backends-velox/src/main/scala/org/apache/gluten/execution/HashJoinExecTransformer.scala

@@ -221,7 +221,7 @@ case class BroadcastHashJoinExecTransformer(
         } else {
           logInfo(s"Using executor-side broadcast hash table build for $buildBroadcastTableId")
         }
-        VeloxBroadcastBuildSideRDD(sparkContext, broadcast, context)
+        VeloxBroadcastBuildSideRDD(sparkContext, broadcast, context, cudfEnabled = offloadCuda)
 
       case unsafe: UnsafeColumnarBuildSideRelation =>
         joinParamsForMetrics.foreach(_.usesDriverSideSerializedHashTable = false)
@@ -235,15 +235,15 @@ case class BroadcastHashJoinExecTransformer(
         } else {
           logInfo(s"Using executor-side broadcast hash table build for $buildBroadcastTableId")
         }
-        VeloxBroadcastBuildSideRDD(sparkContext, broadcast, context)
+        VeloxBroadcastBuildSideRDD(sparkContext, broadcast, context, cudfEnabled = offloadCuda)
 
       case other =>
         joinParamsForMetrics.foreach(_.usesDriverSideSerializedHashTable = false)
         // Fallback for unknown types
         logWarning(
           s"Unknown broadcast relation type: ${other.getClass.getName}, " +
             "using executor-side build")
-        VeloxBroadcastBuildSideRDD(sparkContext, broadcast, context)
+        VeloxBroadcastBuildSideRDD(sparkContext, broadcast, context, cudfEnabled = offloadCuda)
     }
 
     // FIXME: Do we have to make build side a RDD?

backends-velox/src/main/scala/org/apache/gluten/execution/VeloxBroadcastBuildSideRDD.scala

@@ -29,7 +29,8 @@ case class VeloxBroadcastBuildSideRDD(
     @transient private val sc: SparkContext,
     broadcasted: broadcast.Broadcast[BuildSideRelation],
     broadcastContext: BroadcastHashJoinContext,
-    isBNL: Boolean = false)
+    isBNL: Boolean = false,
+    cudfEnabled: Boolean = false)
   extends BroadcastBuildSideRDD(sc, broadcasted) {
 
   override def genBroadcastBuildSideIterator(): Iterator[ColumnarBatch] = {
@@ -48,8 +49,17 @@ case class VeloxBroadcastBuildSideRDD(
     // 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()
+      // cudfEnabled is the consuming stage's own tag (TransformSupport#offloadCuda).
+      val batches = relation match {
+        case columnar: ColumnarBuildSideRelation =>
+          columnar.deserialized(cudfEnabled)
+        case unsafe: UnsafeColumnarBuildSideRelation =>
+          unsafe.deserialized(cudfEnabled)
+        case other =>
+          other.deserialized
+      }
       Iterators
-        .wrap(relation.deserialized)
+        .wrap(batches)
         .recyclePayload(batch => batch.close())
         .create()
     } else {

backends-velox/src/main/scala/org/apache/gluten/execution/VeloxBroadcastNestedLoopJoinExecTransformer.scala

@@ -45,7 +45,13 @@ case class VeloxBroadcastNestedLoopJoinExecTransformer(
   override def columnarInputRDDs: Seq[RDD[ColumnarBatch]] = {
     val streamedRDD = getColumnarInputRDDs(streamedPlan)
     val broadcast = buildPlan.executeBroadcast[BuildSideRelation]()
-    val broadcastRDD = VeloxBroadcastBuildSideRDD(sparkContext, broadcast, null, true)
+    val broadcastRDD =
+      VeloxBroadcastBuildSideRDD(
+        sparkContext,
+        broadcast,
+        null,
+        isBNL = true,
+        cudfEnabled = offloadCuda)
     // FIXME: Do we have to make build side a RDD?
     streamedRDD :+ broadcastRDD
   }

backends-velox/src/main/scala/org/apache/spark/sql/execution/ColumnarBuildSideRelation.scala

@@ -18,6 +18,7 @@ package org.apache.spark.sql.execution
 
 import org.apache.gluten.backendsapi.BackendsApiManager
 import org.apache.gluten.columnarbatch.ColumnarBatches
+import org.apache.gluten.config.GlutenConfig
 import org.apache.gluten.execution.BroadcastHashJoinContext
 import org.apache.gluten.expression.ConverterUtils
 import org.apache.gluten.iterator.Iterators
@@ -40,6 +41,8 @@ import org.apache.spark.util.KnownSizeEstimation
 
 import org.apache.arrow.c.ArrowSchema
 
+import java.util.Collections
+
 import scala.collection.JavaConverters._
 import scala.collection.JavaConverters.asScalaIteratorConverter
 import scala.collection.mutable.{ArrayBuffer, Map}
@@ -110,9 +113,20 @@ case class ColumnarBuildSideRelation(
       }
   }
 
-  override def deserialized: Iterator[ColumnarBatch] = {
+  /** Host-resident deserialization, for CPU consumers. */
+  override def deserialized: Iterator[ColumnarBatch] = deserialized(cudfEnabled = false)
+
+  /**
+   * Residency follows the consuming stage: a cuDF-offloaded stage sources from CudfValueStream and
+   * needs device batches, a non-offloaded stage from RowVectorStream and needs host batches.
+   */
+  def deserialized(cudfEnabled: Boolean): Iterator[ColumnarBatch] = {
     val runtime =
-      Runtimes.contextInstance(BackendsApiManager.getBackendName, "BuildSideRelation#deserialized")
+      Runtimes.contextInstance(
+        BackendsApiManager.getBackendName,
+        "BuildSideRelation#deserialized",
+        Collections.singletonMap(GlutenConfig.COLUMNAR_CUDF_ENABLED.key, cudfEnabled.toString)
+      )
     val jniWrapper = ColumnarBatchSerializerJniWrapper.create(runtime)
     val serializeHandle: Long = {
       val allocator = ArrowBufferAllocators.contextInstance()

backends-velox/src/main/scala/org/apache/spark/sql/execution/unsafe/UnsafeColumnarBuildSideRelation.scala

@@ -18,6 +18,7 @@ package org.apache.spark.sql.execution.unsafe
 
 import org.apache.gluten.backendsapi.BackendsApiManager
 import org.apache.gluten.columnarbatch.ColumnarBatches
+import org.apache.gluten.config.GlutenConfig
 import org.apache.gluten.execution.BroadcastHashJoinContext
 import org.apache.gluten.expression.ConverterUtils
 import org.apache.gluten.iterator.Iterators
@@ -45,6 +46,7 @@ import com.esotericsoftware.kryo.io.{Input, Output}
 import org.apache.arrow.c.ArrowSchema
 
 import java.io.{Externalizable, ObjectInput, ObjectOutput}
+import java.util.Collections
 
 import scala.collection.JavaConverters._
 import scala.collection.JavaConverters.asScalaIteratorConverter
@@ -372,11 +374,20 @@ class UnsafeColumnarBuildSideRelation(
       }
   }
 
-  override def deserialized: Iterator[ColumnarBatch] = {
+  /** Host-resident deserialization, for CPU consumers. */
+  override def deserialized: Iterator[ColumnarBatch] = deserialized(cudfEnabled = false)
+
+  /**
+   * Residency follows the consuming stage: a cuDF-offloaded stage sources from CudfValueStream and
+   * needs device batches, a non-offloaded stage from RowVectorStream and needs host batches.
+   */
+  def deserialized(cudfEnabled: Boolean): Iterator[ColumnarBatch] = {
     val runtime =
       Runtimes.contextInstance(
         BackendsApiManager.getBackendName,
-        "UnsafeBuildSideRelation#deserialize")
+        "UnsafeBuildSideRelation#deserialize",
+        Collections.singletonMap(GlutenConfig.COLUMNAR_CUDF_ENABLED.key, cudfEnabled.toString)
+      )
     val jniWrapper = ColumnarBatchSerializerJniWrapper.create(runtime)
     val serializerHandle: Long = {
       val allocator = ArrowBufferAllocators.contextInstance()