[SPARK-56551][SQL] Add operation metrics for DELETE queries in DSv2

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

@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.connector.write;
+
+import org.apache.spark.annotation.Evolving;
+
+/**
+ * Provides an informational summary of the DELETE operation producing write.
+ *
+ * @since 4.2.0
+ */
+@Evolving
+public interface DeleteSummary extends WriteSummary {
+
+  /**
+   * Returns the number of rows deleted, or -1 if not found.
+   */
+  long numDeletedRows();
+
+  /**
+   * Returns the number of rows copied unmodified, or -1 if not found.
+   */
+  long numCopiedRows();
+}

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

@@ -0,0 +1,27 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.connector.write
+
+/**
+ * Implementation of [[DeleteSummary]] that provides DELETE operation summary.
+ */
+private[sql] case class DeleteSummaryImpl(
+    numDeletedRows: Long,
+    numCopiedRows: Long)
+  extends DeleteSummary {
+}

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

@@ -31,7 +31,7 @@ import org.apache.spark.sql.catalyst.util.RowDeltaUtils.{COPY_OPERATION, DELETE_
 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, RowLevelOperation, UpdateSummaryImpl, Write, WriterCommitMessage, WriteSummary}
+import org.apache.spark.sql.connector.write.{BatchWrite, DataWriter, DataWriterFactory, DeleteSummaryImpl, DeltaWrite, DeltaWriter, MergeSummaryImpl, PhysicalWriteInfoImpl, RowLevelOperation, RowLevelOperationTable, UpdateSummaryImpl, Write, WriterCommitMessage, WriteSummary}
 import org.apache.spark.sql.connector.write.RowLevelOperation.Command._
 import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryExecutionErrors}
 import org.apache.spark.sql.execution.{QueryExecution, SparkPlan, SQLExecution, UnaryExecNode}
@@ -334,6 +334,26 @@ case class ReplaceDataExec(
   override protected def withNewChildInternal(newChild: SparkPlan): ReplaceDataExec = {
     copy(query = newChild)
   }
+
+  override protected def getWriteSummary(query: SparkPlan): Option[WriteSummary] = {
+    if (rowLevelCommand == DELETE) {
+      // DELETE ReplaceData plans filter out the deleted rows early in the plan, and they don't
+      // reach this node. We need to calculate this value as numScannedRows - numCopiedRows.
+      val numScannedRows = collectFirst(query) {
+        case b: BatchScanExec if b.table.isInstanceOf[RowLevelOperationTable] =>
+          getMetricValue(b.metrics, "numOutputRows")
+      }
+      val numCopiedRows = getMetricValue(metrics, "numCopiedRows")
+      val numDeletedRows = if (numScannedRows.exists(_ >= 0) && numCopiedRows >= 0) {
+        numScannedRows.get - numCopiedRows
+      } else {
+        // One of the metrics couldn't be found, also mark numDeletedRows as not found.
+        -1L
+      }
+      metrics("numDeletedRows").set(numDeletedRows)
+    }
+    super.getWriteSummary(query)
+  }
 }
 
 /**
@@ -426,13 +446,17 @@ trait RowLevelWriteExec extends V2ExistingTableWriteExec {
       Map(
         "numUpdatedRows" -> SQLMetrics.createMetric(sparkContext, "number of updated rows"),
         "numCopiedRows" -> SQLMetrics.createMetric(sparkContext, "number of copied rows"))
+    case DELETE =>
+      Map(
+        "numDeletedRows" -> SQLMetrics.createMetric(sparkContext, "number of deleted rows"),
+        "numCopiedRows" -> SQLMetrics.createMetric(sparkContext, "number of copied rows"))
     case _ => Map.empty
   }
 
   /**
    * Returns the value of the named metric, or -1 if the metric is not found.
    */
-  private def getMetricValue(metrics: Map[String, SQLMetric], name: String): Long = {
+  protected def getMetricValue(metrics: Map[String, SQLMetric], name: String): Long = {
     metrics.get(name).map(_.value).getOrElse(-1L)
   }
 
@@ -456,7 +480,9 @@ trait RowLevelWriteExec extends V2ExistingTableWriteExec {
           getMetricValue(operationMetrics, "numUpdatedRows"),
           getMetricValue(operationMetrics, "numCopiedRows")))
       case DELETE =>
-        None
+        Some(DeleteSummaryImpl(
+          getMetricValue(operationMetrics, "numDeletedRows"),
+          getMetricValue(operationMetrics, "numCopiedRows")))
     }
   }
 }
@@ -744,13 +770,15 @@ case class DeltaWritingSparkTask(
   override protected def write(
       writer: DeltaWriter[InternalRow], iter: java.util.Iterator[InternalRow]): Unit = {
     var numUpdatedRows = 0L
+    var numDeletedRows = 0L
 
     while (iter.hasNext) {
       val row = iter.next()
       val operation = row.getInt(0)
 
       operation match {
         case DELETE_OPERATION =>
+          numDeletedRows += 1L
           rowIdProjection.project(row)
           writer.delete(null, rowIdProjection)
 
@@ -777,6 +805,7 @@ case class DeltaWritingSparkTask(
     }
 
     operationMetrics.get("numUpdatedRows").foreach(_.add(numUpdatedRows))
+    operationMetrics.get("numDeletedRows").foreach(_.add(numDeletedRows))
   }
 }
 
@@ -792,13 +821,15 @@ case class DeltaWithMetadataWritingSparkTask(
   override protected def write(
       writer: DeltaWriter[InternalRow], iter: java.util.Iterator[InternalRow]): Unit = {
     var numUpdatedRows = 0L
+    var numDeletedRows = 0L
 
     while (iter.hasNext) {
       val row = iter.next()
       val operation = row.getInt(0)
 
       operation match {
         case DELETE_OPERATION =>
+          numDeletedRows += 1L
           rowIdProjection.project(row)
           metadataProjection.project(row)
           writer.delete(metadataProjection, rowIdProjection)
@@ -828,6 +859,7 @@ case class DeltaWithMetadataWritingSparkTask(
     }
 
     operationMetrics.get("numUpdatedRows").foreach(_.add(numUpdatedRows))
+    operationMetrics.get("numDeletedRows").foreach(_.add(numDeletedRows))
   }
 }