[SPARK-54310][SQL] Add numSourceRows metric for MergeIntoExec

### What changes were proposed in this pull request?

Add `numSourceRows` metric for `MergeIntoExec`, from source node's `numOutputRows`.

Assumption is that all child nodes have `numOutputRows`. If not found, `numSourceRows` would be -1.

### Why are the changes needed?

Improve completeness and debuggability of Merge Into metrics.

### Does this PR introduce _any_ user-facing change?

No.

### How was this patch tested?

Unit test cases for numSourceNodes metric.

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

No.

Closes #52669 from asl3/merge-numsourcerows.

Authored-by: Amanda Liu <amanda.liu@databricks.com>
Signed-off-by: Dongjoon Hyun <dongjoon@apache.org>

Author: asl3

sql/catalyst/src/main/java/org/apache/spark/sql/connector/write/MergeSummary.java

@@ -27,6 +27,11 @@
 @Evolving
 public interface MergeSummary extends WriteSummary {
 
+  /**
+   * Returns the number of source rows.
+   */
+  long numSourceRows();
+
   /**
    * Returns the number of target rows copied unmodified because they did not match any action,
    * or -1 if not found.

sql/catalyst/src/main/scala/org/apache/spark/sql/connector/write/MergeSummaryImpl.scala

@@ -21,6 +21,7 @@ package org.apache.spark.sql.connector.write
  * Implementation of [[MergeSummary]] that provides MERGE operation summary.
  */
 private[sql] case class MergeSummaryImpl(
+    numSourceRows: Long,
     numTargetRowsCopied: Long,
     numTargetRowsDeleted: Long,
     numTargetRowsUpdated: Long,

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/WriteToDataSourceV2Exec.scala

@@ -31,10 +31,11 @@ import org.apache.spark.sql.catalyst.util.RowDeltaUtils.{DELETE_OPERATION, INSER
 import org.apache.spark.sql.connector.catalog.{CatalogV2Util, Column, Identifier, StagedTable, StagingTableCatalog, Table, TableCatalog, TableInfo, TableWritePrivilege}
 import org.apache.spark.sql.connector.expressions.Transform
 import org.apache.spark.sql.connector.metric.CustomMetric
-import org.apache.spark.sql.connector.write.{BatchWrite, DataWriter, DataWriterFactory, DeltaWrite, DeltaWriter, MergeSummaryImpl, PhysicalWriteInfoImpl, Write, WriterCommitMessage, WriteSummary}
+import org.apache.spark.sql.connector.write.{BatchWrite, DataWriter, DataWriterFactory, DeltaWrite, DeltaWriter, MergeSummaryImpl, PhysicalWriteInfoImpl, RowLevelOperationTable, Write, WriterCommitMessage, WriteSummary}
 import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryExecutionErrors}
 import org.apache.spark.sql.execution.{SparkPlan, SQLExecution, UnaryExecNode}
 import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
+import org.apache.spark.sql.execution.joins.BaseJoinExec
 import org.apache.spark.sql.execution.metric.{CustomMetrics, SQLMetric, SQLMetrics}
 import org.apache.spark.sql.types.StructType
 import org.apache.spark.util.{LongAccumulator, Utils}
@@ -481,7 +482,9 @@ trait V2TableWriteExec extends V2CommandExec with UnaryExecNode with AdaptiveSpa
   private def getWriteSummary(query: SparkPlan): Option[WriteSummary] = {
     collectFirst(query) { case m: MergeRowsExec => m }.map { n =>
       val metrics = n.metrics
+      val numSourceRows = getNumSourceRows(n)
       MergeSummaryImpl(
+        numSourceRows,
         metrics.get("numTargetRowsCopied").map(_.value).getOrElse(-1L),
         metrics.get("numTargetRowsDeleted").map(_.value).getOrElse(-1L),
         metrics.get("numTargetRowsUpdated").map(_.value).getOrElse(-1L),
@@ -493,6 +496,40 @@ trait V2TableWriteExec extends V2CommandExec with UnaryExecNode with AdaptiveSpa
       )
     }
   }
+
+  private def getNumSourceRows(mergeRowsExec: MergeRowsExec): Long = {
+    def hasTargetTable(plan: SparkPlan): Boolean = {
+      collectFirst(plan) {
+        case scan @ BatchScanExec(_, _, _, _, _: RowLevelOperationTable, _) => scan
+      }.isDefined
+    }
+
+    def findSourceScan(join: BaseJoinExec): Option[SparkPlan] = {
+      val leftHasTarget = hasTargetTable(join.left)
+      val rightHasTarget = hasTargetTable(join.right)
+
+      val sourceSide = if (leftHasTarget) {
+        Some(join.right)
+      } else if (rightHasTarget) {
+        Some(join.left)
+      } else {
+        None
+      }
+
+      sourceSide.flatMap { side =>
+        collectFirst(side) {
+          case source if source.metrics.contains("numOutputRows") =>
+          source
+        }
+      }
+    }
+
+    (for {
+      join <- collectFirst(mergeRowsExec.child) { case j: BaseJoinExec => j }
+      sourceScan <- findSourceScan(join)
+      metric <- sourceScan.metrics.get("numOutputRows")
+    } yield metric.value).getOrElse(-1L)
+  }
 }
 
 trait WritingSparkTask[W <: DataWriter[InternalRow]] extends Logging with Serializable {