This is an automated email from the ASF dual-hosted git repository.
YannByron pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new b04db5a86a [spark] Support deleted record metrics for V2 delete (#8486)
b04db5a86a is described below
commit b04db5a86a82edc6f8d7e0cd61cc25ea07aa7b2b
Author: Kerwin Zhang <[email protected]>
AuthorDate: Fri Jul 10 10:35:34 2026 +0800
[spark] Support deleted record metrics for V2 delete (#8486)
---
.../org/apache/paimon/spark/PaimonMetrics.scala | 10 +++++++
.../paimon/spark/write/PaimonBatchWriteBase.scala | 34 +++++++++++++++++++---
.../apache/paimon/spark/write/PaimonV2Write.scala | 7 ++++-
.../apache/paimon/spark/sql/PaimonMetricTest.scala | 2 ++
4 files changed, 48 insertions(+), 5 deletions(-)
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonMetrics.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonMetrics.scala
index c8ceba357b..d3d2a39364 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonMetrics.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonMetrics.scala
@@ -40,6 +40,7 @@ object PaimonMetrics {
val ADDED_TABLE_FILES = "addedTableFiles"
val DELETED_TABLE_FILES = "deletedTableFiles"
val INSERTED_RECORDS = "insertedRecords"
+ val DELETED_RECORDS = "deletedRecords"
val APPENDED_CHANGELOG_FILES = "appendedChangelogFiles"
val PARTITIONS_WRITTEN = "partitionsWritten"
val BUCKETS_WRITTEN = "bucketsWritten"
@@ -211,6 +212,15 @@ case class PaimonInsertedRecordsTaskMetric(value: Long)
extends PaimonTaskMetric
override def name(): String = PaimonMetrics.INSERTED_RECORDS
}
+case class PaimonDeletedRecordsMetric() extends PaimonSumMetric {
+ override def name(): String = PaimonMetrics.DELETED_RECORDS
+ override def description(): String = "number of deleted records"
+}
+
+case class PaimonDeletedRecordsTaskMetric(value: Long) extends
PaimonTaskMetric {
+ override def name(): String = PaimonMetrics.DELETED_RECORDS
+}
+
case class PaimonAppendedChangelogFilesMetric() extends PaimonSumMetric {
override def name(): String = PaimonMetrics.APPENDED_CHANGELOG_FILES
override def description(): String = "number of appended changelog files"
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonBatchWriteBase.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonBatchWriteBase.scala
index e925cad76b..b13cea4b3b 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonBatchWriteBase.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonBatchWriteBase.scala
@@ -20,7 +20,7 @@ package org.apache.paimon.spark.write
import org.apache.paimon.Snapshot
import org.apache.paimon.io.{CompactIncrement, DataFileMeta, DataIncrement}
-import org.apache.paimon.spark.SparkTypeUtils
+import org.apache.paimon.spark.{PaimonDeletedRecordsTaskMetric, SparkTypeUtils}
import org.apache.paimon.spark.catalyst.Compatibility
import org.apache.paimon.spark.commands.SparkDataFileMeta
import org.apache.paimon.spark.metric.SparkMetricRegistry
@@ -30,6 +30,7 @@ import org.apache.paimon.table.{FileStoreTable, SpecialFields}
import org.apache.paimon.table.sink.{BatchWriteBuilder, CommitMessage,
CommitMessageImpl}
import org.apache.spark.sql.PaimonSparkSession
+import org.apache.spark.sql.connector.metric.CustomTaskMetric
import org.apache.spark.sql.connector.write.{DataWriterFactory,
PhysicalWriteInfo, WriterCommitMessage}
import org.apache.spark.sql.execution.SQLExecution
import org.apache.spark.sql.execution.metric.SQLMetrics
@@ -132,10 +133,35 @@ abstract class PaimonBatchWriteBase(
} finally {
batchTableCommit.close()
}
- postDriverMetrics()
+ postDriverMetrics(deletedRecordsTaskMetric(operation, addCommitMessage,
deletedCommitMessage))
postCommit(commitMessages)
}
+ /**
+ * For copy-on-write DELETE, the number of deleted records is exactly the
row count of the removed
+ * files minus the row count of the rewritten files. This does not hold for
UPDATE and MERGE,
+ * whose rewritten files mix copied rows with modified rows.
+ */
+ private def deletedRecordsTaskMetric(
+ operation: Snapshot.Operation,
+ addedMessages: Seq[CommitMessage],
+ deletedMessages: Seq[CommitMessage]): Array[CustomTaskMetric] = {
+ if (copyOnWriteScan.isEmpty || operation != Snapshot.Operation.DELETE) {
+ return Array.empty
+ }
+ val addedRecords = addedMessages
+ .collect { case m: CommitMessageImpl => m }
+ .flatMap(_.newFilesIncrement().newFiles().asScala)
+ .map(_.rowCount())
+ .sum
+ val deletedRecords = deletedMessages
+ .collect { case m: CommitMessageImpl => m }
+ .flatMap(_.newFilesIncrement().deletedFiles().asScala)
+ .map(_.rowCount())
+ .sum
+ Array(PaimonDeletedRecordsTaskMetric(deletedRecords - addedRecords))
+ }
+
protected def abortMessages(messages: Array[WriterCommitMessage]): Unit = {
if (commitStarted) {
logWarning(s"Skip abort cleanup for table ${table.name()} because commit
has already started")
@@ -154,10 +180,10 @@ abstract class PaimonBatchWriteBase(
// Spark support v2 write driver metrics since 4.0, see
https://github.com/apache/spark/pull/48573
// To ensure compatibility with 3.x, manually post driver metrics here
instead of using Spark's API.
- protected def postDriverMetrics(): Unit = {
+ protected def postDriverMetrics(extraMetrics: Array[CustomTaskMetric] =
Array.empty): Unit = {
val spark = PaimonSparkSession.active
// todo: find a more suitable way to get metrics.
- val commitMetrics = metricRegistry.buildSparkCommitMetrics()
+ val commitMetrics = metricRegistry.buildSparkCommitMetrics() ++
extraMetrics
val executionId =
spark.sparkContext.getLocalProperty(SQLExecution.EXECUTION_ID_KEY)
val executionMetrics = Compatibility.getExecutionMetrics(spark,
executionId.toLong).distinct
val metricUpdates = executionMetrics.flatMap {
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonV2Write.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonV2Write.scala
index 6dd02172a6..a6b49a7918 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonV2Write.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonV2Write.scala
@@ -84,11 +84,16 @@ class PaimonV2Write(
PaimonAddedTableFilesMetric()
)
if (copyOnWriteScan.isEmpty) {
- // todo: support record metrics for row level ops
buffer += PaimonInsertedRecordsMetric()
}
if (copyOnWriteScan.nonEmpty) {
buffer += PaimonDeletedTableFilesMetric()
+ // For DELETE, the number of deleted records is exactly the row count
difference between
+ // the removed files and the rewritten files. For UPDATE and MERGE, the
rewritten files
+ // mix copied rows with modified rows, so record metrics cannot be
derived from file stats.
+ if (operationType.contains(Snapshot.Operation.DELETE)) {
+ buffer += PaimonDeletedRecordsMetric()
+ }
}
if (!coreOptions.changelogProducer().equals(ChangelogProducer.NONE)) {
buffer += PaimonAppendedChangelogFilesMetric()
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PaimonMetricTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PaimonMetricTest.scala
index c2acc7e928..d2e23d097d 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PaimonMetricTest.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PaimonMetricTest.scala
@@ -185,12 +185,14 @@ class PaimonMetricTest extends PaimonSparkTestBase with
ScanPlanHelper {
val executionMetrics = commandExecutionMetrics(metrics)
assert(executionMetrics(metrics("addedTableFiles").id) == "1")
assert(executionMetrics(metrics("deletedTableFiles").id) == "1")
+ assert(executionMetrics(metrics("deletedRecords").id) == "3")
val df1 = sql("DELETE FROM T WHERE id > 0")
val metrics1 = commandMetrics(df1)
val executionMetrics1 = commandExecutionMetrics(metrics1)
assert(executionMetrics1(metrics1("addedTableFiles").id) == "0")
assert(executionMetrics1(metrics1("deletedTableFiles").id) == "1")
+ assert(executionMetrics1(metrics1("deletedRecords").id) == "7")
}
}
