[VL] Fix empty GPU broadcast joins and CudfVector host read crashes in the cuDF backend

[ReemaAlzaid]

What changes are proposed in this pull request?

Broadcast hash joins on the cuDF (GPU) backend were silently returning empty
results, and GPU batches were crashing whenever CPU code tried to read them.
This PR fixes both:

1. Broadcast joins built from an empty stream (VeloxBroadcastBuildSideRDD.scala)
For broadcast joins, Gluten puts the build-side data into a prebuilt CPU hash
table and feeds the plan an empty iterator. The CPU join uses that prebuilt
table but the GPU join (CudfHashJoin) doesn't know it exists, so it built
its hash table from the empty iterator and every broadcast join returned 0 rows.
Fix: when cuDF is enabled, stream the broadcast batches into the plan, same as
shuffle joins already do.

2. GPU vectors crashing on host reads (new CudfVectorUtils.h + call sites)
A CudfVector on the GPU and has no host side children, so host code
reading or serializing it saw garbage (the childAt crashes). Added a small
helper that copies the vector to host first, and applied it at every
host read site including the ColumnarToRow converter, which was causing a
JVM segfault in q15.

Needs a Velox build with the multi-column hash_with_seed fix
facebookincubator/velox#18047

How was this patch tested?

sbin/gluten-it.sh queries-compare \
  --local --preset=velox --benchmark-type=h --error-on-memleak \
  --off-heap-size=10g -s=1 --threads=1 --iterations=1 \
  --decimal-as-double=true \
  --extra-conf=spark.gluten.sql.columnar.backend.velox.cudf.allowCpuFallback=false \
  --extra-conf=spark.gluten.sql.columnar.cudf=true

gluten-it queries-compare vs vanilla Spark (2× NVIDIA L40S, Spark 3.5,
pure GPU: allowCpuFallback=false, broadcast joins enabled):

• TPC-H: 3/22 → 16/22 passing
• TPC-DS: 2/103 → 43/103 passing, 828 crashes → 0
• TPC-H SF10 with GPU scans: ~5.5× faster than vanilla Spark overall
  (e.g. q21: 82s → 11s)

Remaining failures are pre-existing cuDF expression gaps
(row_constructor_with_null, substring), not related to this change.

cc: @marin-ma @zhouyuan

[marin-ma]

Thanks for working on this issue. Adding check and conversion from CudfVector to RowVector do avoid runtime failure.

However, in my opinion, it's expected that the output from gpu pipeline should always be converted into RowVector when the pipeline ends or the next operator is not a CudfOperator (such as shuffle write, file write, c2r, etc), so I wonder why the input of the c2r for q15 is not being converted. Is there any findings on the cause? Or please point me out if I miss understood something.

I also wonder if making this change general (adding guard on every RowVector retrieval) may burry some real issues.

[marin-ma]

There are test failures in cpp-test-udf-test PTAL. Thanks!

[ReemaAlzaid]

Thanks for working on this issue. Adding check and conversion from CudfVector to RowVector do avoid runtime failure.

However, in my opinion, it's expected that the output from gpu pipeline should always be converted into RowVector when the pipeline ends or the next operator is not a CudfOperator (such as shuffle write, file write, c2r, etc), so I wonder why the input of the c2r for q15 is not being converted. Is there any findings on the cause? Or please point me out if I miss understood something.

I also wonder if making this change general (adding guard on every RowVector retrieval) may burry some real issues.

your right about the design inside a Velox pipeline, but in Gluten, batches also cross the JNI boundary through channels the auto inserter can't see, and their CPU/GPU is only known at runtime we verified GPU vectors reaching C2R (q15) and CPU vectors reaching GPU ops (q18), plus GPU vectors being silently serialized as 0 rows (the empty-broadcast-join root cause). The guards sit only at those boundary consumers, not everywhere. To avoid hiding real bugs, I'll add a warning log when a guard actually fires, and I can post the q15 operator dump if you want to chase the exact insertion gap

[marin-ma]

and I can post the q15 operator dump if you want to chase the exact insertion gap

@ReemaAlzaid This would be helpful. Can you please share it?

plus GPU vectors being silently serialized as 0 rows (the empty-broadcast-join root cause)

This is an issue that can be buried by this conversion. Currently the bhj build once feature is not working in gpu pipeline. We should disable this feature for gpu workload, rather than enable it and add CudfVector to RowVector conversion.

[ReemaAlzaid]

Thanks @marin-ma. Operator dump attached two q15 runs on the current branch with spark.gluten.sql.debug.cudf=true: default, and with autoBroadcastJoinThreshold=-1 (the config the original crash occurred under).

Result: I can no longer reproduce the escape. Every driver whose output leaves the task now ends with a trailing CudfToVelox (the only GPU-terminated pipeline is the join-build sink, which is correct producesGpuOutput=0), and q15 passes value correct in both configs. The original crash
was against the earlier Velox pin, and an intervening change closed the gap; at that time the C2R input was verifiably a device CudfVector (SIGSEGV on garbage VARCHAR offsets). Given that, the conversions are no longer load-bearing — I've made them log a warning when they actually fire, plus an opt-in cudf.strictResidency conf that fails instead of converting, so CI can catch any future insertion regression rather than bury it. Happy to trim the guard sites further if you prefer.

On bhj build-once: agreed that's what this PR does. VeloxBroadcastBuildSideRDD skips the prebuilt hash table path entirely when cuDF is enabled and streams the broadcast batches instead (the GPU join can't consume the opaque table, which is exactly why builds arrived empty). If you'd rather see that as a dedicated config or plan rule instead of the branch in the RDD, I can restructure

q15_log_dump.md

[github-actions]

Run Gluten Clickhouse CI on x86

[marin-ma]

Can you extract this change along with the unit test to another PR?

[ReemaAlzaid]

yes here you go #12812

[marin-ma]

@ReemaAlzaid Apologise for missing your previous comment.

Here's my understanding and some suggestions:

The first issue 1. Broadcast joins built from an empty stream can be workaround by setting spark.gluten.velox.buildHashTableOncePerExecutor.enabled=false because Velox Cudf doesn't support this feature. This PR disable this optimisation in the code so we don't need to explicitly disable the bhj optimisation for GPU workloads, and I agree with this change.

The second issue 2. GPU vectors crashing on host reads has occurred in the previous test but now it cannot be reproduced, while this PR still keep the changes to serve as a temporary remedy in case the CI failure in the future. If so, I do not agree with this change. The major reason is that it still does have the down side to burry some real issues by silencing the conversion, making the real issue hard to trace. In terms of GPU CI, currently we only have the CI job to test the GPU compilation, and there's no test running. If we can have the CI to run GPU testing in the future and once it's broken and needs fix from Velox, we can temporary disable the GPU CI rather than switching on this conversion in the code.

[ReemaAlzaid]

@ReemaAlzaid Apologise for missing your previous comment.

Here's my understanding and some suggestions:

The first issue 1. Broadcast joins built from an empty stream can be workaround by setting spark.gluten.velox.buildHashTableOncePerExecutor.enabled=false because Velox Cudf doesn't support this feature. This PR disable this optimisation in the code so we don't need to explicitly disable the bhj optimisation for GPU workloads, and I agree with this change.

The second issue 2. GPU vectors crashing on host reads has occurred in the previous test but now it cannot be reproduced, while this PR still keep the changes to serve as a temporary remedy in case the CI failure in the future. If so, I do not agree with this change. The major reason is that it still does have the down side to burry some real issues by silencing the conversion, making the real issue hard to trace. In terms of GPU CI, currently we only have the CI job to test the GPU compilation, and there's no test running. If we can have the CI to run GPU testing in the future and once it's broken and needs fix from Velox, we can temporary disable the GPU CI rather than switching on this conversion in the code.

@marin-ma no worries, thanks for getting back!

I Kept the VeloxBroadcastBuildSideRDD change, and the conversion is gone completely materializeVeloxRowVector, all call sites, and the strictResidency config.

You were right about it hiding things too. Once I removed it, TPC-H on GPU started failing loudly and pointed at two real bugs in the value streams: RowVectorStream::next() was rebuilding device batches into RowVectors with no children (the childAt() crash in CudfFromVelox, q17/q19), and CudfValueStream was forwarding host batches straight into GPU-only operators (the cudfInput != nullptr crash in CudfTopN, q18). Both fixed device batches pass through as-is, host batches get uploaded. The matching Velox side fix is up here with tests that reproduce the crash: facebookincubator/velox#18539

With that, TPC-H sf1 with allowCpuFallback=false is at 19/22 — the remaining three (q1/q17/q22)

Test gating is now the CudfTest tag per your earlier comment, and I deleted the q15-style test since the code it covered is gone.

[marin-ma]

RowVectorStream::next() was rebuilding device batches into RowVectors

shouldn't happen if CudfToVelox is the last operator to the GPU pipeline. This seems more like the bug from the planner phase. Can you please look into it?

[ReemaAlzaid]

RowVectorStream::next() was rebuilding device batches into RowVectors

shouldn't happen if CudfToVelox is the last operator to the GPU pipeline. This seems more like the bug from the planner phase. Can you please look into it?

I looked into it and as u said here a GPU pipeline always ends with CudfToVelox, so device batches can't leak from its output. The issue is the broadcast build side, which bypasses that path entirely

In q16, the NOT IN becomes a null aware anti join and Spark forces it to broadcast even with autoBroadcastJoinThreshold=-1. The consuming stage is planned for CPU, but ColumnarBuildSideRelation#deserialized creates its Runtime from the session config, so it picks VeloxGpuColumnarBatchSerializer, which uploads the broadcast batches to the device into a stage that expects host batches. It fails deterministically on the first batch of every task.

The fix (#12838): broadcast bytes are host resident by nature the only question is where the single host to device upload happens. The deserializer can't know whether the consuming stage is GPU or CPU, so it should not be the one deciding; it now always hands over host batches (via a COLUMNAR_CUDF_ENABLED=false override scoped to its own Runtime the session config is untouched).

The residency decision then happens at the consumer, the only place that knows the stage's contract: a GPU stage uploads through CudfVectorStream as its first operator, a CPU stage uses the batches directly. GPU execution is unchanged same operators, same single upload, just performed where it can't go wrong.