This is an automated email from the ASF dual-hosted git repository.
JingsongLi 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 fa62f98a7f [spark] Support V2 delta-based row-level operations for
deletion-vector append tables (#8539)
fa62f98a7f is described below
commit fa62f98a7ffaa54098fc75b69e00da49e1d6620f
Author: Kerwin Zhang <[email protected]>
AuthorDate: Mon Jul 13 10:37:17 2026 +0800
[spark] Support V2 delta-based row-level operations for deletion-vector
append tables (#8539)
---
.../scala/org/apache/paimon/spark/SparkTable.scala | 2 +
.../scala/org/apache/paimon/spark/SparkTable.scala | 2 +
.../scala/org/apache/paimon/spark/SparkTable.scala | 2 +
.../paimon/spark/write/PaimonDeltaBatchWrite.scala | 50 ++++
.../apache/spark/sql/paimon/shims/Spark4Shim.scala | 10 +-
.../org/apache/paimon/spark/PaimonMetrics.scala | 10 +
.../apache/paimon/spark/PaimonSparkTableBase.scala | 19 +-
.../scala/org/apache/paimon/spark/SparkTable.scala | 49 +++-
.../spark/rowops/PaimonSparkDeltaOperation.scala | 84 ++++++
.../paimon/spark/schema/PaimonMetadataColumn.scala | 9 +-
.../paimon/spark/write/PaimonBatchWriteBase.scala | 26 +-
.../paimon/spark/write/PaimonDeltaWrite.scala | 77 ++++++
.../paimon/spark/write/PaimonDeltaWriteBase.scala | 225 +++++++++++++++
.../spark/write/PaimonDeltaWriteBuilder.scala | 43 +++
.../paimon/spark/write/PaimonDeltaWriter.scala | 150 ++++++++++
.../apache/paimon/spark/write/WriteHelper.scala | 21 ++
.../apache/spark/sql/paimon/shims/SparkShim.scala | 13 +
.../paimon/spark/sql/DeltaRowLevelOpsTest.scala | 302 +++++++++++++++++++++
.../apache/paimon/spark/sql/PaimonMetricTest.scala | 31 +++
.../paimon/spark/write/PaimonDeltaBatchWrite.scala | 49 ++++
.../apache/spark/sql/paimon/shims/Spark3Shim.scala | 10 +-
.../paimon/spark/write/PaimonDeltaBatchWrite.scala | 51 ++++
.../catalyst/analysis/PureAppendOnlyScope.scala | 22 +-
.../analysis/Spark41DeleteMetadataRestore.scala | 34 ++-
.../analysis/Spark41MergeIntoRewrite.scala | 191 +++++++++++--
.../analysis/Spark41UpdateTableRewrite.scala | 77 +++++-
.../apache/spark/sql/paimon/shims/Spark4Shim.scala | 10 +-
27 files changed, 1499 insertions(+), 70 deletions(-)
diff --git
a/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/paimon/spark/SparkTable.scala
b/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/paimon/spark/SparkTable.scala
index b36f8075f0..7f4295bd65 100644
---
a/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/paimon/spark/SparkTable.scala
+++
b/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/paimon/spark/SparkTable.scala
@@ -35,4 +35,6 @@ object SparkTable {
def of(table: Table): SparkTable = SparkTable(table)
private[spark] def supportsV2RowLevelOps(sparkTable: SparkTable): Boolean =
false
+
+ def supportsV2DeltaOps(sparkTable: SparkTable): Boolean = false
}
diff --git
a/paimon-spark/paimon-spark-3.3/src/main/scala/org/apache/paimon/spark/SparkTable.scala
b/paimon-spark/paimon-spark-3.3/src/main/scala/org/apache/paimon/spark/SparkTable.scala
index 57951ce9b2..c8751348e1 100644
---
a/paimon-spark/paimon-spark-3.3/src/main/scala/org/apache/paimon/spark/SparkTable.scala
+++
b/paimon-spark/paimon-spark-3.3/src/main/scala/org/apache/paimon/spark/SparkTable.scala
@@ -36,4 +36,6 @@ object SparkTable {
def of(table: Table): SparkTable = SparkTable(table)
private[spark] def supportsV2RowLevelOps(sparkTable: SparkTable): Boolean =
false
+
+ def supportsV2DeltaOps(sparkTable: SparkTable): Boolean = false
}
diff --git
a/paimon-spark/paimon-spark-3.4/src/main/scala/org/apache/paimon/spark/SparkTable.scala
b/paimon-spark/paimon-spark-3.4/src/main/scala/org/apache/paimon/spark/SparkTable.scala
index 3d50edfff1..c7d4e9156f 100644
---
a/paimon-spark/paimon-spark-3.4/src/main/scala/org/apache/paimon/spark/SparkTable.scala
+++
b/paimon-spark/paimon-spark-3.4/src/main/scala/org/apache/paimon/spark/SparkTable.scala
@@ -35,4 +35,6 @@ object SparkTable {
def of(table: Table): SparkTable = SparkTable(table)
private[spark] def supportsV2RowLevelOps(sparkTable: SparkTable): Boolean =
false
+
+ def supportsV2DeltaOps(sparkTable: SparkTable): Boolean = false
}
diff --git
a/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/write/PaimonDeltaBatchWrite.scala
b/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/write/PaimonDeltaBatchWrite.scala
new file mode 100644
index 0000000000..cb3d3c6ba0
--- /dev/null
+++
b/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/write/PaimonDeltaBatchWrite.scala
@@ -0,0 +1,50 @@
+/*
+ * 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.paimon.spark.write
+
+import org.apache.paimon.Snapshot
+import org.apache.paimon.table.FileStoreTable
+
+import org.apache.spark.sql.connector.write.{DeltaBatchWrite,
DeltaWriterFactory, PhysicalWriteInfo, WriterCommitMessage}
+import org.apache.spark.sql.types.StructType
+
+/**
+ * Spark 4.0 shadow of the `paimon-spark4-common` wrapper: identical source
compiled against Spark
+ * 4.0.2 so the mixed-in `BatchWrite` methods carry no Spark 4.1-only
`WriteSummary` signature. See
+ * [[PaimonDeltaWriteBase]].
+ */
+class PaimonDeltaBatchWrite(
+ table: FileStoreTable,
+ rowSchema: StructType,
+ rowIdSchema: StructType,
+ operationType: Snapshot.Operation,
+ readSnapshotId: Option[Long])
+ extends PaimonDeltaWriteBase(table, rowSchema, rowIdSchema, operationType,
readSnapshotId)
+ with DeltaBatchWrite
+ with Serializable {
+
+ override def createBatchWriterFactory(info: PhysicalWriteInfo):
DeltaWriterFactory =
+ createPaimonDeltaWriterFactory(info)
+
+ override def useCommitCoordinator(): Boolean = false
+
+ override def commit(messages: Array[WriterCommitMessage]): Unit =
commitMessages(messages)
+
+ override def abort(messages: Array[WriterCommitMessage]): Unit =
abortMessages(messages)
+}
diff --git
a/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala
b/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala
index b92411d8bd..e3046f564f 100644
---
a/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala
+++
b/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala
@@ -25,7 +25,7 @@ import
org.apache.paimon.spark.catalyst.parser.extensions.PaimonSpark4SqlExtensi
import org.apache.paimon.spark.data.{Spark4ArrayData, Spark4InternalRow,
Spark4InternalRowWithBlob, SparkArrayData, SparkInternalRow}
import org.apache.paimon.spark.format.FormatTableBatchWrite
import org.apache.paimon.spark.rowops.PaimonCopyOnWriteScan
-import org.apache.paimon.spark.write.PaimonBatchWrite
+import org.apache.paimon.spark.write.{PaimonBatchWrite, PaimonDeltaBatchWrite}
import org.apache.paimon.table.{FileStoreTable, FormatTable}
import org.apache.paimon.types.{DataType, RowType}
@@ -220,6 +220,14 @@ class Spark4Shim extends SparkShim {
copyOnWriteScan,
operationType)
+ override def createPaimonDeltaBatchWrite(
+ table: FileStoreTable,
+ rowSchema: StructType,
+ rowIdSchema: StructType,
+ operationType: Snapshot.Operation,
+ readSnapshotId: Option[Long]): BatchWrite =
+ new PaimonDeltaBatchWrite(table, rowSchema, rowIdSchema, operationType,
readSnapshotId)
+
override def createFormatTableBatchWrite(
table: FormatTable,
overwriteDynamic: Option[Boolean],
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 d3d2a39364..85e639a842 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 UPDATED_RECORDS = "updatedRecords"
val DELETED_RECORDS = "deletedRecords"
val APPENDED_CHANGELOG_FILES = "appendedChangelogFiles"
val PARTITIONS_WRITTEN = "partitionsWritten"
@@ -212,6 +213,15 @@ case class PaimonInsertedRecordsTaskMetric(value: Long)
extends PaimonTaskMetric
override def name(): String = PaimonMetrics.INSERTED_RECORDS
}
+case class PaimonUpdatedRecordsMetric() extends PaimonSumMetric {
+ override def name(): String = PaimonMetrics.UPDATED_RECORDS
+ override def description(): String = "number of updated records"
+}
+
+case class PaimonUpdatedRecordsTaskMetric(value: Long) extends
PaimonTaskMetric {
+ override def name(): String = PaimonMetrics.UPDATED_RECORDS
+}
+
case class PaimonDeletedRecordsMetric() extends PaimonSumMetric {
override def name(): String = PaimonMetrics.DELETED_RECORDS
override def description(): String = "number of deleted records"
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonSparkTableBase.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonSparkTableBase.scala
index 6be314cee8..d42d129925 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonSparkTableBase.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonSparkTableBase.scala
@@ -133,10 +133,25 @@ abstract class PaimonSparkTableBase(val table: Table)
_metadataColumns.append(PaimonMetadataColumn.SEARCH_SCORE)
}
+ // For tables on the delta-based row-level path these two columns form the
delta row ID, and
+ // Spark rejects nullable row ID columns. Keep them nullable everywhere
else: V1 commands
+ // project them through outer joins where the target side can be null
(e.g. the not-matched
+ // rows of MERGE INTO).
+ val (filePathColumn, rowIndexColumn) =
+ if (
+ this.isInstanceOf[SparkTable] &&
+ SparkTable.supportsV2DeltaOps(this.asInstanceOf[SparkTable])
+ ) {
+ (
+ PaimonMetadataColumn.FILE_PATH.copy(nullable = false),
+ PaimonMetadataColumn.ROW_INDEX.copy(nullable = false))
+ } else {
+ (PaimonMetadataColumn.FILE_PATH, PaimonMetadataColumn.ROW_INDEX)
+ }
_metadataColumns.appendAll(
Seq(
- PaimonMetadataColumn.FILE_PATH,
- PaimonMetadataColumn.ROW_INDEX,
+ filePathColumn,
+ rowIndexColumn,
PaimonMetadataColumn.PARTITION(partitionType),
PaimonMetadataColumn.BUCKET
))
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/SparkTable.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/SparkTable.scala
index 9ea20de190..2dd70920cd 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/SparkTable.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/SparkTable.scala
@@ -19,9 +19,9 @@
package org.apache.paimon.spark
import org.apache.paimon.spark.read.ObjectTableScanBuilder
-import org.apache.paimon.spark.rowops.PaimonSparkCopyOnWriteOperation
+import org.apache.paimon.spark.rowops.{PaimonSparkCopyOnWriteOperation,
PaimonSparkDeltaOperation}
import org.apache.paimon.table.`object`.ObjectTable
-import org.apache.paimon.table.{FileStoreTable, Table}
+import org.apache.paimon.table.{BucketMode, FileStoreTable, Table}
import org.apache.spark.sql.connector.catalog.{SupportsRead,
SupportsRowLevelOperations, TableCapability}
import org.apache.spark.sql.connector.read.ScanBuilder
@@ -40,15 +40,17 @@ import java.util.{EnumSet => JEnumSet, Set => JSet}
* If this base class implemented `SupportsRowLevelOperations`, Spark 4.1
would immediately call
* `newRowLevelOperationBuilder` on tables whose V2 write is disabled (e.g.
dynamic bucket or
* primary-key tables that fall back to V1 write) and fail before Paimon has a
chance to rewrite the
- * plan to a V1 command. Likewise, deletion-vector, data-evolution, and
fixed-length CHAR tables
- * need to stay on Paimon's V1 postHoc path even when `useV2Write=true`, so
they must also not
- * expose `SupportsRowLevelOperations`.
+ * plan to a V1 command. Likewise, data-evolution and fixed-length CHAR tables
need to stay on
+ * Paimon's V1 postHoc path even when `useV2Write=true`, so they must also not
expose
+ * `SupportsRowLevelOperations`.
*
* Tables that DO support V2 row-level operations use the
[[SparkTableWithRowLevelOps]] subclass
* instead; the [[SparkTable.of]] factory picks the right variant via
* [[SparkTable.supportsV2RowLevelOps]]. Append-only tables, including
row-tracking-only tables,
* expose `SupportsRowLevelOperations` so DELETE, UPDATE, and MERGE INTO can
go through the V2
- * copy-on-write path when the table has no PK, deletion vectors, data
evolution, or CHAR columns.
+ * copy-on-write path when the table has no PK, deletion vectors, data
evolution, or CHAR columns;
+ * unaware-bucket deletion-vector append tables expose it so DELETE can go
through the delta path
+ * (see [[SparkTable.supportsV2DeltaOps]]).
*/
case class SparkTable(override val table: Table) extends
PaimonSparkTableBase(table)
@@ -64,7 +66,8 @@ class SparkTableWithRowLevelOps(tableArg: Table)
info: RowLevelOperationInfo): RowLevelOperationBuilder = {
table match {
case t: FileStoreTable =>
- () => new PaimonSparkCopyOnWriteOperation(t, info)
+ if (SparkTable.supportsV2DeltaOps(this)) { () => new
PaimonSparkDeltaOperation(t, info) }
+ else { () => new PaimonSparkCopyOnWriteOperation(t, info) }
case _ =>
throw new UnsupportedOperationException(
"Row-level write operation is only supported for FileStoreTable. " +
@@ -109,6 +112,10 @@ object SparkTable {
* `NoSuchMethodError` at the first DML statement.
*/
private[spark] def supportsV2RowLevelOps(sparkTable: SparkTable): Boolean = {
+ supportsV2CopyOnWriteOps(sparkTable) || supportsV2DeltaOps(sparkTable)
+ }
+
+ private def supportsV2CopyOnWriteOps(sparkTable: SparkTable): Boolean = {
if (org.apache.spark.SPARK_VERSION < "3.5") return false
if (!sparkTable.useV2Write) return false
sparkTable.getTable match {
@@ -123,6 +130,34 @@ object SparkTable {
case _ => false
}
}
+
+ /**
+ * Whether the table takes the delta-based row-level path
([[PaimonSparkDeltaOperation]]): an
+ * unaware-bucket append-only table with deletion vectors enabled. Bucketed
append tables are
+ * excluded because commit conflict detection does not cover concurrent
deletion-vector index
+ * changes of the same bucket (`IndexManifestFileHandler` overwrites the
whole bucket's index);
+ * row-tracking deletion-vector tables are excluded until the delta path
defines the row lineage
+ * semantics for rewritten rows.
+ *
+ * The Spark 3.2/3.3/3.4 shim `SparkTable` companions must expose a method
with this exact
+ * signature (hard-coded `false`), same as [[supportsV2RowLevelOps]]. Public
because the Spark 4.1
+ * rewrite scope (`PureAppendOnlyScope.targetsV2DeltaTable`) delegates to it
as the single source
+ * of truth for the delta gating conditions.
+ */
+ def supportsV2DeltaOps(sparkTable: SparkTable): Boolean = {
+ if (org.apache.spark.SPARK_VERSION < "3.5") return false
+ if (!sparkTable.useV2Write) return false
+ sparkTable.getTable match {
+ case fs: FileStoreTable =>
+ fs.primaryKeys().isEmpty &&
+ fs.bucketMode() == BucketMode.BUCKET_UNAWARE &&
+ sparkTable.coreOptions.deletionVectorsEnabled() &&
+ !sparkTable.coreOptions.rowTrackingEnabled() &&
+ !sparkTable.coreOptions.dataEvolutionEnabled() &&
+ !SparkTypeUtils.containsCharType(fs.rowType())
+ case _ => false
+ }
+ }
}
case class SparkIcebergTable(table: Table) extends BaseTable
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/rowops/PaimonSparkDeltaOperation.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/rowops/PaimonSparkDeltaOperation.scala
new file mode 100644
index 0000000000..f91d85a74d
--- /dev/null
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/rowops/PaimonSparkDeltaOperation.scala
@@ -0,0 +1,84 @@
+/*
+ * 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.paimon.spark.rowops
+
+import org.apache.paimon.CoreOptions
+import org.apache.paimon.Snapshot
+import org.apache.paimon.spark.PaimonScanBuilder
+import org.apache.paimon.spark.schema.PaimonMetadataColumn.{FILE_PATH_COLUMN,
ROW_INDEX_COLUMN}
+import org.apache.paimon.spark.write.PaimonDeltaWriteBuilder
+import org.apache.paimon.table.{FileStoreTable, InnerTable}
+
+import org.apache.spark.sql.connector.expressions.{Expressions, NamedReference}
+import org.apache.spark.sql.connector.read.ScanBuilder
+import org.apache.spark.sql.connector.write.{DeltaWriteBuilder,
LogicalWriteInfo, RowLevelOperation, RowLevelOperationInfo, SupportsDelta}
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+import java.util.{HashMap => JHashMap}
+
+/**
+ * A delta-based [[RowLevelOperation]] for deletion-vector enabled append
tables: instead of
+ * rewriting whole files like [[PaimonSparkCopyOnWriteOperation]], deleted
rows are marked in
+ * deletion vectors keyed by the `(__paimon_file_path, __paimon_row_index)`
row ID.
+ *
+ * The scan side is a regular pushdown scan: Spark's runtime group filtering
only applies to
+ * group-based (`ReplaceData`) plans, delta plans converge on matching rows
through the rewrite
+ * condition and ordinary predicate pushdown.
+ */
+class PaimonSparkDeltaOperation(table: FileStoreTable, info:
RowLevelOperationInfo)
+ extends RowLevelOperation
+ with SupportsDelta {
+
+ // Captured before the scan is planned. The deletion-vector maintainer must
merge with the
+ // deletion vectors of the snapshot the job read, never a newer one:
initializing it from the
+ // commit-time latest snapshot would silently absorb concurrent
deletion-vector changes that
+ // commit conflict detection is supposed to reject (and an UPDATE/MERGE
could resurrect a
+ // concurrently deleted row as its new version). Same contract as the V1
commands' readSnapshot.
+ private val readSnapshotId: Option[Long] =
+ Option(table.snapshotManager().latestSnapshot()).map(_.id())
+
+ override def command(): RowLevelOperation.Command = info.command()
+
+ override def newScanBuilder(options: CaseInsensitiveStringMap): ScanBuilder
= {
+ // Pin the scan to the captured snapshot so the read, the deletion-vector
maintainer and the
+ // commit-time file mapping all observe the same version; a commit that
lands in between (e.g.
+ // a concurrent compaction) then surfaces as a commit conflict instead of
a spurious mismatch
+ // between the scanned files and the read snapshot.
+ val conf = new JHashMap[String, String](options.asCaseSensitiveMap())
+ readSnapshotId.foreach(id => conf.put(CoreOptions.SCAN_SNAPSHOT_ID.key(),
id.toString))
+ new PaimonScanBuilder(table.copy(conf).asInstanceOf[InnerTable])
+ }
+
+ override def newWriteBuilder(info: LogicalWriteInfo): DeltaWriteBuilder =
+ new PaimonDeltaWriteBuilder(
+ table,
+ info,
+ Snapshot.Operation.valueOf(command().toString),
+ readSnapshotId)
+
+ override def rowId(): Array[NamedReference] =
+ Array(Expressions.column(FILE_PATH_COLUMN),
Expressions.column(ROW_INDEX_COLUMN))
+
+ // Paimon cannot update rows in place: an update is a deletion-vector mark
on the old row plus
+ // an appended new row. Still answer false so that Spark plans a single
UPDATE operation per row
+ // (no Expand doubling); the delta writer decomposes it internally.
+ override def representUpdateAsDeleteAndInsert(): Boolean = false
+
+ override def requiredMetadataAttributes(): Array[NamedReference] =
Array.empty
+}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/schema/PaimonMetadataColumn.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/schema/PaimonMetadataColumn.scala
index 41c6cdaaaf..8dc90a7c4f 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/schema/PaimonMetadataColumn.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/schema/PaimonMetadataColumn.scala
@@ -32,9 +32,16 @@ case class PaimonMetadataColumn(
override val dataType: DataType,
preserveOnDelete: Boolean = true,
preserveOnUpdate: Boolean = true,
- preserveOnReinsert: Boolean = false)
+ preserveOnReinsert: Boolean = false,
+ nullable: Boolean = true)
extends PaimonMetadataColumnBase {
+ // Only affects the Spark `MetadataColumn` capability (thus the relation's
metadata output and
+ // the delta row ID nullability check). `toStructField` / `toAttribute` stay
nullable: V1
+ // commands project these columns through outer joins where the target side
can be null, e.g.
+ // the not-matched rows of MERGE INTO.
+ override def isNullable: Boolean = nullable
+
def toPaimonDataField: DataField = {
new DataField(id, name, SparkTypeUtils.toPaimonType(dataType));
}
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 b13cea4b3b..5ac1688223 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
@@ -21,7 +21,6 @@ package org.apache.paimon.spark.write
import org.apache.paimon.Snapshot
import org.apache.paimon.io.{CompactIncrement, DataFileMeta, DataIncrement}
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
import org.apache.paimon.spark.rowops.PaimonCopyOnWriteScan
@@ -29,11 +28,8 @@ import
org.apache.paimon.spark.schema.PaimonMetadataColumn.{FILE_PATH, ROW_ID, S
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
import org.apache.spark.sql.types.StructType
import java.util.Collections
@@ -133,7 +129,9 @@ abstract class PaimonBatchWriteBase(
} finally {
batchTableCommit.close()
}
- postDriverMetrics(deletedRecordsTaskMetric(operation, addCommitMessage,
deletedCommitMessage))
+ postDriverMetrics(
+ metricRegistry.buildSparkCommitMetrics() ++
+ deletedRecordsTaskMetric(operation, addCommitMessage,
deletedCommitMessage))
postCommit(commitMessages)
}
@@ -178,24 +176,6 @@ 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(extraMetrics: Array[CustomTaskMetric] =
Array.empty): Unit = {
- val spark = PaimonSparkSession.active
- // todo: find a more suitable way to get metrics.
- 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 {
- m =>
- commitMetrics.find(x =>
m.metricType.toLowerCase.contains(x.name.toLowerCase)) match {
- case Some(customTaskMetric) => Some((m.accumulatorId,
customTaskMetric.value()))
- case None => None
- }
- }
- SQLMetrics.postDriverMetricsUpdatedByValue(spark.sparkContext,
executionId, metricUpdates)
- }
-
private def buildDeletedCommitMessage(
deletedFiles: Seq[SparkDataFileMeta]): Seq[CommitMessage] = {
logInfo(s"[V2 Write] Building deleted commit message for
${deletedFiles.size} files")
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonDeltaWrite.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonDeltaWrite.scala
new file mode 100644
index 0000000000..2117c50fde
--- /dev/null
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonDeltaWrite.scala
@@ -0,0 +1,77 @@
+/*
+ * 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.paimon.spark.write
+
+import org.apache.paimon.Snapshot
+import org.apache.paimon.spark.{PaimonAddedTableFilesMetric,
PaimonCommitDurationMetric, PaimonDeletedRecordsMetric,
PaimonInsertedRecordsMetric, PaimonNumWritersMetric, PaimonUpdatedRecordsMetric}
+import org.apache.paimon.table.FileStoreTable
+
+import org.apache.spark.sql.connector.metric.CustomMetric
+import org.apache.spark.sql.connector.write.{DeltaBatchWrite, DeltaWrite}
+import org.apache.spark.sql.paimon.shims.SparkShimLoader
+import org.apache.spark.sql.types.StructType
+
+import scala.collection.mutable
+
+/**
+ * A [[DeltaWrite]] for deletion-vector enabled append tables. `toBatch` goes
through the shim so
+ * the `DeltaBatchWrite` mixin is compiled against the right Spark minor
version, see
+ * [[PaimonDeltaWriteBase]].
+ */
+class PaimonDeltaWrite(
+ table: FileStoreTable,
+ rowSchema: StructType,
+ rowIdSchema: StructType,
+ operationType: Snapshot.Operation,
+ readSnapshotId: Option[Long])
+ extends DeltaWrite {
+
+ override def toBatch: DeltaBatchWrite = {
+ // The shim factory is declared to return `BatchWrite` so that loading the
shim class on
+ // Spark < 3.4 runtimes never links against the delta write API; this
class itself is only
+ // loaded behind the `supportsV2DeltaOps` gate (Spark 3.5+).
+ SparkShimLoader.shim
+ .createPaimonDeltaBatchWrite(table, rowSchema, rowIdSchema,
operationType, readSnapshotId)
+ .asInstanceOf[DeltaBatchWrite]
+ }
+
+ override def supportedCustomMetrics(): Array[CustomMetric] = {
+ val buffer =
mutable.ArrayBuffer[CustomMetric](PaimonCommitDurationMetric())
+ operationType match {
+ case Snapshot.Operation.DELETE =>
+ buffer += PaimonDeletedRecordsMetric()
+ case Snapshot.Operation.UPDATE =>
+ buffer += PaimonAddedTableFilesMetric()
+ buffer += PaimonNumWritersMetric()
+ buffer += PaimonUpdatedRecordsMetric()
+ case Snapshot.Operation.MERGE =>
+ buffer += PaimonAddedTableFilesMetric()
+ buffer += PaimonNumWritersMetric()
+ buffer += PaimonDeletedRecordsMetric()
+ buffer += PaimonUpdatedRecordsMetric()
+ buffer += PaimonInsertedRecordsMetric()
+ case _ =>
+ }
+ buffer.toArray
+ }
+
+ override def toString: String =
s"PaimonDeltaWrite(table=${table.fullName()})"
+
+ override def description(): String = toString
+}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonDeltaWriteBase.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonDeltaWriteBase.scala
new file mode 100644
index 0000000000..3f1c6001fa
--- /dev/null
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonDeltaWriteBase.scala
@@ -0,0 +1,225 @@
+/*
+ * 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.paimon.spark.write
+
+import org.apache.paimon.Snapshot
+import org.apache.paimon.data.BinaryRow
+import org.apache.paimon.deletionvectors.DeletionVector
+import org.apache.paimon.deletionvectors.append.BaseAppendDeleteFileMaintainer
+import org.apache.paimon.fs.Path
+import org.apache.paimon.io.{CompactIncrement, DataFileMeta, DataIncrement}
+import org.apache.paimon.manifest.FileKind
+import org.apache.paimon.spark.PaimonMetrics
+import org.apache.paimon.spark.metric.SparkMetricRegistry
+import org.apache.paimon.spark.schema.PaimonMetadataColumn.{FILE_PATH_COLUMN,
ROW_INDEX_COLUMN}
+import org.apache.paimon.table.FileStoreTable
+import org.apache.paimon.table.sink.{BatchWriteBuilder, CommitMessage,
CommitMessageImpl}
+import org.apache.paimon.table.source.DataSplit
+
+import org.apache.spark.sql.connector.write.{DeltaWriterFactory,
PhysicalWriteInfo, WriterCommitMessage}
+import org.apache.spark.sql.types.StructType
+
+import java.util.Collections
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable
+
+/**
+ * Business logic for Paimon delta batch writes on deletion-vector enabled
append tables,
+ * deliberately *not* extending
`org.apache.spark.sql.connector.write.DeltaBatchWrite` for the same
+ * `BatchWrite.commit(.., WriteSummary)` cross-version reason documented on
+ * [[PaimonBatchWriteBase]]: per-version `paimon-spark{3,4}-common` modules
supply a thin
+ * `PaimonDeltaBatchWrite` wrapper that mixes in `DeltaBatchWrite`, and
`paimon-spark-4.0/src/main`
+ * shadows that wrapper at the 4.0.2 compile target.
+ *
+ * Deleted rows arrive as per-task per-file bitmaps and appended rows as
regular data commit
+ * messages (see [[PaimonDeltaWriter]]); this class merges the bitmaps across
tasks at commit time,
+ * merges the result with the files' existing deletion vectors through
+ * [[BaseAppendDeleteFileMaintainer]], and commits the data messages together
with index-only
+ * deletion-vector [[CommitMessage]]s.
+ */
+abstract class PaimonDeltaWriteBase(
+ val table: FileStoreTable,
+ val rowSchema: StructType,
+ val rowIdSchema: StructType,
+ operationType: Snapshot.Operation,
+ readSnapshotId: Option[Long])
+ extends WriteHelper
+ with Serializable {
+
+ protected val metricRegistry: SparkMetricRegistry = SparkMetricRegistry()
+
+ @volatile protected var commitStarted: Boolean = false
+
+ protected val batchWriteBuilder: BatchWriteBuilder =
table.newBatchWriteBuilder()
+
+ protected def createPaimonDeltaWriterFactory(info: PhysicalWriteInfo):
DeltaWriterFactory = {
+ PaimonDeltaWriterFactory(
+ batchWriteBuilder,
+ rowSchema,
+ coreOptions,
+ catalogContextForBlobDescriptor,
+ rowIdSchema.fieldIndex(FILE_PATH_COLUMN),
+ rowIdSchema.fieldIndex(ROW_INDEX_COLUMN)
+ )
+ }
+
+ protected def commitMessages(messages: Array[WriterCommitMessage]): Unit = {
+ commitStarted = true
+ logInfo(s"Committing delta write to table ${table.name()}")
+ val taskMessages = messages.filter(_ != null)
+ val dataMessages = WriteTaskResult.merge(taskMessages)
+ val mergedDvs = mergeTaskDeletionVectors(taskMessages)
+ val dvMessages = if (mergedDvs.isEmpty) Seq.empty else
buildDvCommitMessages(mergedDvs)
+ val commitMessages = dataMessages ++ dvMessages
+
+ val batchTableCommit = batchWriteBuilder.newCommit()
+ batchTableCommit.withMetricRegistry(metricRegistry)
+ batchTableCommit.withOperation(operationType)
+ try {
+ val start = System.currentTimeMillis()
+ batchTableCommit.commit(commitMessages.asJava)
+ logInfo(s"Committed in ${System.currentTimeMillis() - start} ms")
+ } finally {
+ batchTableCommit.close()
+ }
+ // Drop the commit-level insertedRecords gauge: it is derived from
+ // `CommitStats.deltaRecordsAppended` which mixes ADD and DELETE row
counts and would
+ // overwrite the exact per-row record metrics reported by the delta writer
tasks.
+ postDriverMetrics(
+ metricRegistry
+ .buildSparkCommitMetrics()
+ .filterNot(_.name() == PaimonMetrics.INSERTED_RECORDS))
+ postCommit(commitMessages)
+ }
+
+ protected def abortMessages(messages: Array[WriterCommitMessage]): Unit = {
+ if (commitStarted) {
+ logWarning(s"Skip abort cleanup for table ${table.name()} because commit
has already started")
+ return
+ }
+ // Deletion-vector index files are only written inside `commitMessages`,
so an abort before
+ // commit only needs to clean up the appended data files. If
`commitMessages` fails after the
+ // maintainer persisted the index files, they are left behind like any
data file of a failed
+ // commit and reclaimed by orphan-file cleanup.
+ logInfo(s"Aborting delta write to table ${table.name()}")
+ val batchTableCommit = batchWriteBuilder.newCommit()
+ try {
+ val dataMessages = WriteTaskResult.merge(messages.filter(_ != null))
+ batchTableCommit.abort(dataMessages.asJava)
+ } finally {
+ batchTableCommit.close()
+ }
+ }
+
+ /** Merge the per-task per-file bitmaps into one deletion vector per data
file. */
+ private def mergeTaskDeletionVectors(
+ messages: Array[WriterCommitMessage]): mutable.HashMap[String,
DeletionVector] = {
+ val merged = mutable.HashMap.empty[String, DeletionVector]
+ messages.foreach {
+ case result: PaimonDeltaWriteTaskResult =>
+ result.deletionVectors.foreach {
+ serialized =>
+ val dv =
DeletionVector.deserializeFromBytes(serialized.deletionVector)
+ merged.get(serialized.dataFilePath) match {
+ case Some(existing) => existing.merge(dv)
+ case None => merged.put(serialized.dataFilePath, dv)
+ }
+ }
+ case other =>
+ throw new IllegalStateException(s"Unexpected delta writer commit
message: $other")
+ }
+ merged
+ }
+
+ /**
+ * Persist the merged deletion vectors as index files and build index-only
commit messages. The
+ * maintainer merges the new deletions with the files' deletion vectors of
the snapshot captured
+ * BEFORE the scan (same contract as the V1 commands' readSnapshot): a
concurrent deletion-vector
+ * change of the same file then surfaces as a commit conflict instead of
being silently merged,
+ * which could otherwise resurrect a concurrently deleted row as the new
version of an UPDATE.
+ */
+ private def buildDvCommitMessages(
+ mergedDvs: mutable.HashMap[String, DeletionVector]): Seq[CommitMessage]
= {
+ val snapshot = readSnapshotId
+ .map {
+ id =>
+ try {
+ table.snapshotManager().snapshot(id)
+ } catch {
+ case e: Exception =>
+ throw new IllegalStateException(
+ s"Cannot load the read snapshot $id of table ${table.name()},
it may have " +
+ "expired during the write, please retry.",
+ e)
+ }
+ }
+ .getOrElse(throw new IllegalStateException(
+ s"Table ${table.name()} had no snapshot when the scan was planned but
rows were deleted."))
+ val fileNameToDv = mergedDvs.map { case (path, dv) => new
Path(path).getName -> dv }
+
+ val fileNameToPartition = mutable.HashMap.empty[String, BinaryRow]
+ val snapshotReader = table.newSnapshotReader().withSnapshot(snapshot)
+ if (coreOptions.manifestDeleteFileDropStats()) {
+ snapshotReader.dropStats()
+ }
+ snapshotReader.withDataFileNameFilter(fileName =>
fileNameToDv.contains(fileName))
+ snapshotReader.read().splits().asScala.foreach {
+ case split: DataSplit =>
+ split
+ .dataFiles()
+ .asScala
+ .foreach(f => fileNameToPartition.put(f.fileName(),
split.partition()))
+ case _ =>
+ }
+
+ val missingFiles = fileNameToDv.keySet.diff(fileNameToPartition.keySet)
+ if (missingFiles.nonEmpty) {
+ throw new IllegalStateException(
+ s"Data files $missingFiles of table ${table.name()} are missing in the
read snapshot " +
+ s"${snapshot.id()}, its manifests may have expired during the write,
please retry.")
+ }
+
+ fileNameToDv
+ .groupBy { case (fileName, _) => fileNameToPartition(fileName) }
+ .map {
+ case (partition, dvs) =>
+ val indexHandler = table.store().newIndexFileHandler()
+ val maintainer =
+ BaseAppendDeleteFileMaintainer.forUnawareAppend(indexHandler,
snapshot, partition)
+ dvs.foreach { case (fileName, dv) =>
maintainer.notifyNewDeletionVector(fileName, dv) }
+ val indexEntries = maintainer.persist()
+ val (added, deleted) = indexEntries.asScala.partition(_.kind() ==
FileKind.ADD)
+ new CommitMessageImpl(
+ maintainer.getPartition,
+ maintainer.getBucket,
+ null,
+ new DataIncrement(
+ Collections.emptyList[DataFileMeta],
+ Collections.emptyList[DataFileMeta],
+ Collections.emptyList[DataFileMeta],
+ added.map(_.indexFile).asJava,
+ deleted.map(_.indexFile).asJava
+ ),
+ CompactIncrement.emptyIncrement()
+ ): CommitMessage
+ }
+ .toSeq
+ }
+}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonDeltaWriteBuilder.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonDeltaWriteBuilder.scala
new file mode 100644
index 0000000000..6a459ccbf9
--- /dev/null
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonDeltaWriteBuilder.scala
@@ -0,0 +1,43 @@
+/*
+ * 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.paimon.spark.write
+
+import org.apache.paimon.Snapshot
+import org.apache.paimon.table.FileStoreTable
+
+import org.apache.spark.sql.connector.write.{DeltaWrite, DeltaWriteBuilder,
LogicalWriteInfo}
+
+/** A [[DeltaWriteBuilder]] for deletion-vector enabled append tables. */
+class PaimonDeltaWriteBuilder(
+ table: FileStoreTable,
+ info: LogicalWriteInfo,
+ operationType: Snapshot.Operation,
+ readSnapshotId: Option[Long])
+ extends DeltaWriteBuilder {
+
+ assert(info.rowIdSchema().isPresent, "Delta write requires a row ID schema.")
+
+ override def build(): DeltaWrite =
+ new PaimonDeltaWrite(
+ table,
+ info.schema(),
+ info.rowIdSchema().get(),
+ operationType,
+ readSnapshotId)
+}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonDeltaWriter.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonDeltaWriter.scala
new file mode 100644
index 0000000000..72bd73b846
--- /dev/null
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonDeltaWriter.scala
@@ -0,0 +1,150 @@
+/*
+ * 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.paimon.spark.write
+
+import org.apache.paimon.CoreOptions
+import org.apache.paimon.catalog.CatalogContext
+import org.apache.paimon.deletionvectors.{Bitmap64DeletionVector,
BitmapDeletionVector, DeletionVector}
+import org.apache.paimon.spark.{PaimonDeletedRecordsTaskMetric,
PaimonInsertedRecordsTaskMetric, PaimonUpdatedRecordsTaskMetric}
+import org.apache.paimon.table.sink.{BatchWriteBuilder, CommitMessage}
+
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.connector.metric.CustomTaskMetric
+import org.apache.spark.sql.connector.write.{DeltaWriter, DeltaWriterFactory,
WriterCommitMessage}
+import org.apache.spark.sql.types.StructType
+
+import scala.collection.mutable
+
+/** A deletion vector of one data file produced by a [[PaimonDeltaWriter]]
task. */
+case class SerializedDeletionVector(dataFilePath: String, deletionVector:
Array[Byte])
+
+/**
+ * The [[WriterCommitMessage]] of a [[PaimonDeltaWriter]] task: the commit
messages of the appended
+ * data files (inserted rows and the new versions of updated rows) plus the
per-file deletion
+ * bitmaps of the deleted rows.
+ */
+case class PaimonDeltaWriteTaskResult(
+ dataWriteResult: Option[InnerTableWriteTaskResult],
+ deletionVectors: Array[SerializedDeletionVector])
+ extends WriteTaskResult {
+
+ override def commitMessages(): Seq[CommitMessage] =
+ dataWriteResult.map(_.commitMessages()).getOrElse(Seq.empty)
+}
+
+case class PaimonDeltaWriterFactory(
+ writeBuilder: BatchWriteBuilder,
+ rowSchema: StructType,
+ coreOptions: CoreOptions,
+ catalogContext: CatalogContext,
+ filePathOrdinal: Int,
+ rowIndexOrdinal: Int)
+ extends DeltaWriterFactory {
+
+ override def createWriter(partitionId: Int, taskId: Long):
DeltaWriter[InternalRow] =
+ PaimonDeltaWriter(
+ writeBuilder,
+ rowSchema,
+ coreOptions,
+ catalogContext,
+ filePathOrdinal,
+ rowIndexOrdinal)
+}
+
+/**
+ * A [[DeltaWriter]] for deletion-vector enabled append tables. Deleted rows
are accumulated as
+ * per-file bitmaps keyed by the `__paimon_file_path` and `__paimon_row_index`
row ID; the driver
+ * merges the bitmaps across tasks and persists them as deletion-vector index
files at commit time,
+ * because Spark does not cluster the delta rows by file and several tasks may
delete rows of the
+ * same data file. Inserted rows and the new versions of updated rows go
through the regular append
+ * writer.
+ */
+case class PaimonDeltaWriter(
+ writeBuilder: BatchWriteBuilder,
+ rowSchema: StructType,
+ coreOptions: CoreOptions,
+ catalogContext: CatalogContext,
+ filePathOrdinal: Int,
+ rowIndexOrdinal: Int)
+ extends DeltaWriter[InternalRow] {
+
+ private val deletionVectors = mutable.HashMap.empty[String, DeletionVector]
+
+ private var deletedRecords = 0L
+ private var updatedRecords = 0L
+ private var insertedRecords = 0L
+
+ // Only UPDATE and MERGE INTO write rows; DELETE never initializes the
append writer.
+ private var appendWriter: Option[PaimonV2DataWriter] = None
+
+ private def getOrCreateAppendWriter: PaimonV2DataWriter = {
+ appendWriter.getOrElse {
+ val writer =
+ PaimonV2DataWriter(writeBuilder, rowSchema, rowSchema, coreOptions,
catalogContext)
+ appendWriter = Some(writer)
+ writer
+ }
+ }
+
+ private def markDeleted(id: InternalRow): Unit = {
+ val filePath = id.getUTF8String(filePathOrdinal).toString
+ val dv = deletionVectors.getOrElseUpdate(
+ filePath,
+ if (coreOptions.deletionVectorBitmap64()) new Bitmap64DeletionVector()
+ else new BitmapDeletionVector())
+ dv.delete(id.getLong(rowIndexOrdinal))
+ }
+
+ override def delete(metadata: InternalRow, id: InternalRow): Unit = {
+ markDeleted(id)
+ deletedRecords += 1
+ }
+
+ override def update(metadata: InternalRow, id: InternalRow, row:
InternalRow): Unit = {
+ markDeleted(id)
+ getOrCreateAppendWriter.write(row)
+ updatedRecords += 1
+ }
+
+ override def insert(row: InternalRow): Unit = {
+ getOrCreateAppendWriter.write(row)
+ insertedRecords += 1
+ }
+
+ override def commit(): WriterCommitMessage = {
+ val dataWriteResult = appendWriter.map(_.commit)
+ val serialized = deletionVectors.map {
+ case (filePath, dv) =>
+ SerializedDeletionVector(filePath, DeletionVector.serializeToBytes(dv))
+ }.toArray
+ PaimonDeltaWriteTaskResult(dataWriteResult, serialized)
+ }
+
+ override def abort(): Unit = appendWriter.foreach(_.abort())
+
+ override def close(): Unit = appendWriter.foreach(_.close())
+
+ override def currentMetricsValues(): Array[CustomTaskMetric] = {
+ Array(
+ PaimonDeletedRecordsTaskMetric(deletedRecords),
+ PaimonUpdatedRecordsTaskMetric(updatedRecords),
+ PaimonInsertedRecordsTaskMetric(insertedRecords)
+ ) ++
appendWriter.map(_.currentMetricsValues()).getOrElse(Array.empty[CustomTaskMetric])
+ }
+}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/WriteHelper.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/WriteHelper.scala
index df73bd35d0..c039bf80ed 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/WriteHelper.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/WriteHelper.scala
@@ -22,12 +22,17 @@ import org.apache.paimon.CoreOptions
import org.apache.paimon.CoreOptions.TagCreationMode
import org.apache.paimon.catalog.CatalogContext
import org.apache.paimon.partition.actions.PartitionMarkDoneAction
+import org.apache.paimon.spark.catalyst.Compatibility
import org.apache.paimon.table.FileStoreTable
import org.apache.paimon.table.sink.CommitMessage
import org.apache.paimon.tag.TagBatchCreation
import org.apache.paimon.utils.{BlobDescriptorUtils,
InternalRowPartitionComputer, PartitionPathUtils, PartitionStatisticsReporter,
TypeUtils}
import org.apache.spark.internal.Logging
+import org.apache.spark.sql.PaimonSparkSession
+import org.apache.spark.sql.connector.metric.CustomTaskMetric
+import org.apache.spark.sql.execution.SQLExecution
+import org.apache.spark.sql.execution.metric.SQLMetrics
import scala.collection.JavaConverters._
@@ -42,6 +47,22 @@ trait WriteHelper extends Logging {
table.catalogEnvironment().catalogContext(),
coreOptions.toConfiguration)
+ // 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.
+ def postDriverMetrics(commitMetrics: Array[CustomTaskMetric]): Unit = {
+ val spark = PaimonSparkSession.active
+ val executionId =
spark.sparkContext.getLocalProperty(SQLExecution.EXECUTION_ID_KEY)
+ val executionMetrics = Compatibility.getExecutionMetrics(spark,
executionId.toLong).distinct
+ val metricUpdates = executionMetrics.flatMap {
+ m =>
+ commitMetrics.find(x =>
m.metricType.toLowerCase.contains(x.name.toLowerCase)) match {
+ case Some(customTaskMetric) => Some((m.accumulatorId,
customTaskMetric.value()))
+ case None => None
+ }
+ }
+ SQLMetrics.postDriverMetricsUpdatedByValue(spark.sparkContext,
executionId, metricUpdates)
+ }
+
def postCommit(messages: Seq[CommitMessage]): Unit = {
if (messages.isEmpty) {
return
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/paimon/shims/SparkShim.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/paimon/shims/SparkShim.scala
index b925036463..85c923325d 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/paimon/shims/SparkShim.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/paimon/shims/SparkShim.scala
@@ -133,6 +133,19 @@ trait SparkShim {
copyOnWriteScan: Option[PaimonCopyOnWriteScan],
operationType: Option[Snapshot.Operation]): BatchWrite
+ /**
+ * Same `BatchWrite` mixin problem as [[createPaimonBatchWrite]], but for
the delta write of
+ * deletion-vector append tables. Declared to return `BatchWrite` instead of
`DeltaBatchWrite` so
+ * that loading the shim class on Spark runtimes without the delta write API
never links against
+ * it; the caller (only loaded behind the Spark 3.5+ delta gate) casts to
`DeltaBatchWrite`.
+ */
+ def createPaimonDeltaBatchWrite(
+ table: FileStoreTable,
+ rowSchema: StructType,
+ rowIdSchema: StructType,
+ operationType: Snapshot.Operation,
+ readSnapshotId: Option[Long]): BatchWrite
+
/** Same `BatchWrite` mixin problem as [[createPaimonBatchWrite]], but for
`FormatTable` writes. */
def createFormatTableBatchWrite(
table: FormatTable,
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DeltaRowLevelOpsTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DeltaRowLevelOpsTest.scala
new file mode 100644
index 0000000000..b7190502b9
--- /dev/null
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DeltaRowLevelOpsTest.scala
@@ -0,0 +1,302 @@
+/*
+ * 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.paimon.spark.sql
+
+import org.apache.paimon.Snapshot
+import org.apache.paimon.deletionvectors.{BitmapDeletionVector, DeletionVector}
+import org.apache.paimon.spark.PaimonSparkTestBase
+import org.apache.paimon.spark.schema.PaimonMetadataColumn.{FILE_PATH_COLUMN,
ROW_INDEX_COLUMN}
+import org.apache.paimon.spark.write.{PaimonDeltaBatchWrite,
PaimonDeltaWriteTaskResult, SerializedDeletionVector}
+
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.types.{LongType, StringType, StructField,
StructType}
+
+import scala.util.Random
+
+/**
+ * Tests for the delta-based (`SupportsDelta` / `WriteDelta`) DELETE, UPDATE
and MERGE INTO of
+ * deletion-vector enabled unaware-bucket append tables.
+ */
+class DeltaRowLevelOpsTest extends PaimonSparkTestBase {
+
+ private val v2WriteConf = "spark.paimon.write.use-v2-write" -> "true"
+
+ private def explainContains(statement: String): String = {
+ sql(s"EXPLAIN EXTENDED $statement").head().getString(0)
+ }
+
+ private def dataFilePaths(): Set[String] = {
+ sql("SELECT file_path FROM `T$files`").collect().map(_.getString(0)).toSet
+ }
+
+ private def createDvAppendTable(extraProps: String = ""): Unit = {
+ sql(s"""
+ |CREATE TABLE T (id INT, v INT)
+ |TBLPROPERTIES (
+ | 'deletion-vectors.enabled' = 'true',
+ | 'deletion-vectors.bitmap64' = '${Random.nextBoolean()}'
+ | $extraProps
+ |)
+ |""".stripMargin)
+ }
+
+ test("Paimon delta ops: basic delete goes through WriteDelta") {
+ withSparkSQLConf(v2WriteConf) {
+ createDvAppendTable()
+ // file 1
+ sql("INSERT INTO T SELECT /*+ REPARTITION(1) */ id, id FROM range(1,
1001)")
+ // file 2
+ sql("INSERT INTO T SELECT /*+ REPARTITION(1) */ id, id FROM range(1001,
2001)")
+
+ assert(explainContains("DELETE FROM T WHERE id <=
10").contains("WriteDelta"))
+
+ val filesBefore = dataFilePaths()
+ sql("DELETE FROM T WHERE id <= 10")
+ checkAnswer(sql("SELECT COUNT(*), MIN(id) FROM T"), Row(1990, 11))
+ // Delta delete marks rows in deletion vectors and must not rewrite data
files.
+ assert(dataFilePaths() === filesBefore)
+
+ // Delete rows of both files, merging with the existing deletion vector
of file 1.
+ sql("DELETE FROM T WHERE id IN (11, 1001)")
+ checkAnswer(sql("SELECT COUNT(*), MIN(id) FROM T"), Row(1988, 12))
+ checkAnswer(sql("SELECT COUNT(*) FROM T WHERE id = 1001"), Row(0))
+ assert(dataFilePaths() === filesBefore)
+
+ // Delete the remaining rows.
+ sql("DELETE FROM T WHERE id > 0")
+ checkAnswer(sql("SELECT COUNT(*) FROM T"), Row(0))
+ }
+ }
+
+ test("Paimon delta ops: delete with zero matched rows") {
+ withSparkSQLConf(v2WriteConf) {
+ createDvAppendTable()
+ sql("INSERT INTO T VALUES (1, 1), (2, 2)")
+ sql("DELETE FROM T WHERE id > 100")
+ checkAnswer(sql("SELECT * FROM T ORDER BY id"), Row(1, 1) :: Row(2, 2)
:: Nil)
+ }
+ }
+
+ test("Paimon delta ops: delete on partitioned table") {
+ withSparkSQLConf(v2WriteConf) {
+ sql(s"""
+ |CREATE TABLE T (id INT, v INT, pt STRING)
+ |PARTITIONED BY (pt)
+ |TBLPROPERTIES ('deletion-vectors.enabled' = 'true')
+ |""".stripMargin)
+ sql("INSERT INTO T VALUES (1, 1, 'a'), (2, 2, 'a'), (3, 3, 'b'), (4, 4,
'b')")
+
+ // A row-level condition across partitions goes through WriteDelta.
+ assert(explainContains("DELETE FROM T WHERE id % 2 =
0").contains("WriteDelta"))
+ sql("DELETE FROM T WHERE id % 2 = 0")
+ checkAnswer(sql("SELECT id, pt FROM T ORDER BY id"), Row(1, "a") ::
Row(3, "b") :: Nil)
+
+ // A partition-only (metadata only) condition falls back to the V1
command to drop whole
+ // files instead of writing deletion vectors.
+ assert(explainContains("DELETE FROM T WHERE pt =
'a'").contains("DeleteFromPaimonTable"))
+ sql("DELETE FROM T WHERE pt = 'a'")
+ checkAnswer(sql("SELECT id, pt FROM T"), Row(3, "b") :: Nil)
+ }
+ }
+
+ test("Paimon delta ops: basic update goes through WriteDelta") {
+ withSparkSQLConf(v2WriteConf) {
+ createDvAppendTable()
+ sql("INSERT INTO T SELECT /*+ REPARTITION(1) */ id, id FROM range(1,
1001)")
+
+ assert(explainContains("UPDATE T SET v = v + 1 WHERE id <=
10").contains("WriteDelta"))
+
+ val filesBefore = dataFilePaths()
+ sql("UPDATE T SET v = v + 1 WHERE id <= 10")
+ checkAnswer(sql("SELECT COUNT(*) FROM T"), Row(1000))
+ checkAnswer(sql("SELECT SUM(v) FROM T"), Row(500510))
+ checkAnswer(sql("SELECT v FROM T WHERE id = 1"), Row(2))
+ // The old rows are marked in deletion vectors, the new versions are
appended: the original
+ // data files must remain untouched.
+ assert(filesBefore.subsetOf(dataFilePaths()))
+
+ // Update rows whose current version lives in the appended file, merging
deletion vectors
+ // with the previous update.
+ sql("UPDATE T SET v = v + 1 WHERE id <= 5")
+ checkAnswer(sql("SELECT SUM(v) FROM T"), Row(500515))
+ checkAnswer(sql("SELECT v FROM T WHERE id = 1"), Row(3))
+
+ // Zero matched rows.
+ sql("UPDATE T SET v = 0 WHERE id > 10000")
+ checkAnswer(sql("SELECT SUM(v) FROM T"), Row(500515))
+
+ // NULL assignment.
+ sql("UPDATE T SET v = NULL WHERE id = 1000")
+ checkAnswer(sql("SELECT v FROM T WHERE id = 1000"), Row(null))
+ }
+ }
+
+ test("Paimon delta ops: update moving rows across partitions") {
+ withSparkSQLConf(v2WriteConf) {
+ sql(s"""
+ |CREATE TABLE T (id INT, pt STRING)
+ |PARTITIONED BY (pt)
+ |TBLPROPERTIES ('deletion-vectors.enabled' = 'true')
+ |""".stripMargin)
+ sql("INSERT INTO T VALUES (1, 'a'), (2, 'a'), (3, 'b')")
+
+ sql("UPDATE T SET pt = 'b' WHERE id = 1")
+ checkAnswer(
+ sql("SELECT id, pt FROM T ORDER BY id"),
+ Row(1, "b") :: Row(2, "a") :: Row(3, "b") :: Nil)
+ checkAnswer(sql("SELECT COUNT(*) FROM T WHERE pt = 'a'"), Row(1))
+ }
+ }
+
+ test("Paimon delta ops: merge into goes through WriteDelta") {
+ withSparkSQLConf(v2WriteConf) {
+ createDvAppendTable()
+ sql("INSERT INTO T VALUES (1, 1), (2, 2), (3, 3), (4, 4)")
+
+ val merge =
+ """
+ |MERGE INTO T
+ |USING (SELECT * FROM VALUES (1, 10), (2, 20), (5, 50) AS s(id, v)) s
+ |ON T.id = s.id
+ |WHEN MATCHED AND s.id = 1 THEN UPDATE SET T.v = s.v
+ |WHEN MATCHED AND s.id = 2 THEN DELETE
+ |WHEN NOT MATCHED THEN INSERT (id, v) VALUES (s.id, s.v)
+ |""".stripMargin
+
+ assert(explainContains(merge).contains("WriteDelta"))
+
+ val filesBefore = dataFilePaths()
+ sql(merge)
+ checkAnswer(
+ sql("SELECT id, v FROM T ORDER BY id"),
+ Row(1, 10) :: Row(3, 3) :: Row(4, 4) :: Row(5, 50) :: Nil)
+ assert(filesBefore.subsetOf(dataFilePaths()))
+ }
+ }
+
+ test("Paimon delta ops: merge into with not matched by source") {
+ withSparkSQLConf(v2WriteConf) {
+ createDvAppendTable()
+ sql("INSERT INTO T VALUES (1, 1), (2, 2), (3, 3)")
+
+ sql("""
+ |MERGE INTO T
+ |USING (SELECT * FROM VALUES (1, 10) AS s(id, v)) s
+ |ON T.id = s.id
+ |WHEN MATCHED THEN UPDATE SET T.v = s.v
+ |WHEN NOT MATCHED BY SOURCE AND T.id = 3 THEN DELETE
+ |WHEN NOT MATCHED BY SOURCE THEN UPDATE SET T.v = -1
+ |""".stripMargin)
+ checkAnswer(sql("SELECT id, v FROM T ORDER BY id"), Row(1, 10) :: Row(2,
-1) :: Nil)
+ }
+ }
+
+ test("Paimon delta ops: merge into cardinality violation") {
+ withSparkSQLConf(v2WriteConf) {
+ createDvAppendTable()
+ sql("INSERT INTO T VALUES (1, 1), (2, 2)")
+
+ val e = intercept[Throwable] {
+ sql("""
+ |MERGE INTO T
+ |USING (SELECT * FROM VALUES (1, 10), (1, 11) AS s(id, v)) s
+ |ON T.id = s.id
+ |WHEN MATCHED THEN UPDATE SET T.v = s.v
+ |""".stripMargin)
+ }
+ def messages(t: Throwable): Seq[String] =
+ Iterator.iterate(t)(_.getCause).takeWhile(_ !=
null).map(_.toString).toSeq
+ assert(messages(e).exists(_.contains("MERGE_CARDINALITY_VIOLATION")))
+ }
+ }
+
+ test("Paimon delta ops: concurrent deletion vector commits conflict") {
+ withSparkSQLConf(v2WriteConf) {
+ createDvAppendTable()
+ sql("INSERT INTO T SELECT /*+ REPARTITION(1) */ id, id FROM range(0,
100)")
+
+ // A delta write planned against the current snapshot, deleting row 0 of
the file.
+ val table = loadTable("T")
+ val readSnapshotId = table.snapshotManager().latestSnapshot().id()
+ val filePath = sql("SELECT __paimon_file_path FROM T LIMIT
1").head().getString(0)
+ val dv = new BitmapDeletionVector()
+ dv.delete(0L)
+ val taskResult = PaimonDeltaWriteTaskResult(
+ None,
+ Array(SerializedDeletionVector(filePath,
DeletionVector.serializeToBytes(dv))))
+
+ // A concurrent DELETE commits a deletion vector for the same file in
between.
+ sql("DELETE FROM T WHERE id = 50")
+
+ // Committing against the stale read snapshot must fail with a conflict
instead of silently
+ // merging with (or dropping) the concurrent deletion vector.
+ val rowIdSchema = StructType(
+ Seq(StructField(FILE_PATH_COLUMN, StringType),
StructField(ROW_INDEX_COLUMN, LongType)))
+ val batchWrite = new PaimonDeltaBatchWrite(
+ table,
+ new StructType(),
+ rowIdSchema,
+ Snapshot.Operation.DELETE,
+ Some(readSnapshotId))
+ val e = intercept[Throwable](batchWrite.commit(Array(taskResult)))
+ def messages(t: Throwable): Seq[String] =
+ Iterator.iterate(t)(_.getCause).takeWhile(_ !=
null).map(_.toString).toSeq
+ assert(messages(e).exists(_.toLowerCase.contains("conflict")))
+
+ // Only the concurrent DELETE applied; the failed commit is invisible.
+ checkAnswer(sql("SELECT COUNT(*), MIN(id) FROM T"), Row(99, 0))
+ }
+ }
+
+ test("Paimon delta ops: bucketed deletion-vector append table stays on V1") {
+ withSparkSQLConf(v2WriteConf) {
+ createDvAppendTable(", 'bucket-key' = 'id', 'bucket' = '2'")
+ sql("INSERT INTO T VALUES (1, 1), (2, 2), (3, 3)")
+ assert(explainContains("DELETE FROM T WHERE id =
1").contains("DeleteFromPaimonTable"))
+ sql("DELETE FROM T WHERE id = 1")
+ checkAnswer(sql("SELECT COUNT(*) FROM T"), Row(2))
+
+ val updatePlan = explainContains("UPDATE T SET v = v + 1 WHERE id = 2")
+ assert(!updatePlan.contains("WriteDelta") &&
!updatePlan.contains("ReplaceData"))
+ sql("UPDATE T SET v = v + 1 WHERE id = 2")
+ checkAnswer(sql("SELECT v FROM T WHERE id = 2"), Row(3))
+ }
+ }
+
+ test("Paimon delta ops: append table without deletion vectors keeps
copy-on-write") {
+ withSparkSQLConf(v2WriteConf) {
+ sql("CREATE TABLE T (id INT, v INT)")
+ sql("INSERT INTO T VALUES (1, 1), (2, 2), (3, 3)")
+ val plan = explainContains("DELETE FROM T WHERE id = 1")
+ assert(plan.contains("ReplaceData") && !plan.contains("WriteDelta"))
+ sql("DELETE FROM T WHERE id = 1")
+ checkAnswer(sql("SELECT COUNT(*) FROM T"), Row(2))
+ }
+ }
+
+ test("Paimon delta ops: falls back to V1 when v2 write is disabled") {
+ withSparkSQLConf("spark.paimon.write.use-v2-write" -> "false") {
+ createDvAppendTable()
+ sql("INSERT INTO T VALUES (1, 1), (2, 2)")
+ assert(explainContains("DELETE FROM T WHERE id =
1").contains("DeleteFromPaimonTable"))
+ sql("DELETE FROM T WHERE id = 1")
+ checkAnswer(sql("SELECT id FROM T"), Row(2))
+ }
+ }
+}
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 d2e23d097d..3b67841892 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
@@ -196,6 +196,37 @@ class PaimonMetricTest extends PaimonSparkTestBase with
ScanPlanHelper {
}
}
+ test(s"Paimon Metric: delta row-level operation record metrics") {
+ withSparkSQLConf("spark.paimon.write.use-v2-write" -> "true") {
+ sql("CREATE TABLE T (id INT, v INT) TBLPROPERTIES
('deletion-vectors.enabled' = 'true')")
+ sql(s"INSERT INTO T SELECT /*+ REPARTITION(1) */ id, id as v FROM
range(0, 10)")
+
+ val deleteDf = sql("DELETE FROM T WHERE id < 3")
+ val deleteMetrics = commandMetrics(deleteDf)
+ val deleteExecutionMetrics = commandExecutionMetrics(deleteMetrics)
+ assert(deleteExecutionMetrics(deleteMetrics("deletedRecords").id) == "3")
+
+ val updateDf = sql("UPDATE T SET v = v + 1 WHERE id >= 8")
+ val updateMetrics = commandMetrics(updateDf)
+ val updateExecutionMetrics = commandExecutionMetrics(updateMetrics)
+ assert(updateExecutionMetrics(updateMetrics("updatedRecords").id) == "2")
+
+ val mergeDf = sql("""
+ |MERGE INTO T
+ |USING (SELECT * FROM VALUES (3, 30), (4, 40), (100,
1) AS s(id, v)) s
+ |ON T.id = s.id
+ |WHEN MATCHED AND s.id = 3 THEN UPDATE SET T.v = s.v
+ |WHEN MATCHED AND s.id = 4 THEN DELETE
+ |WHEN NOT MATCHED THEN INSERT (id, v) VALUES (s.id,
s.v)
+ |""".stripMargin)
+ val mergeMetrics = commandMetrics(mergeDf)
+ val mergeExecutionMetrics = commandExecutionMetrics(mergeMetrics)
+ assert(mergeExecutionMetrics(mergeMetrics("updatedRecords").id) == "1")
+ assert(mergeExecutionMetrics(mergeMetrics("deletedRecords").id) == "1")
+ assert(mergeExecutionMetrics(mergeMetrics("insertedRecords").id) == "1")
+ }
+ }
+
def metric(metrics: Array[CustomTaskMetric], name: String): Long = {
metrics.find(_.name() == name).get.value()
}
diff --git
a/paimon-spark/paimon-spark3-common/src/main/scala/org/apache/paimon/spark/write/PaimonDeltaBatchWrite.scala
b/paimon-spark/paimon-spark3-common/src/main/scala/org/apache/paimon/spark/write/PaimonDeltaBatchWrite.scala
new file mode 100644
index 0000000000..e2c9250e83
--- /dev/null
+++
b/paimon-spark/paimon-spark3-common/src/main/scala/org/apache/paimon/spark/write/PaimonDeltaBatchWrite.scala
@@ -0,0 +1,49 @@
+/*
+ * 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.paimon.spark.write
+
+import org.apache.paimon.Snapshot
+import org.apache.paimon.table.FileStoreTable
+
+import org.apache.spark.sql.connector.write.{DeltaBatchWrite,
DeltaWriterFactory, PhysicalWriteInfo, WriterCommitMessage}
+import org.apache.spark.sql.types.StructType
+
+/**
+ * Spark-3.x thin wrapper that mixes `DeltaBatchWrite` into
[[PaimonDeltaWriteBase]]. See the base
+ * class scaladoc for why the inheritance lives here rather than in
`paimon-spark-common`.
+ */
+class PaimonDeltaBatchWrite(
+ table: FileStoreTable,
+ rowSchema: StructType,
+ rowIdSchema: StructType,
+ operationType: Snapshot.Operation,
+ readSnapshotId: Option[Long])
+ extends PaimonDeltaWriteBase(table, rowSchema, rowIdSchema, operationType,
readSnapshotId)
+ with DeltaBatchWrite
+ with Serializable {
+
+ override def createBatchWriterFactory(info: PhysicalWriteInfo):
DeltaWriterFactory =
+ createPaimonDeltaWriterFactory(info)
+
+ override def useCommitCoordinator(): Boolean = false
+
+ override def commit(messages: Array[WriterCommitMessage]): Unit =
commitMessages(messages)
+
+ override def abort(messages: Array[WriterCommitMessage]): Unit =
abortMessages(messages)
+}
diff --git
a/paimon-spark/paimon-spark3-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark3Shim.scala
b/paimon-spark/paimon-spark3-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark3Shim.scala
index d4993fdce4..9bde530b28 100644
---
a/paimon-spark/paimon-spark3-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark3Shim.scala
+++
b/paimon-spark/paimon-spark3-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark3Shim.scala
@@ -25,7 +25,7 @@ import
org.apache.paimon.spark.catalyst.parser.extensions.PaimonSpark3SqlExtensi
import org.apache.paimon.spark.data.{Spark3ArrayData, Spark3InternalRow,
Spark3InternalRowWithBlob, SparkArrayData, SparkInternalRow}
import org.apache.paimon.spark.format.FormatTableBatchWrite
import org.apache.paimon.spark.rowops.PaimonCopyOnWriteScan
-import org.apache.paimon.spark.write.PaimonBatchWrite
+import org.apache.paimon.spark.write.{PaimonBatchWrite, PaimonDeltaBatchWrite}
import org.apache.paimon.table.{FileStoreTable, FormatTable}
import org.apache.paimon.types.{DataType, RowType}
@@ -206,6 +206,14 @@ class Spark3Shim extends SparkShim {
copyOnWriteScan,
operationType)
+ override def createPaimonDeltaBatchWrite(
+ table: FileStoreTable,
+ rowSchema: StructType,
+ rowIdSchema: StructType,
+ operationType: Snapshot.Operation,
+ readSnapshotId: Option[Long]): BatchWrite =
+ new PaimonDeltaBatchWrite(table, rowSchema, rowIdSchema, operationType,
readSnapshotId)
+
override def createFormatTableBatchWrite(
table: FormatTable,
overwriteDynamic: Option[Boolean],
diff --git
a/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/paimon/spark/write/PaimonDeltaBatchWrite.scala
b/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/paimon/spark/write/PaimonDeltaBatchWrite.scala
new file mode 100644
index 0000000000..4584cfcc30
--- /dev/null
+++
b/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/paimon/spark/write/PaimonDeltaBatchWrite.scala
@@ -0,0 +1,51 @@
+/*
+ * 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.paimon.spark.write
+
+import org.apache.paimon.Snapshot
+import org.apache.paimon.table.FileStoreTable
+
+import org.apache.spark.sql.connector.write.{DeltaBatchWrite,
DeltaWriterFactory, PhysicalWriteInfo, WriterCommitMessage}
+import org.apache.spark.sql.types.StructType
+
+/**
+ * Spark-4.x thin wrapper that mixes `DeltaBatchWrite` into
[[PaimonDeltaWriteBase]]. See the base
+ * class scaladoc for why the inheritance lives here rather than in
`paimon-spark-common`; this
+ * class is compiled against Spark 4.1 and shadowed in
`paimon-spark-4.0/src/main` for the 4.0.2
+ * compile target.
+ */
+class PaimonDeltaBatchWrite(
+ table: FileStoreTable,
+ rowSchema: StructType,
+ rowIdSchema: StructType,
+ operationType: Snapshot.Operation,
+ readSnapshotId: Option[Long])
+ extends PaimonDeltaWriteBase(table, rowSchema, rowIdSchema, operationType,
readSnapshotId)
+ with DeltaBatchWrite
+ with Serializable {
+
+ override def createBatchWriterFactory(info: PhysicalWriteInfo):
DeltaWriterFactory =
+ createPaimonDeltaWriterFactory(info)
+
+ override def useCommitCoordinator(): Boolean = false
+
+ override def commit(messages: Array[WriterCommitMessage]): Unit =
commitMessages(messages)
+
+ override def abort(messages: Array[WriterCommitMessage]): Unit =
abortMessages(messages)
+}
diff --git
a/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/spark/sql/catalyst/analysis/PureAppendOnlyScope.scala
b/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/spark/sql/catalyst/analysis/PureAppendOnlyScope.scala
index 4fdf2bafc0..2f2c3755d4 100644
---
a/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/spark/sql/catalyst/analysis/PureAppendOnlyScope.scala
+++
b/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/spark/sql/catalyst/analysis/PureAppendOnlyScope.scala
@@ -29,10 +29,11 @@ import
org.apache.spark.sql.execution.datasources.v2.ExtractV2Table
* ([[Spark41UpdateTableRewrite]] for UPDATE + metadata-only DELETE
reverse-optimization,
* [[Spark41MergeIntoRewrite]] for MERGE).
*
- * These rules only intercept operations against Paimon tables that are valid
for Spark's V2
- * copy-on-write rewrite: no primary key, data evolution, deletion vectors, or
fixed-length
- * `CHAR(n)` columns. Row-tracking-only tables are included; tables that
violate any of these
- * constraints go through Paimon's postHoc V1 commands or Spark's built-in
analysis path.
+ * These rules intercept operations against Paimon tables that are valid for
Spark's V2
+ * copy-on-write rewrite (no primary key, data evolution, deletion vectors, or
fixed-length
+ * `CHAR(n)` columns; row-tracking-only tables are included) or for the
delta-based rewrite
+ * (unaware-bucket deletion-vector append tables, see
[[targetsV2DeltaTable]]). Tables that violate
+ * these constraints go through Paimon's postHoc V1 commands or Spark's
built-in analysis path.
*
* Kept as a mix-in trait so the two rewrite objects stay
single-responsibility (one rule per Spark
* row-level command, mirroring Spark's own `RewriteUpdateTable` /
`RewriteMergeIntoTable` layout)
@@ -50,6 +51,19 @@ trait PureAppendOnlyScope {
}
}
+ /**
+ * Whether the target is a Paimon table on the delta-based row-level path.
Delegates to
+ * `SparkTable.supportsV2DeltaOps` so the delta gating conditions have a
single source of truth
+ * (unlike the copy-on-write scope above, the full capability predicate is
also the correct plan
+ * scope: its extra `useV2Write` / Spark version checks are trivially true
once the table has
+ * exposed `SupportsRowLevelOperations` on 4.1).
+ */
+ protected def targetsV2DeltaTable(aliasedTable: LogicalPlan): Boolean = {
+ targetsPaimonFileStoreTable(aliasedTable) {
+ case (sparkTable, _) => SparkTable.supportsV2DeltaOps(sparkTable)
+ }
+ }
+
private def targetsPaimonFileStoreTable(aliasedTable: LogicalPlan)(
predicate: (SparkTable, FileStoreTable) => Boolean): Boolean = {
EliminateSubqueryAliases(aliasedTable) match {
diff --git
a/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/spark/sql/catalyst/analysis/Spark41DeleteMetadataRestore.scala
b/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/spark/sql/catalyst/analysis/Spark41DeleteMetadataRestore.scala
index d21bc8098e..7267b14fc2 100644
---
a/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/spark/sql/catalyst/analysis/Spark41DeleteMetadataRestore.scala
+++
b/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/spark/sql/catalyst/analysis/Spark41DeleteMetadataRestore.scala
@@ -23,7 +23,7 @@ import
org.apache.paimon.spark.catalyst.optimizer.OptimizeMetadataOnlyDeleteFrom
import org.apache.paimon.spark.commands.DeleteFromPaimonTableCommand
import org.apache.paimon.table.FileStoreTable
-import org.apache.spark.sql.catalyst.plans.logical.{AnalysisHelper,
LogicalPlan, ReplaceData}
+import org.apache.spark.sql.catalyst.plans.logical.{AnalysisHelper,
LogicalPlan, ReplaceData, WriteDelta}
import org.apache.spark.sql.connector.write.RowLevelOperation.Command.DELETE
import org.apache.spark.sql.connector.write.RowLevelOperationTable
import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
@@ -59,10 +59,42 @@ object Spark41DeleteMetadataRestore extends
RewriteRowLevelCommand with PureAppe
val origRelation =
rd.originalTable.asInstanceOf[DataSourceV2Relation]
val fs =
origRelation.table.asInstanceOf[SparkTable].getTable.asInstanceOf[FileStoreTable]
DeleteFromPaimonTableCommand(origRelation, fs, rd.condition)
+ // The delta-based DELETE form of unaware-bucket deletion-vector
append tables: restore
+ // metadata-only DELETE to the V1 command for the same truncate fast
path, instead of
+ // marking every row of the dropped partitions in deletion vectors.
+ case wd: WriteDelta if isMetadataOnlyDeleteOnDvPaimon(wd) =>
+ val origRelation =
wd.originalTable.asInstanceOf[DataSourceV2Relation]
+ val fs =
origRelation.table.asInstanceOf[SparkTable].getTable.asInstanceOf[FileStoreTable]
+ DeleteFromPaimonTableCommand(origRelation, fs, wd.condition)
}
}
}
+ /** The [[WriteDelta]] counterpart of
[[isMetadataOnlyDeleteOnAppendOnlyPaimon]]. */
+ private def isMetadataOnlyDeleteOnDvPaimon(wd: WriteDelta): Boolean = {
+ val writeIsDelete = wd.table match {
+ case r: DataSourceV2Relation =>
+ r.table match {
+ case op: RowLevelOperationTable => op.operation.command() == DELETE
+ case _ => false
+ }
+ case _ => false
+ }
+ writeIsDelete && (wd.originalTable match {
+ case r: DataSourceV2Relation if targetsV2DeltaTable(r) =>
+ r.table match {
+ case spk: SparkTable =>
+ spk.getTable match {
+ case fs: FileStoreTable =>
+
OptimizeMetadataOnlyDeleteFromPaimonTable.isMetadataOnlyDelete(fs, wd.condition)
+ case _ => false
+ }
+ case _ => false
+ }
+ case _ => false
+ })
+ }
+
/**
* Whether a `ReplaceData` node (Spark 4.1's post-rewrite DELETE form)
targets a Paimon table
* eligible for V2 copy-on-write with a metadata-only predicate, such that
converting back to
diff --git
a/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/spark/sql/catalyst/analysis/Spark41MergeIntoRewrite.scala
b/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/spark/sql/catalyst/analysis/Spark41MergeIntoRewrite.scala
index f3dd21f15f..0d012e18f2 100644
---
a/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/spark/sql/catalyst/analysis/Spark41MergeIntoRewrite.scala
+++
b/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/spark/sql/catalyst/analysis/Spark41MergeIntoRewrite.scala
@@ -25,13 +25,13 @@ import org.apache.spark.sql.{AnalysisException,
SparkSession}
import org.apache.spark.sql.catalyst.expressions.{Alias, And, Attribute,
AttributeReference, Exists, Expression, IsNotNull, Literal, MetadataAttribute,
MonotonicallyIncreasingID, OuterReference, PredicateHelper, SubqueryExpression}
import org.apache.spark.sql.catalyst.expressions.Literal.{FalseLiteral,
TrueLiteral}
import org.apache.spark.sql.catalyst.expressions.aggregate.AggregateExpression
-import org.apache.spark.sql.catalyst.plans.{FullOuter, JoinType, LeftAnti,
LeftOuter}
-import org.apache.spark.sql.catalyst.plans.logical.{AnalysisHelper,
AppendData, DeleteAction, Filter, HintInfo, InsertAction, Join, JoinHint,
LogicalPlan, MergeAction, MergeIntoTable, MergeRows,
NO_BROADCAST_AND_REPLICATION, Project, ReplaceData, UpdateAction}
-import org.apache.spark.sql.catalyst.plans.logical.MergeRows.{Copy, Discard,
Insert, Instruction, Keep, ROW_ID, Update}
+import org.apache.spark.sql.catalyst.plans.{FullOuter, Inner, JoinType,
LeftAnti, LeftOuter, RightOuter}
+import org.apache.spark.sql.catalyst.plans.logical.{AnalysisHelper,
AppendData, DeleteAction, Filter, HintInfo, InsertAction, Join, JoinHint,
LogicalPlan, MergeAction, MergeIntoTable, MergeRows,
NO_BROADCAST_AND_REPLICATION, Project, ReplaceData, UpdateAction, WriteDelta}
+import org.apache.spark.sql.catalyst.plans.logical.MergeRows.{Copy, Delete,
Discard, Insert, Instruction, Keep, ROW_ID, Update}
import org.apache.spark.sql.catalyst.util.RowDeltaUtils.{OPERATION_COLUMN,
WRITE_OPERATION, WRITE_WITH_METADATA_OPERATION}
import org.apache.spark.sql.connector.catalog.SupportsRowLevelOperations
+import org.apache.spark.sql.connector.write.{RowLevelOperationTable,
SupportsDelta}
import org.apache.spark.sql.connector.write.RowLevelOperation.Command.MERGE
-import org.apache.spark.sql.connector.write.RowLevelOperationTable
import org.apache.spark.sql.errors.QueryCompilationErrors
import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation,
ExtractV2Table}
import org.apache.spark.sql.types.IntegerType
@@ -47,14 +47,16 @@ import org.apache.spark.sql.util.CaseInsensitiveStringMap
* already marked analyzed and silently skipped, so the planner rejects it with
* `UNSUPPORTED_FEATURE.TABLE_OPERATION`. We intercept via `transformDown`
under
* `allowInvokingTransformsInAnalyzer` and inline the three
`ReplaceData`/`AppendData` branches.
- * `SupportsDelta` is omitted — Paimon is copy-on-write only.
+ * Copy-on-write tables mirror the non-delta rewrite; deletion-vector append
tables mirror the
+ * `SupportsDelta` (`WriteDelta`) rewrite.
*
* We fire before `ResolveAssignments`, so `m.aligned` is `false`. The rule
pre-aligns each action
* list via `PaimonAssignmentUtils.alignActions` (shared with the postHoc
`PaimonMergeInto` rule).
*
- * Row-tracking-only tables use the same V2 copy-on-write rewrite. CHAR
columns are excluded —
- * `readSidePadding` races with the rewrite and trips CheckAnalysis; those
plans fall back to the
- * postHoc `PaimonMergeInto` V1 path, which also owns PK / DE / DV tables via
+ * Row-tracking-only tables use the same V2 copy-on-write rewrite;
unaware-bucket deletion-vector
+ * append tables take the transcribed `WriteDelta` branch (delta-based
row-level operations). CHAR
+ * columns are excluded — `readSidePadding` races with the rewrite and trips
CheckAnalysis; those
+ * plans fall back to the postHoc `PaimonMergeInto` V1 path, which also owns
PK / DE tables via
* `RowLevelHelper.shouldFallbackToV1MergeInto`.
*/
object Spark41MergeIntoRewrite
@@ -72,7 +74,7 @@ object Spark41MergeIntoRewrite
plan.transformDown {
case m: MergeIntoTable
if m.resolved && m.rewritable && !m.needSchemaEvolution &&
- targetsV2CopyOnWriteTable(m.targetTable) =>
+ (targetsV2CopyOnWriteTable(m.targetTable) ||
targetsV2DeltaTable(m.targetTable)) =>
// Pure append-only tables skip postHoc `PaimonMergeInto`, so evolve
schema here.
val evolved = evolveSchemaIfPaimon(m)
rewrite(alignAllMergeActions(evolved, evolved.targetTable.output))
@@ -112,14 +114,26 @@ object Spark41MergeIntoRewrite
buildNotMatchedOnlyAppendPlan(relation, source, cond,
notMatchedActions)
} else {
val operationTable = buildOperationTable(tbl, MERGE,
CaseInsensitiveStringMap.empty())
- buildReplaceDataPlan(
- relation,
- operationTable,
- source,
- cond,
- matchedActions,
- notMatchedActions,
- notMatchedBySourceActions)
+ operationTable.operation match {
+ case _: SupportsDelta =>
+ buildWriteDeltaPlan(
+ relation,
+ operationTable,
+ source,
+ cond,
+ matchedActions,
+ notMatchedActions,
+ notMatchedBySourceActions)
+ case _ =>
+ buildReplaceDataPlan(
+ relation,
+ operationTable,
+ source,
+ cond,
+ matchedActions,
+ notMatchedActions,
+ notMatchedBySourceActions)
+ }
}
case _ =>
m
@@ -174,6 +188,149 @@ object Spark41MergeIntoRewrite
AppendData.byPosition(r, mergeRows)
}
+ // Delta path producing a `WriteDelta` plan for deletion-vector append
tables. Mirrors Spark
+ // 4.1's `RewriteMergeIntoTable.{buildWriteDeltaPlan,
buildWriteDeltaMergeRowsPlan,
+ // chooseWriteDeltaJoinType, pushDownTargetPredicates}`; only the non-split
UPDATE branch is
+ // transcribed because
`PaimonSparkDeltaOperation.representUpdateAsDeleteAndInsert` is false.
+ private def buildWriteDeltaPlan(
+ relation: DataSourceV2Relation,
+ operationTable: RowLevelOperationTable,
+ source: LogicalPlan,
+ cond: Expression,
+ matchedActions: Seq[MergeAction],
+ notMatchedActions: Seq[MergeAction],
+ notMatchedBySourceActions: Seq[MergeAction]): WriteDelta = {
+
+ val operation = operationTable.operation.asInstanceOf[SupportsDelta]
+ assert(
+ !operation.representUpdateAsDeleteAndInsert,
+ "Paimon delta operations represent UPDATE as a single operation")
+
+ val rowAttrs = relation.output
+ val rowIdAttrs = resolveRowIdAttrs(relation, operation)
+ val metadataAttrs = resolveRequiredMetadataAttrs(relation, operation)
+
+ val readRelation = buildRelationWithAttrs(relation, operationTable,
metadataAttrs, rowIdAttrs)
+
+ // if there is no NOT MATCHED BY SOURCE clause, predicates of the ON
condition that
+ // reference only the target table can be pushed down
+ val (filteredReadRelation, joinCond) = if
(notMatchedBySourceActions.isEmpty) {
+ pushDownTargetPredicates(readRelation, cond)
+ } else {
+ (readRelation, cond)
+ }
+
+ val checkCardinality = shouldCheckCardinality(matchedActions)
+
+ val joinType = chooseWriteDeltaJoinType(notMatchedActions,
notMatchedBySourceActions)
+ val joinPlan = join(filteredReadRelation, source, joinType, joinCond,
checkCardinality)
+
+ val mergeRowsPlan = buildWriteDeltaMergeRowsPlan(
+ readRelation,
+ joinPlan,
+ matchedActions,
+ notMatchedActions,
+ notMatchedBySourceActions,
+ rowIdAttrs,
+ checkCardinality)
+
+ val writeRelation = relation.copy(table = operationTable)
+ val projections = buildWriteDeltaProjections(mergeRowsPlan, rowAttrs,
rowIdAttrs, metadataAttrs)
+ WriteDelta(writeRelation, cond, mergeRowsPlan, relation, projections)
+ }
+
+ private def chooseWriteDeltaJoinType(
+ notMatchedActions: Seq[MergeAction],
+ notMatchedBySourceActions: Seq[MergeAction]): JoinType = {
+ val unmatchedTargetRowsRequired = notMatchedBySourceActions.nonEmpty
+ val unmatchedSourceRowsRequired = notMatchedActions.nonEmpty
+ if (unmatchedTargetRowsRequired && unmatchedSourceRowsRequired) {
+ FullOuter
+ } else if (unmatchedTargetRowsRequired) {
+ LeftOuter
+ } else if (unmatchedSourceRowsRequired) {
+ RightOuter
+ } else {
+ Inner
+ }
+ }
+
+ private def buildWriteDeltaMergeRowsPlan(
+ targetTable: DataSourceV2Relation,
+ joinPlan: LogicalPlan,
+ matchedActions: Seq[MergeAction],
+ notMatchedActions: Seq[MergeAction],
+ notMatchedBySourceActions: Seq[MergeAction],
+ rowIdAttrs: Seq[Attribute],
+ checkCardinality: Boolean): MergeRows = {
+
+ val (metadataAttrs, rowAttrs) =
+ targetTable.output.partition(attr =>
MetadataAttribute.isValid(attr.metadata))
+
+ // original row ID values must be preserved and passed back to the table
to encode updates
+ // if there are any assignments to row ID attributes, add extra columns
for original values
+ val updateAssignments = (matchedActions ++
notMatchedBySourceActions).flatMap {
+ case UpdateAction(_, assignments, _) => assignments
+ case _ => Nil
+ }
+ val originalRowIdValues = buildOriginalRowIdValues(rowIdAttrs,
updateAssignments)
+
+ def toDeltaInstruction(action: MergeAction): Instruction = {
+ action match {
+ case UpdateAction(cond, assignments, _) =>
+ val output = deltaUpdateOutput(assignments, metadataAttrs,
originalRowIdValues)
+ Keep(Update, cond.getOrElse(TrueLiteral), output)
+ case DeleteAction(cond) =>
+ val output = deltaDeleteOutput(rowAttrs, rowIdAttrs, metadataAttrs,
originalRowIdValues)
+ Keep(Delete, cond.getOrElse(TrueLiteral), output)
+ case InsertAction(cond, assignments) =>
+ val output = deltaInsertOutput(assignments, metadataAttrs,
originalRowIdValues)
+ Keep(Insert, cond.getOrElse(TrueLiteral), output)
+ case other =>
+ throw new AnalysisException(
+ errorClass = "_LEGACY_ERROR_TEMP_3052",
+ messageParameters = Map("other" -> other.toString))
+ }
+ }
+
+ val matchedInstructions = matchedActions.map(toDeltaInstruction)
+ val notMatchedInstructions = notMatchedActions.map(toDeltaInstruction)
+ val notMatchedBySourceInstructions =
notMatchedBySourceActions.map(toDeltaInstruction)
+
+ val rowFromSourceAttr = resolveAttrRef(ROW_FROM_SOURCE, joinPlan)
+ val rowFromTargetAttr = resolveAttrRef(ROW_FROM_TARGET, joinPlan)
+
+ val outputs = matchedInstructions.flatMap(_.outputs) ++
+ notMatchedInstructions.flatMap(_.outputs) ++
+ notMatchedBySourceInstructions.flatMap(_.outputs)
+
+ val operationTypeAttr = AttributeReference(OPERATION_COLUMN, IntegerType,
nullable = false)()
+ val originalRowIdAttrs = originalRowIdValues.map(_.toAttribute)
+ val attrs = Seq(operationTypeAttr) ++ targetTable.output ++
originalRowIdAttrs
+
+ MergeRows(
+ isSourceRowPresent = IsNotNull(rowFromSourceAttr),
+ isTargetRowPresent = IsNotNull(rowFromTargetAttr),
+ matchedInstructions = matchedInstructions,
+ notMatchedInstructions = notMatchedInstructions,
+ notMatchedBySourceInstructions = notMatchedBySourceInstructions,
+ checkCardinality = checkCardinality,
+ output = generateExpandOutput(attrs, outputs),
+ joinPlan
+ )
+ }
+
+ private def pushDownTargetPredicates(
+ targetTable: LogicalPlan,
+ cond: Expression): (LogicalPlan, Expression) = {
+ val predicates = splitConjunctivePredicates(cond)
+ val (targetPredicates, joinPredicates) =
+ predicates.partition(predicate =>
predicate.references.subsetOf(targetTable.outputSet))
+ val targetCond = targetPredicates.reduceOption(And).getOrElse(TrueLiteral)
+ val joinCond = joinPredicates.reduceOption(And).getOrElse(TrueLiteral)
+ (Filter(targetCond, targetTable), joinCond)
+ }
+
// General path producing a `ReplaceData` plan. Mirrors Spark 4.1.1's
// `RewriteMergeIntoTable.buildReplaceDataPlan` +
`buildReplaceDataMergeRowsPlan`.
private def buildReplaceDataPlan(
diff --git
a/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/spark/sql/catalyst/analysis/Spark41UpdateTableRewrite.scala
b/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/spark/sql/catalyst/analysis/Spark41UpdateTableRewrite.scala
index 97edbdc780..e669f3a1df 100644
---
a/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/spark/sql/catalyst/analysis/Spark41UpdateTableRewrite.scala
+++
b/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/spark/sql/catalyst/analysis/Spark41UpdateTableRewrite.scala
@@ -20,13 +20,13 @@ package org.apache.spark.sql.catalyst.analysis
import org.apache.paimon.spark.catalyst.analysis.PaimonAssignmentUtils
-import org.apache.spark.sql.catalyst.expressions.{Alias, EqualNullSafe,
Expression, If, Literal, MetadataAttribute, Not, SubqueryExpression}
+import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute,
EqualNullSafe, Expression, If, Literal, MetadataAttribute, Not,
SubqueryExpression}
import org.apache.spark.sql.catalyst.expressions.Literal.TrueLiteral
-import org.apache.spark.sql.catalyst.plans.logical.{AnalysisHelper,
Assignment, Filter, LogicalPlan, Project, ReplaceData, Union, UpdateTable}
-import
org.apache.spark.sql.catalyst.util.RowDeltaUtils.WRITE_WITH_METADATA_OPERATION
+import org.apache.spark.sql.catalyst.plans.logical.{AnalysisHelper,
Assignment, Filter, LogicalPlan, Project, ReplaceData, Union, UpdateTable,
WriteDelta}
+import org.apache.spark.sql.catalyst.util.RowDeltaUtils.{OPERATION_COLUMN,
UPDATE_OPERATION, WRITE_WITH_METADATA_OPERATION}
import org.apache.spark.sql.connector.catalog.SupportsRowLevelOperations
+import org.apache.spark.sql.connector.write.{RowLevelOperationTable,
SupportsDelta}
import org.apache.spark.sql.connector.write.RowLevelOperation.Command.UPDATE
-import org.apache.spark.sql.connector.write.RowLevelOperationTable
import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation,
ExtractV2Table}
import org.apache.spark.sql.util.CaseInsensitiveStringMap
@@ -47,9 +47,10 @@ import org.apache.spark.sql.util.CaseInsensitiveStringMap
* We fire before `ResolveAssignments`, so `u.aligned` is `false`; the rule
pre-aligns via
* `PaimonAssignmentUtils.alignUpdateAssignments` before building the plan.
*
- * Row-tracking-only tables use the same V2 copy-on-write rewrite. PK / DE /
DV tables go through
- * the postHoc V1 rule because they do not expose
`SupportsRowLevelOperations`. DELETE is handled by
- * [[Spark41DeleteMetadataRestore]]; MERGE by [[Spark41MergeIntoRewrite]].
+ * Row-tracking-only tables use the same V2 copy-on-write rewrite;
unaware-bucket deletion-vector
+ * append tables take the transcribed `WriteDelta` branch (delta-based
row-level operations). PK /
+ * DE tables go through the postHoc V1 rule because they do not expose
`SupportsRowLevelOperations`.
+ * DELETE is handled by [[Spark41DeleteMetadataRestore]]; MERGE by
[[Spark41MergeIntoRewrite]].
*/
object Spark41UpdateTableRewrite extends RewriteRowLevelCommand with
PureAppendOnlyScope {
@@ -58,7 +59,8 @@ object Spark41UpdateTableRewrite extends
RewriteRowLevelCommand with PureAppendO
AnalysisHelper.allowInvokingTransformsInAnalyzer {
plan.transformDown {
case u @ UpdateTable(aliasedTable, assignments, cond)
- if u.resolved && u.rewritable &&
targetsV2CopyOnWriteTable(aliasedTable) =>
+ if u.resolved && u.rewritable &&
+ (targetsV2CopyOnWriteTable(aliasedTable) ||
targetsV2DeltaTable(aliasedTable)) =>
EliminateSubqueryAliases(aliasedTable) match {
case r @ ExtractV2Table(tbl: SupportsRowLevelOperations) =>
val table = buildOperationTable(tbl, UPDATE,
CaseInsensitiveStringMap.empty())
@@ -70,10 +72,13 @@ object Spark41UpdateTableRewrite extends
RewriteRowLevelCommand with PureAppendO
assignments,
fromStar = false,
mergeSchemaEnabled = false)
- if (SubqueryExpression.hasSubquery(updateCond)) {
- buildReplaceDataWithUnionPlan(r, table, alignedAssignments,
updateCond)
- } else {
- buildReplaceDataPlan(r, table, alignedAssignments, updateCond)
+ table.operation match {
+ case _: SupportsDelta =>
+ buildWriteDeltaPlan(r, table, alignedAssignments, updateCond)
+ case _ if SubqueryExpression.hasSubquery(updateCond) =>
+ buildReplaceDataWithUnionPlan(r, table, alignedAssignments,
updateCond)
+ case _ =>
+ buildReplaceDataPlan(r, table, alignedAssignments,
updateCond)
}
case _ =>
u
@@ -82,6 +87,54 @@ object Spark41UpdateTableRewrite extends
RewriteRowLevelCommand with PureAppendO
}
}
+ // Mirrors Spark 4.1 `RewriteUpdateTable.buildWriteDeltaPlan` for delta
tables (Paimon's
+ // `PaimonSparkDeltaOperation` answers `representUpdateAsDeleteAndInsert =
false`, so only the
+ // single-row UPDATE projection branch is transcribed).
+ private def buildWriteDeltaPlan(
+ relation: DataSourceV2Relation,
+ operationTable: RowLevelOperationTable,
+ assignments: Seq[Assignment],
+ cond: Expression): WriteDelta = {
+ val operation = operationTable.operation.asInstanceOf[SupportsDelta]
+ val rowAttrs = relation.output
+ val rowIdAttrs = resolveRowIdAttrs(relation, operation)
+ val metadataAttrs = resolveRequiredMetadataAttrs(relation, operation)
+ val readRelation = buildRelationWithAttrs(relation, operationTable,
metadataAttrs, rowIdAttrs)
+ val matchedRowsPlan = Filter(cond, readRelation)
+ assert(
+ !operation.representUpdateAsDeleteAndInsert,
+ "Paimon delta operations represent UPDATE as a single operation")
+ val rowDeltaPlan = buildWriteDeltaUpdateProjection(matchedRowsPlan,
assignments, rowIdAttrs)
+ val writeRelation = relation.copy(table = operationTable)
+ val projections = buildWriteDeltaProjections(rowDeltaPlan, rowAttrs,
rowIdAttrs, metadataAttrs)
+ WriteDelta(writeRelation, cond, rowDeltaPlan, relation, projections)
+ }
+
+ // Mirrors Spark 4.1 `RewriteUpdateTable.buildWriteDeltaUpdateProjection`.
+ private def buildWriteDeltaUpdateProjection(
+ plan: LogicalPlan,
+ assignments: Seq[Assignment],
+ rowIdAttrs: Seq[Attribute]): LogicalPlan = {
+ val assignedValues = assignments.map(_.value)
+ val updatedValues = plan.output.zipWithIndex.map {
+ case (attr, index) =>
+ if (index < assignments.size) {
+ val assignedExpr = assignedValues(index)
+ Alias(assignedExpr, attr.name)()
+ } else {
+ assert(MetadataAttribute.isValid(attr.metadata))
+ if (MetadataAttribute.isPreservedOnUpdate(attr)) {
+ attr
+ } else {
+ Alias(Literal(null, attr.dataType), attr.name)(explicitMetadata =
Some(attr.metadata))
+ }
+ }
+ }
+ val originalRowIdValues = buildOriginalRowIdValues(rowIdAttrs, assignments)
+ val operationType = Alias(Literal(UPDATE_OPERATION), OPERATION_COLUMN)()
+ Project(Seq(operationType) ++ updatedValues ++ originalRowIdValues, plan)
+ }
+
// Mirrors Spark 4.1.1 `RewriteUpdateTable.{buildReplaceDataPlan,
buildReplaceDataWithUnionPlan,
// buildReplaceDataUpdateProjection}`.
private def buildReplaceDataPlan(
diff --git
a/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala
b/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala
index 00d6024362..7a7cdc70f5 100644
---
a/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala
+++
b/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala
@@ -25,7 +25,7 @@ import
org.apache.paimon.spark.catalyst.parser.extensions.PaimonSpark4SqlExtensi
import org.apache.paimon.spark.data.{Spark4ArrayData, Spark4InternalRow,
Spark4InternalRowWithBlob, SparkArrayData, SparkInternalRow}
import org.apache.paimon.spark.format.FormatTableBatchWrite
import org.apache.paimon.spark.rowops.PaimonCopyOnWriteScan
-import org.apache.paimon.spark.write.PaimonBatchWrite
+import org.apache.paimon.spark.write.{PaimonBatchWrite, PaimonDeltaBatchWrite}
import org.apache.paimon.table.{FileStoreTable, FormatTable}
import org.apache.paimon.types.{DataType, RowType}
@@ -204,6 +204,14 @@ class Spark4Shim extends SparkShim {
copyOnWriteScan,
operationType)
+ override def createPaimonDeltaBatchWrite(
+ table: FileStoreTable,
+ rowSchema: StructType,
+ rowIdSchema: StructType,
+ operationType: Snapshot.Operation,
+ readSnapshotId: Option[Long]): BatchWrite =
+ new PaimonDeltaBatchWrite(table, rowSchema, rowIdSchema, operationType,
readSnapshotId)
+
override def createFormatTableBatchWrite(
table: FormatTable,
overwriteDynamic: Option[Boolean],
