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

[ReemaAlzaid]

What changes are proposed in this pull request?

BuildSideRelation#deserialized built its Runtime from the session conf, so
broadcast batch residency was decided session wide, while the consuming
stage's value stream contract is decided per stage by the cuDF tag. The two
disagree in both directions: an untagged stage receives device-resident
batches on a host contract ValueStreamNode (TPC-H q16), and a tagged stage's
CudfHashJoinBuild requires device input (TPC-DS q95).

Pass the consuming stage's offloadCuda into deserialized instead, so a
cuDF offloaded stage gets the GPU serializer and a non-offloaded stage gets
the host one. This mirrors the shuffle read path, which already selects its
deserializer from the consumer's output type rather than the session conf.

BroadcastUtils keeps the no arg deserialized, which is now host-resident —
it feeds VeloxColumnarToRow and wants host batches in every mode.

How was this patch tested?

Tested on an L40S without #12471 applied: TPC-DS q95 and TPC-H q16 both pass
in hybrid and pure-GPU.

Was this patch authored or co-authored using generative AI tooling?

[marin-ma]

The join stage of BHJ is supported on GPU. We shouldn't set spark.sql.autoBroadcastJoinThreshold=-1.

[marin-ma]

Based on my understanding, the gap lies in the VeloxGpuColumnarBatchSerializer is always called regardless of the hash join stage is executed on cpu or gpu. When cudf is enabled, the query fails when the hash join is executed on CPU, but can pass on GPU.

This change will cause the hash join to fail when it is executed on the GPU.

[marin-ma]

Tested tpcds q95 locally with a small dataset. Before this change it can pass, but with this change it fails.

GLUTEN_IT_JVM_ARGS=-Xmx5G sbin/gluten-it.sh queries \
      --local --preset=velox --benchmark-type=ds --error-on-memleak --off-heap-size=10g -s=1 --threads=1 --iterations=1 --decimal-as-double=true --enable-history --enable-ui --data-gen=once --extra-conf=spark.gluten.sql.columnar.cudf=true --extra-conf=spark.gluten.sql.debug=true --extra-conf=spark.gluten.sql.debug.cudf=true --extra-conf=spark.gluten.velox.buildHashTableOncePerExecutor.enabled=false --extra-conf=spark.sql.autoBroadcastJoinThreshold=100000 --queries=q95

[ReemaAlzaid]

Based on my understanding, the gap lies in the VeloxGpuColumnarBatchSerializer is always called regardless of the hash join stage is executed on cpu or gpu. When cudf is enabled, the query fails when the hash join is executed on CPU, but can pass on GPU.

This change will cause the hash join to fail when it is executed on the GPU.

I agree with you on this. CudfValueStream extends CudfOperator, so the adapter marks it as producing GPU output and inserts no CudfFromVelox after it grep -c from-velox on a q95 run is 0. The GPU serializer was the only thing uploading. This PR removes it and puts nothing in its place, so CudfHashJoinBuild gets a host batch and the cast fails. Exactly what you saw.

The missing piece is #12471, which uploads in CudfVectorStream instead. With both applied q95 passes, same plan and same build tables as before one H2D copy, just moved to the consumer

Also on autoBroadcastJoinThreshold=-1; I only used it to isolate q16 to a single broadcast, and I'll re run at the default.

So there are two options we could do:

• Land [VL] Fix empty GPU broadcast joins and CudfVector host read crashes in the cuDF backend #12471 first and rebase this on it.

• Or Make this self contained: pass the consuming stage's offloadCuda into BuildSideRelation#deserialized instead of a hard coded false. GPU stage → GPU serializer, CPU stage → host serializer, no cross PR dependency

I lean toward 2. Happy to do either

[marin-ma]

I wouldn't agree with the changes in #12471 Cudf related execution shouldn't be introduced into RowVectorStream. The errors should mostly come from the incorrect shape of the pipeline. We should fix the pipeline construction rather than adding the "RowVector from/to CudfVector" conversions in Gluten.

Please continue with solution 2. Based on the current support status, after the fix the correct pipeline for BHJ should be:

GPU stage:
GPU Deserialiser (host to device) -> CudfValueStream -> Cudf pipeline -> CudfToVelox (device to host) -> shuffle write/broadcast serialization/c2r/...

CPU stage:
CPU Deserialiser -> RowVectorStream -> CPU pipeline -> shuffle write/broadcast serialization/c2r/...

[ReemaAlzaid]

@marin-ma Pushed, the deserializer now follows the consuming stage's offloadCuda

Traced on q95: CudfValueStream → CudfHashJoinBuild on the build side, CudfValueStream → … → CudfToVelox on the probe, 0 CudfFromVelox inserted. Untagged stages take the host deserializer and RowVectorStream

One flag: dropping #12471 exposes q3/q10/q18, failing at CudfTopN::doAddInput TakeOrderedAndProject's runtime built exchange skips AdjustStageExecutionMode. A control run with the old serializer fails identically, so not a regression here, but it's the same wrong pipeline shape. Should I open a separate issue for it? or continue on it from the #12471 patch?