(spark) branch master updated: [SPARK-52962][SQL] BroadcastExchangeExec should not reset metrics

[vi...@apache.org]

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

viirya 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 a823f95c5220 [SPARK-52962][SQL] BroadcastExchangeExec should not reset metrics
a823f95c5220 is described below

commit a823f95c5220bfed2de4c30b630cd04f409edd76
Author: Liang-Chi Hsieh <vi...@gmail.com>
AuthorDate: Sat Jul 26 09:11:38 2025 -0700

    [SPARK-52962][SQL] BroadcastExchangeExec should not reset metrics
    
    ### What changes were proposed in this pull request?
    
    This patch implements a no-op `resetMetrics` method for `BroadcastExchangeExec`.
    
    ### Why are the changes needed?
    
    This was found during debugging test failures at https://github.com/apache/spark/pull/51623.
    
    A materialized broadcast exchange (i.e., `BroadcastExchangeExec`) operator will be trimmed by AQE's empty relation propagation rule to an empty local relation. That's is because its metrics was reset. `BroadcastExchangeExec` after materialized won't be materialized again, so its metrics won't be updated again.
    
    It is weird and inconsistent that a materialized broadcast exchange still hold broadcast value but its metrics are reset.
    
    ### Does this PR introduce _any_ user-facing change?
    
    No
    
    ### How was this patch tested?
    
    Unit test
    
    ### Was this patch authored or co-authored using generative AI tooling?
    
    No
    
    Closes #51673 from viirya/broadcast_resetmetrics.
    
    Authored-by: Liang-Chi Hsieh <vi...@gmail.com>
    Signed-off-by: Liang-Chi Hsieh <vi...@gmail.com>
---
 .../sql/execution/exchange/BroadcastExchangeExec.scala    |  6 ++++++
 .../spark/sql/execution/BroadcastExchangeSuite.scala      | 15 +++++++++++++++
 2 files changed, 21 insertions(+)

diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/BroadcastExchangeExec.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/BroadcastExchangeExec.scala
index c70ee637a248..7f69657988a4 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/BroadcastExchangeExec.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/BroadcastExchangeExec.scala
@@ -145,6 +145,12 @@ case class BroadcastExchangeExec(
     Statistics(dataSize, Some(rowCount))
   }
 
+  override def resetMetrics(): Unit = {
+    // no-op
+    // BroadcastExchangeExec after materialized won't be materialized again, so we should not
+    // reset the metrics. Otherwise, we will lose the metrics collected in the broadcast job.
+  }
+
   @transient
   private lazy val promise = Promise[broadcast.Broadcast[Any]]()
 
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/BroadcastExchangeSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/BroadcastExchangeSuite.scala
index 60a74a553bc4..8d6ee83f5e6b 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/BroadcastExchangeSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/BroadcastExchangeSuite.scala
@@ -98,6 +98,21 @@ class BroadcastExchangeSuite extends SparkPlanTest
       assert(joinDF.collect().length == 1)
     }
   }
+
+  test("SPARK-52962: broadcast exchange should not reset metrics") {
+    val df = spark.range(1).toDF()
+    val joinDF = df.join(broadcast(df), "id")
+    joinDF.collect()
+    val broadcastExchangeExec = collect(
+      joinDF.queryExecution.executedPlan) { case p: BroadcastExchangeExec => p }
+    assert(broadcastExchangeExec.size == 1, "one and only BroadcastExchangeExec")
+
+    val broadcastExchangeNode = broadcastExchangeExec.head
+    val metrics = broadcastExchangeNode.metrics
+    assert(metrics("numOutputRows").value == 1)
+    broadcastExchangeNode.resetMetrics()
+    assert(metrics("numOutputRows").value == 1)
+  }
 }
 
 // Additional tests run in 'local-cluster' mode.


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