Skip to content

[VL] Fix broadcast hash table reuse for reused exchanges#12264

Open
wecharyu wants to merge 3 commits into
apache:mainfrom
wecharyu:fix_shared_bhj_table
Open

[VL] Fix broadcast hash table reuse for reused exchanges#12264
wecharyu wants to merge 3 commits into
apache:mainfrom
wecharyu:fix_shared_bhj_table

Conversation

@wecharyu

@wecharyu wecharyu commented Jun 8, 2026

Copy link
Copy Markdown
Contributor

What changes are proposed in this pull request?

Create unique hash table ID with plan_id, build_keys, join_type and null_aware, which determine the hash table contents.

How was this patch tested?

Add UT.

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

Generated-by: Codex GPT-5.5

@github-actions

github-actions Bot commented Jun 8, 2026

Copy link
Copy Markdown

Run Gluten Clickhouse CI on x86

}
}

override def doCanonicalizeForBroadcastMode(mode: BroadcastMode): BroadcastMode = {

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.

Is this part still needed when buildHashTableOncePerExecutor is disabled?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

doCanonicalizeForBroadcastMode() is not invoked anywhere, broadcast canonicalization goes through ColumnarBroadcastExchangeExec.doCanonicalize().

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.

This code is actually useful — when buildHashTableOncePerExecutor is disabled, it provides more opportunities to reuse broadcast exchanges. Moreover, I now think that even when buildHashTableOncePerExecutor is enabled, the comment in doCanonicalizeForBroadcastMode still holds true: we still broadcast byte arrays and build HashRelation at the executor side.
@JkSelf Can you explain why this was removed? This allows us to reuse broadcast exchanges for different build keys with the same data.
We should either restore the code before ColumnarBroadcastExchangeExec.doCanonicalize, or at least follow the original logic when buildHashTableOncePerExecutor is disabled.

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.

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.

@JkSelf Thank you for the explanation, I understand now. @wecharyu According to the instructions here, you can restore the original behavior of doCanonicalizeForBroadcastMode when enableBroadcastBuildOncePerExecutor=false.

@github-actions

github-actions Bot commented Jun 9, 2026

Copy link
Copy Markdown

Run Gluten Clickhouse CI on x86

@wecharyu wecharyu force-pushed the fix_shared_bhj_table branch from 075dc89 to 7552606 Compare June 9, 2026 16:07
@github-actions

github-actions Bot commented Jun 9, 2026

Copy link
Copy Markdown

Run Gluten Clickhouse CI on x86

1 similar comment
@github-actions

github-actions Bot commented Jun 9, 2026

Copy link
Copy Markdown

Run Gluten Clickhouse CI on x86

@wecharyu wecharyu force-pushed the fix_shared_bhj_table branch from d7d69eb to d88b181 Compare June 10, 2026 02:08
@github-actions

Copy link
Copy Markdown

Run Gluten Clickhouse CI on x86

@JkSelf

JkSelf commented Jun 10, 2026

Copy link
Copy Markdown
Contributor

@wecharyu Thanks for your work.

For ColumnarBroadcastExchangeExec, reuse is already decided by doCanonicalize(). It now directly uses mode.canonicalized, and that canonicalized BroadcastMode already captures the key BHJ semantics such as bound build keys and null-aware behavior. So we no longer need this handling for the unique broadcastId.

@wecharyu

wecharyu commented Jun 10, 2026

Copy link
Copy Markdown
Contributor Author

@JkSelf Thanks for the prompt response. HashedRelationBroadcastMode does not contains joinType, which could also change the build hash table data. For example in this PR's test the LEFT SEMI JOIN and INNER JOIN would use the same hash table, which cause the test failed before this PR. https://github.com/apache/spark/blob/62ae4db28f3be8a0ca2c3016d27ca5a62f02915d/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashedRelation.scala#L1152-L1153

And since current Gluten is still broadcast the raw build plan data bytes instead of hash table,I agree with @liujiayi771 that BroadcastExchangeExec can be reused for the same build plan even when the build keys differ. This reuse avoids broadcasting the same data multiple times.

I think now we can make following improvements:

  1. Move doCanonicalizeForBroadcastMode() back to reuse the broadcast exchange as much as possible.
  2. Associate the constructed hash table with the plan_id, build_keys, drop_duplicates, and null_aware flag, as these attributes uniquely identify the hash table.

@JkSelf

JkSelf commented Jun 11, 2026

Copy link
Copy Markdown
Contributor

HashedRelationBroadcastMode does not contains joinType, which could also change the build hash table data. For example in this PR's test the LEFT SEMI JOIN and INNER JOIN would use the same hash table, which cause the test failed before this PR.

Gluten's implementation is the same as vanilla Spark's. Just out of curiosity, can Spark pass this failed test?

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

Labels

CORE works for Gluten Core VELOX

Projects

None yet

Development

Successfully merging this pull request may close these issues.

3 participants