[SPARK-58163][SQL] Extract the BroadcastHashJoinExec read-only relation setup into a shared helper

[LuciferYang]

What changes were proposed in this pull request?

BroadcastHashJoinExec takes a read-only copy of the broadcast HashedRelation and records its size as the task's peak execution memory in three places using identical logic:

• the interpreted path in doExecute, both the null-aware anti-join branch and the regular branch, and
• the generated code in prepareBroadcast, which re-emits the two statements into every BroadcastHashJoinExec whole-stage-codegen stage.

This PR extracts that type-independent setup into a shared helper BroadcastHashJoinExec.buildReadOnlyRelation(broadcast): HashedRelation and calls it from all three paths. The generated relation initializer shrinks from two statements to a single static call.

Before (generated code, per BroadcastHashJoinExec stage):

relation = ((org.apache.spark.sql.execution.joins.LongHashedRelation) references[N] /* broadcast */
  .value()).asReadOnlyCopy();
incPeakExecutionMemory(relation.estimatedSize());

After:

relation = (org.apache.spark.sql.execution.joins.LongHashedRelation)
  org.apache.spark.sql.execution.joins.BroadcastHashJoinExec.buildReadOnlyRelation(
    references[N] /* broadcast */);

The helper returns the HashedRelation supertype, so the generated code casts the result back to the concrete relation class; the mutable-state field keeps its precise static type for the hot-path getValue/get dispatch, exactly as before.

This follows the same "extract type-independent generated machinery into a compiled helper" pattern as SPARK-57909 (ColumnarToRowExec.advanceBatch), and additionally de-duplicates the two interpreted doExecute copies.

Measured with WholeStageCodegenSizeBenchmark (added under SPARK-57915; plans all 135 TPC-DS queries over empty tables), current master vs. this change:

metric	master	this PR	delta
constant pool, summed over stages	432,881	430,503	-0.55%
max method bytecode, summed over stages	841,185	841,161	~unchanged
source code size (chars)	21,846,001	21,846,031	~unchanged
inner classes	258	258	unchanged
codegen fallbacks	0	0	unchanged

The reduction shows up in the constant pool, since the asReadOnlyCopy / estimatedSize / incPeakExecutionMemory method references are now emitted once instead of per stage. max method bytecode is unchanged because the relation initializer is not a stage's largest method, and the source-char count is flat because the fully-qualified helper call is about as long as the two inline statements it replaces -- the win is in the compiled constant pool, which is what gates the 64KB method / constant-pool limits.

Why are the changes needed?

This is part of the umbrella SPARK-56908 (reduce the size of code generated by whole-stage codegen). The relation setup is real bytecode and constant-pool method-references that Janino cannot fold away; collapsing it to a single helper call compiles it once per JVM instead of re-emitting it into every BroadcastHashJoinExec stage. It also removes two copies of the interpreted-path logic, so the paths can no longer drift.

Does this PR introduce any user-facing change?

No. The helper performs the same asReadOnlyCopy and peak-memory bookkeeping in the same order as the previous code; this is a pure code-organization change with no behavioral difference.

How was this patch tested?

Existing tests, run with whole-stage codegen both on and off:

• OuterJoinSuite, InnerJoinSuite, ExistenceJoinSuite (the join operators' correctness across build sides and codegen on/off), and
• BroadcastJoinSuite (broadcast build path, null-aware anti join, and metrics).

No behavior changes, so no new tests are added; the extracted setup is covered by the existing broadcast-join suites on both the interpreted and generated paths.

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

Generated-by: Claude Code (Opus 4.8)

[LuciferYang]

cc @gengliangwang

[uros-b]

Thank you @LuciferYang!

[LuciferYang]

Merge Summary:

• merged into master cf2a6f5
• merged into branch-4.x ebb2e9b

Posted by merge_spark_pr.py