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 b36f8075f04d..7f4295bd657a 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 57951ce9b279..c8751348e18b 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 3d50edfff11c..c7d4e9156ff2 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 000000000000..cb3d3c6ba059 --- /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 b92411d8bd9e..e3046f564f02 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 d3d2a393646d..85e639a842e2 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 6be314cee847..d42d129925dd 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 9ea20de1909d..2dd70920cd83 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 000000000000..f91d85a74d2b --- /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 41c6cdaaafae..8dc90a7c4f5f 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 b13cea4b3bd2..5ac1688223ee 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 000000000000..2117c50fde38 --- /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 000000000000..3f1c6001fa96 --- /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 000000000000..6a459ccbf922 --- /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 000000000000..72bd73b84625 --- /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 df73bd35d0c0..c039bf80ed10 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 b925036463e8..85c923325dd3 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 000000000000..b7190502b952 --- /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 d2e23d097d88..3b6784189255 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 000000000000..e2c9250e8356 --- /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 d4993fdce44f..9bde530b28c7 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 000000000000..4584cfcc30be --- /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 4fdf2bafc02f..2f2c3755d43b 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 d21bc8098e49..7267b14fc226 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 f3dd21f15f45..0d012e18f249 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 97edbdc780c3..e669f3a1dffd 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 00d602436256..7a7cdc70f531 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],