(spark) branch master updated: [SPARK-58163][SQL] Extract the BroadcastHashJoinExec read-only relation setup into a shared helper

[ya...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

LuciferYang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git


The following commit(s) were added to refs/heads/master by this push:
     new cf2a6f5d23e7 [SPARK-58163][SQL] Extract the BroadcastHashJoinExec read-only relation setup into a shared helper
cf2a6f5d23e7 is described below

commit cf2a6f5d23e7b0c2213697d873fd11a5f7b4c877
Author: YangJie <ya...@baidu.com>
AuthorDate: Tue Jul 21 14:18:49 2026 +0800

    [SPARK-58163][SQL] Extract the BroadcastHashJoinExec read-only relation setup into a shared helper
    
    ### 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):
    ```java
    relation = ((org.apache.spark.sql.execution.joins.LongHashedRelation) references[N] /* broadcast */
      .value()).asReadOnlyCopy();
    incPeakExecutionMemory(relation.estimatedSize());
    ```
    
    After:
    ```java
    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 meth [...]
    
    ### 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)
    
    Closes #57294 from LuciferYang/broadcasthashjoin-readonly-relation-helper.
    
    Authored-by: YangJie <ya...@baidu.com>
    Signed-off-by: yangjie01 <ya...@baidu.com>
---
 .../execution/joins/BroadcastHashJoinExec.scala    | 25 ++++++++++++++++------
 1 file changed, 18 insertions(+), 7 deletions(-)

diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastHashJoinExec.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastHashJoinExec.scala
index 2881aeac55d8..d8bcd01f256d 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastHashJoinExec.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastHashJoinExec.scala
@@ -131,8 +131,7 @@ case class BroadcastHashJoinExec private(
     val broadcastRelation = buildPlan.executeBroadcast[HashedRelation]()
     if (isNullAwareAntiJoin) {
       streamedPlan.execute().mapPartitionsInternal { streamedIter =>
-        val hashed = broadcastRelation.value.asReadOnlyCopy()
-        TaskContext.get().taskMetrics().incPeakExecutionMemory(hashed.estimatedSize)
+        val hashed = BroadcastHashJoinExec.buildReadOnlyRelation(broadcastRelation)
         if (hashed == EmptyHashedRelation) {
           streamedIter.map { row =>
             numOutputRows += 1
@@ -158,8 +157,7 @@ case class BroadcastHashJoinExec private(
       }
     } else {
       streamedPlan.execute().mapPartitions { streamedIter =>
-        val hashed = broadcastRelation.value.asReadOnlyCopy()
-        TaskContext.get().taskMetrics().incPeakExecutionMemory(hashed.estimatedSize)
+        val hashed = BroadcastHashJoinExec.buildReadOnlyRelation(broadcastRelation)
         join(streamedIter, hashed, numOutputRows)
       }
     }
@@ -195,11 +193,13 @@ case class BroadcastHashJoinExec private(
     val broadcast = ctx.addReferenceObj("broadcast", broadcastRelation)
     val clsName = broadcastRelation.value.getClass.getName
 
-    // Inline mutable state since not many join operations in a task
+    // Inline mutable state since not many join operations in a task.
+    // Cast the helper's HashedRelation result back to the concrete class so the field stays
+    // concretely typed for the hot-path getValue/get calls.
     val relationTerm = ctx.addMutableState(clsName, "relation",
       v => s"""
-         | $v = (($clsName) $broadcast.value()).asReadOnlyCopy();
-         | incPeakExecutionMemory($v.estimatedSize());
+         | $v = ($clsName) org.apache.spark.sql.execution.joins.BroadcastHashJoinExec
+         |   .buildReadOnlyRelation($broadcast);
        """.stripMargin, forceInline = true)
     (broadcastRelation, relationTerm)
   }
@@ -252,6 +252,17 @@ case class BroadcastHashJoinExec private(
 }
 
 object BroadcastHashJoinExec extends JoinSelectionHelper {
+  /**
+   * Returns a read-only copy of the broadcast relation and records its size as peak execution
+   * memory. Shared by `doExecute` and the generated code of `prepareBroadcast`. This is called by
+   * generated Java class, should be public.
+   */
+  def buildReadOnlyRelation(broadcast: Broadcast[HashedRelation]): HashedRelation = {
+    val relation = broadcast.value.asReadOnlyCopy()
+    TaskContext.get().taskMetrics().incPeakExecutionMemory(relation.estimatedSize)
+    relation
+  }
+
   def apply(
       leftKeys: Seq[Expression],
       rightKeys: Seq[Expression],


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org