Skip to content

[VL] Deserialize broadcast build side on host regardless of session cuDF conf - #12838

Open
ReemaAlzaid wants to merge 1 commit into
apache:mainfrom
ReemaAlzaid:cudf-broadcast-host-deserialize
Open

[VL] Deserialize broadcast build side on host regardless of session cuDF conf#12838
ReemaAlzaid wants to merge 1 commit into
apache:mainfrom
ReemaAlzaid:cudf-broadcast-host-deserialize

Conversation

@ReemaAlzaid

Copy link
Copy Markdown
Contributor

What changes are proposed in this pull request?

With spark.gluten.sql.columnar.cudf=true, BuildSideRelation#deserialized creates its Runtime from the session conf, so VeloxRuntime::createColumnarBatchSerializer returns 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=false override 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?

…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.
@marin-ma

Copy link
Copy Markdown
Contributor

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

@marin-ma marin-ma left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

Copy link
Copy Markdown
Contributor

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

Copy link
Copy Markdown
Contributor Author

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:

I lean toward 2. Happy to do either

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants