[VL] Deserialize broadcast build side on host regardless of session cuDF conf - #12838
[VL] Deserialize broadcast build side on host regardless of session cuDF conf#12838ReemaAlzaid wants to merge 1 commit into
Conversation
…uDF conf BuildSideRelation#deserialized creates its Runtime from the session conf, so with spark.gluten.sql.columnar.cudf=true VeloxRuntime::createColumnarBatchSerializer returns the GPU serializer and every broadcast batch is uploaded to the device, even when the consuming stage was not offloaded to cuDF and planned a host-contract value stream. TPC-H q16 reproduces this deterministically: its not-in subquery is a null-aware anti join, which Spark always executes as a broadcast join even with spark.sql.autoBroadcastJoinThreshold=-1, so the untagged consumer receives device-resident CudfVectors and fails. Pass a per-instance COLUMNAR_CUDF_ENABLED=false override so broadcast batches are always deserialized to host. Both serializer flavors share the same wire format (the GPU one only overrides deserialize), and cuDF consumers upload host batches themselves via CudfVectorStream.
|
The join stage of BHJ is supported on GPU. We shouldn't set |
marin-ma
left a comment
There was a problem hiding this comment.
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.
|
Tested tpcds q95 locally with a small dataset. Before this change it can pass, but with this change it fails. |
I agree with you on this. The missing piece is #12471, which uploads in 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:
I lean toward 2. Happy to do either |
What changes are proposed in this pull request?
With
spark.gluten.sql.columnar.cudf=true,BuildSideRelation#deserializedcreates its Runtime from the session conf, soVeloxRuntime::createColumnarBatchSerializerreturns VeloxGpuColumnarBatchSerializer and every broadcast batch is uploaded to the device even when the consuming stage was not offloaded to cuDF and planned a host-contract value stream. The stream contract is decided per stage (the cudf tag), but broadcast residency was decided per session, and the two can disagree.TPCH q16 reproduces this deterministically: its not in subquery is a nul -aware anti join, which Spark always executes as a broadcast join even with
spark.sql.autoBroadcastJoinThreshold=-1, so the untagged consumer receives device resident CudfVectors on its first batch and fails.Fix: pass a per instance
COLUMNAR_CUDF_ENABLED=falseoverride so broadcast batches always deserialize to host. Both serializer flavors share the same wire format (the GPU one only overrides deserialize), and cuDF offloaded consumers upload host batches themselves via CudfVectorStream (#12471 this PR depends on that upload path for the GPU consumer case).How was this patch tested?
Verified on an L40S (TPC-H sf1, queries-compare): fallback mode with broadcasts disabled goes from failing to 22/22; pure-GPU mode with broadcasts enabled holds its baseline (remaining failures are pre-existing cuDF expression-coverage gaps: row_constructor_with_null, spark_legacy_cast).
Was this patch authored or co-authored using generative AI tooling?