This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 18d9d74bcc [spark] Rebase data evolution merge updates after
compaction (#9091)
18d9d74bcc is described below
commit 18d9d74bcc82f0375758fb919932aa3ffb84f754
Author: Jingsong Lee <[email protected]>
AuthorDate: Fri Aug 7 15:22:06 2026 +0800
[spark] Rebase data evolution merge updates after compaction (#9091)
---
docs/docs/multimodal-table/data-evolution.mdx | 9 +
.../paimon/operation/commit/ConflictDetection.java | 12 +-
.../commit/RowIdExistenceConflictException.java | 34 ++
.../operation/commit/ConflictDetectionTest.java | 8 +-
.../MergeIntoPaimonDataEvolutionTable.scala | 13 +-
.../DataEvolutionRowIdConflictRewriter.scala | 400 +++++++++++++++++++++
.../MergeIntoPaimonDataEvolutionTable.scala | 13 +-
.../paimon/spark/sql/RowTrackingTestBase.scala | 59 ++-
8 files changed, 530 insertions(+), 18 deletions(-)
diff --git a/docs/docs/multimodal-table/data-evolution.mdx
b/docs/docs/multimodal-table/data-evolution.mdx
index f630c7dc3e..e40b976783 100644
--- a/docs/docs/multimodal-table/data-evolution.mdx
+++ b/docs/docs/multimodal-table/data-evolution.mdx
@@ -211,6 +211,15 @@ Notes:
`spark.paimon.write.data-evolution.update-conflict-retry.max-attempts` and
retry wait time with
`spark.paimon.write.data-evolution.update-conflict-retry.wait-ms`.
+- If concurrent compaction changes row-ID file boundaries after Spark SQL
+ `MERGE INTO` stages regular partial-column files, Spark rebases those staged
+ files onto the latest boundaries before committing instead of rerunning the
+ MERGE source and join. Spark performs this rewrite with distributed DataFrame
+ processing, so the PyPaimon-only
+ `data-evolution.row-id-conflict-rewrite.max-size` limit does not apply.
+ Recovery does not apply when deletion vectors are enabled or when
+ existing-row BLOB or VECTOR files are staged, and it does not hide logical
+ concurrent-update conflicts.
- In Spark SQL, `MERGE INTO` supports `WHEN NOT MATCHED BY SOURCE` for delete
actions on Data Evolution tables.
- The Flink `data_evolution_merge_into` procedure currently supports updating
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/commit/ConflictDetection.java
b/paimon-core/src/main/java/org/apache/paimon/operation/commit/ConflictDetection.java
index 99bae20226..ca7e57224e 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/commit/ConflictDetection.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/commit/ConflictDetection.java
@@ -811,15 +811,9 @@ public class ConflictDetection {
return Optional.empty();
}
- private RuntimeException rowIdExistenceConflict(SimpleFileEntry entry) {
- return new RuntimeException(
- String.format(
- "Row ID existence conflict: file '%s' references "
- + "firstRowId=%d, rowCount=%d in bucket %d, "
- + "but no matching file exists in the current
snapshot. "
- + "The referenced file may have been rewritten
by a "
- + "concurrent compaction or removed by an
overwrite.",
- entry.fileName(), entry.firstRowId(),
entry.rowCount(), entry.bucket()));
+ private RowIdExistenceConflictException
rowIdExistenceConflict(SimpleFileEntry entry) {
+ return new RowIdExistenceConflictException(
+ entry.fileName(), entry.firstRowId(), entry.rowCount(),
entry.bucket());
}
private static boolean dedicatedStorageFile(String fileName) {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/commit/RowIdExistenceConflictException.java
b/paimon-core/src/main/java/org/apache/paimon/operation/commit/RowIdExistenceConflictException.java
new file mode 100644
index 0000000000..4b14f928c2
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/commit/RowIdExistenceConflictException.java
@@ -0,0 +1,34 @@
+/*
+ * 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.operation.commit;
+
+/** Conflict caused by a staged file referencing a row-id range absent from
the latest snapshot. */
+public final class RowIdExistenceConflictException extends RuntimeException {
+
+ RowIdExistenceConflictException(String fileName, long firstRowId, long
rowCount, int bucket) {
+ super(
+ String.format(
+ "Row ID existence conflict: file '%s' references "
+ + "firstRowId=%d, rowCount=%d in bucket %d, "
+ + "but no matching file exists in the current
snapshot. "
+ + "The referenced file may have been rewritten
by a "
+ + "concurrent compaction or removed by an
overwrite.",
+ fileName, firstRowId, rowCount, bucket));
+ }
+}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/commit/ConflictDetectionTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/commit/ConflictDetectionTest.java
index d73cfac0e0..71757cb49d 100644
---
a/paimon-core/src/test/java/org/apache/paimon/operation/commit/ConflictDetectionTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/commit/ConflictDetectionTest.java
@@ -461,7 +461,9 @@ class ConflictDetectionTest {
detection.checkRowIdExistence(
baseEntries, deltaEntries, 100L,
Snapshot.CommitKind.APPEND);
assertThat(result).isPresent();
- assertThat(result.get().getMessage()).contains("Row ID existence
conflict");
+ assertThat(result.get())
+ .isInstanceOf(RowIdExistenceConflictException.class)
+ .hasMessageContaining("Row ID existence conflict");
}
@Test
@@ -478,7 +480,9 @@ class ConflictDetectionTest {
detection.checkRowIdExistence(
baseEntries, deltaEntries, 200L,
Snapshot.CommitKind.APPEND);
assertThat(result).isPresent();
- assertThat(result.get().getMessage()).contains("Row ID existence
conflict");
+ assertThat(result.get())
+ .isInstanceOf(RowIdExistenceConflictException.class)
+ .hasMessageContaining("Row ID existence conflict");
}
@Test
diff --git
a/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala
b/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala
index 79e09bc380..4585f75885 100644
---
a/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala
+++
b/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala
@@ -287,9 +287,16 @@ case class MergeIntoPaimonDataEvolutionTable(
if (readSnapshot != null) {
writer.rowIdCheckConflict(readSnapshot.id())
}
- writer.commit(
- matchedResult.commitMessages ++ deleteCommit ++ insertCommit,
- Snapshot.Operation.MERGE)
+ DataEvolutionRowIdConflictCommitter.commit(
+ sparkSession,
+ table,
+ targetRelation,
+ writer,
+ matchedResult.commitMessages,
+ deleteCommit ++ insertCommit,
+ if (readSnapshot == null) -1L else readSnapshot.id(),
+ Snapshot.Operation.MERGE
+ )
} finally {
targetActionCleanup()
if (persistSourceDss.isDefined) {
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/DataEvolutionRowIdConflictRewriter.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/DataEvolutionRowIdConflictRewriter.scala
new file mode 100644
index 0000000000..5512c735fb
--- /dev/null
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/DataEvolutionRowIdConflictRewriter.scala
@@ -0,0 +1,400 @@
+/*
+ * 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.commands
+
+import org.apache.paimon.Snapshot
+import org.apache.paimon.data.BinaryRow
+import org.apache.paimon.format.blob.BlobFileFormat.isBlobFile
+import org.apache.paimon.io.{DataFileMeta, DataIncrement}
+import org.apache.paimon.operation.commit.RowIdExistenceConflictException
+import org.apache.paimon.spark.util.ScanPlanHelper
+import org.apache.paimon.table.{FileStoreTable, SpecialFields}
+import org.apache.paimon.table.sink.{CommitMessage, CommitMessageImpl}
+import org.apache.paimon.table.source.DataSplit
+import org.apache.paimon.types.VectorType.isVectorStoreFile
+import org.apache.paimon.utils.{ExceptionUtils, Range, RetryWaiter}
+
+import org.apache.spark.sql.{Row, SparkSession}
+import org.apache.spark.sql.PaimonUtils.createDataset
+import org.apache.spark.sql.catalyst.analysis.SimpleAnalyzer.resolver
+import org.apache.spark.sql.catalyst.expressions.AttributeReference
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+import org.apache.spark.sql.functions.{col, udf}
+import org.apache.spark.sql.paimon.shims.SparkShimLoader
+import org.slf4j.LoggerFactory
+
+import scala.collection.JavaConverters._
+import scala.collection.immutable
+
+/** Rebase staged partial-column files onto current row-id file boundaries. */
+private[spark] class DataEvolutionRowIdConflictRewriter(
+ table: FileStoreTable,
+ targetRelation: DataSourceV2Relation)
+ extends ScanPlanHelper {
+
+ import DataEvolutionRowIdConflictRewriter._
+
+ def rewrite(
+ sparkSession: SparkSession,
+ latestSnapshot: Snapshot,
+ commitMessages: Seq[CommitMessage]): Option[RewriteResult] = {
+ if (table.coreOptions().deletionVectorsEnabled()) {
+ return None
+ }
+
+ val messageImpls = commitMessages.collect { case message:
CommitMessageImpl => message }
+ if (messageImpls.size != commitMessages.size) {
+ return None
+ }
+
+ val stagedFiles = messageImpls.flatMap(
+ message =>
+ message
+ .newFilesIncrement()
+ .newFiles()
+ .asScala
+ .map(file => StagedFile(message, file)))
+ val nextRowId =
Option(latestSnapshot.nextRowId()).map(_.longValue()).getOrElse(return None)
+
+ if (
+ stagedFiles.exists(
+ staged =>
+ isDedicatedFile(staged.file) && staged.file.firstRowId() != null &&
+ staged.file.firstRowId() < nextRowId)
+ ) {
+ return None
+ }
+
+ val currentSplits = table
+ .newSnapshotReader()
+ .withSnapshot(latestSnapshot)
+ .read()
+ .splits()
+ .asScala
+ .collect { case split: DataSplit => split }
+ .toSeq
+ val currentFiles = currentSplits.flatMap(
+ split =>
+ split
+ .dataFiles()
+ .asScala
+ .filter(isNormalRowIdFile)
+ .map(file => CurrentFile(split, file)))
+ val currentExactRanges = currentFiles.map(file => rangeKey(file.split,
file.file)).toSet
+ val candidates = stagedFiles.filter(
+ staged =>
+ isRewriteCandidate(staged.file, nextRowId) &&
+ !currentExactRanges.contains(rangeKey(staged.message, staged.file)))
+
+ if (candidates.isEmpty || !rangesAreStillCovered(currentFiles,
candidates)) {
+ return None
+ }
+
+ val affectedSplits = currentSplits.flatMap(
+ split => {
+ val filtered = split.filterDataFile(
+ file =>
+ isNormalRowIdFile(file) && candidates.exists(
+ candidate =>
+ sameBucket(split, candidate.message) &&
+
file.nonNullRowIdRange().hasIntersection(candidate.file.nonNullRowIdRange())))
+ if (filtered.isPresent) Some(filtered.get()) else None
+ })
+ val firstRowIds: immutable.IndexedSeq[Long] = affectedSplits
+ .flatMap(_.dataFiles().asScala)
+ .filter(isNormalRowIdFile)
+ .map(_.firstRowId().longValue())
+ .distinct
+ .sorted
+ .toIndexedSeq
+
+ val rewrittenMessages = candidates
+ .groupBy(staged => staged.file.writeCols().asScala.toSeq)
+ .toSeq
+ .flatMap {
+ case (columnNames, files) =>
+ rewriteFiles(sparkSession, columnNames, files, affectedSplits,
firstRowIds)
+ }
+
+ val candidateKeys = candidates.map(staged => fileKey(staged.message,
staged.file)).toSet
+ val remainingMessages =
+ messageImpls.flatMap(message => withoutCandidates(message,
candidateKeys))
+ Some(RewriteResult(remainingMessages ++ rewrittenMessages,
candidates.size))
+ }
+
+ private def rewriteFiles(
+ sparkSession: SparkSession,
+ columnNames: Seq[String],
+ stagedFiles: Seq[StagedFile],
+ affectedSplits: Seq[DataSplit],
+ firstRowIds: immutable.IndexedSeq[Long]): Seq[CommitMessage] = {
+ val stagedSplits = stagedFiles.map(
+ staged =>
+ DataSplit
+ .builder()
+ .withPartition(staged.message.partition())
+ .withBucket(staged.message.bucket())
+ .withTotalBuckets(staged.message.totalBuckets())
+ .withBucketPath(
+ table
+ .store()
+ .pathFactory()
+ .bucketPath(staged.message.partition(), staged.message.bucket())
+ .toString)
+ .withDataFiles(java.util.Collections.singletonList(staged.file))
+ .rawConvertible(true)
+ .build())
+
+ val relationAttributes = (targetRelation.output ++
targetRelation.metadataOutput).collect {
+ case attribute: AttributeReference => attribute
+ }
+ def attribute(name: String): AttributeReference = {
+ relationAttributes
+ .find(attr => resolver(attr.name, name))
+ .getOrElse(throw new RuntimeException(s"Cannot find column $name for
row-id rewrite."))
+ }
+
+ val rowIdAttribute = attribute(ROW_ID_NAME)
+ val readOutput = columnNames.map(attribute) :+ rowIdAttribute
+ val stagedRelation = createNewScanPlan(stagedSplits, targetRelation)
+ val readPlan = SparkShimLoader.shim.copyDataSourceV2Relation(
+ stagedRelation,
+ stagedRelation.table,
+ readOutput)
+ val firstRowIdUdf = udf((rowId: Long) => floorBinarySearch(firstRowIds,
rowId))
+ val rewrittenRows = createDataset(sparkSession, readPlan)
+ .select((columnNames.map(quotedColumn) :+ quotedColumn(ROW_ID_NAME)): _*)
+ .withColumn(FIRST_ROW_ID_NAME, firstRowIdUdf(quotedColumn(ROW_ID_NAME)))
+ .repartition(col(FIRST_ROW_ID_NAME))
+ .sortWithinPartitions(FIRST_ROW_ID_NAME, ROW_ID_NAME)
+
+ DataEvolutionPaimonWriter(table, affectedSplits)
+ .writePartialFields(rewrittenRows, columnNames)
+ }
+
+ private def rangesAreStillCovered(
+ currentFiles: Seq[CurrentFile],
+ candidates: Seq[StagedFile]): Boolean = {
+ val currentRanges = currentFiles
+ .groupBy(current => bucketKey(current.split))
+ .map {
+ case (key, files) =>
+ key ->
Range.sortAndMergeOverlap(files.map(_.file.nonNullRowIdRange()).asJava, true)
+ }
+ candidates.forall(
+ candidate => {
+ val ranges = currentRanges.getOrElse(
+ bucketKey(candidate.message),
+ java.util.Collections.emptyList[Range]())
+ candidate.file.nonNullRowIdRange().exclude(ranges).isEmpty
+ })
+ }
+
+ private def withoutCandidates(
+ message: CommitMessageImpl,
+ candidates: Set[FileKey]): Option[CommitMessage] = {
+ val increment = message.newFilesIncrement()
+ val newFiles = increment
+ .newFiles()
+ .asScala
+ .filterNot(file => candidates.contains(fileKey(message, file)))
+ .asJava
+ val remaining = new CommitMessageImpl(
+ message.partition(),
+ message.bucket(),
+ message.totalBuckets(),
+ new DataIncrement(
+ newFiles,
+ increment.deletedFiles(),
+ increment.changelogFiles(),
+ increment.newIndexFiles(),
+ increment.deletedIndexFiles()),
+ message.compactIncrement()
+ )
+ if (remaining.isEmpty) None else Some(remaining)
+ }
+}
+
+private[spark] object DataEvolutionRowIdConflictRewriter {
+
+ private val ROW_ID_NAME = "_ROW_ID"
+ private val FIRST_ROW_ID_NAME = "_FIRST_ROW_ID"
+
+ private case class StagedFile(message: CommitMessageImpl, file: DataFileMeta)
+
+ private case class CurrentFile(split: DataSplit, file: DataFileMeta)
+
+ private case class BucketKey(partition: BinaryRow, bucket: Int)
+
+ private case class FileKey(partition: BinaryRow, bucket: Int, fileName:
String)
+
+ private case class RangeKey(partition: BinaryRow, bucket: Int, firstRowId:
Long, rowCount: Long)
+
+ case class RewriteResult(commitMessages: Seq[CommitMessage],
rewrittenFileCount: Int)
+
+ private def isRewriteCandidate(file: DataFileMeta, nextRowId: Long): Boolean
= {
+ isNormalRowIdFile(file) &&
+ file.firstRowId() < nextRowId &&
+ Option(file.writeCols()).exists(
+ columns =>
+ !columns.isEmpty && columns.asScala.forall(column =>
!SpecialFields.isSystemField(column)))
+ }
+
+ private def isNormalRowIdFile(file: DataFileMeta): Boolean = {
+ file.firstRowId() != null && !isDedicatedFile(file)
+ }
+
+ private def isDedicatedFile(file: DataFileMeta): Boolean = {
+ isBlobFile(file.fileName()) || isVectorStoreFile(file.fileName())
+ }
+
+ private def bucketKey(split: DataSplit): BucketKey = {
+ BucketKey(split.partition(), split.bucket())
+ }
+
+ private def bucketKey(message: CommitMessage): BucketKey = {
+ BucketKey(message.partition(), message.bucket())
+ }
+
+ private def sameBucket(split: DataSplit, message: CommitMessage): Boolean = {
+ bucketKey(split) == bucketKey(message)
+ }
+
+ private def fileKey(message: CommitMessage, file: DataFileMeta): FileKey = {
+ FileKey(message.partition(), message.bucket(), file.fileName())
+ }
+
+ private def rangeKey(split: DataSplit, file: DataFileMeta): RangeKey = {
+ RangeKey(split.partition(), split.bucket(), file.firstRowId(),
file.rowCount())
+ }
+
+ private def rangeKey(message: CommitMessage, file: DataFileMeta): RangeKey =
{
+ RangeKey(message.partition(), message.bucket(), file.firstRowId(),
file.rowCount())
+ }
+
+ private def quotedColumn(name: String) = {
+ col("`" + name.replace("`", "``") + "`")
+ }
+
+ private def floorBinarySearch(firstRowIds: immutable.IndexedSeq[Long],
rowId: Long): Long = {
+ val index =
+ java.util.Collections.binarySearch(firstRowIds.map(Long.box).asJava,
Long.box(rowId))
+ if (index >= 0) {
+ firstRowIds(index)
+ } else {
+ val insertionPoint = -index - 1
+ if (insertionPoint == 0) {
+ throw new IllegalArgumentException(
+ s"Row ID $rowId is less than the first current row ID boundary.")
+ }
+ firstRowIds(insertionPoint - 1)
+ }
+ }
+}
+
+private[spark] object DataEvolutionRowIdConflictCommitter {
+
+ private val LOG = LoggerFactory.getLogger(getClass)
+
+ def commit(
+ sparkSession: SparkSession,
+ table: FileStoreTable,
+ targetRelation: DataSourceV2Relation,
+ writer: PaimonSparkWriter,
+ updateMessages: Seq[CommitMessage],
+ otherMessages: Seq[CommitMessage],
+ readSnapshotId: Long,
+ operation: Snapshot.Operation): Unit = {
+ var currentUpdateMessages = updateMessages
+ var retryCount = 0
+ val startMillis = System.currentTimeMillis()
+ val options = table.coreOptions()
+ val retryWaiter = new RetryWaiter(options.commitMinRetryWait(),
options.commitMaxRetryWait())
+ val rewriter = new DataEvolutionRowIdConflictRewriter(table,
targetRelation)
+
+ val latestBeforeCommit = table.snapshotManager().latestSnapshot()
+ if (latestBeforeCommit != null && latestBeforeCommit.id() !=
readSnapshotId) {
+ rewriter.rewrite(sparkSession, latestBeforeCommit,
currentUpdateMessages).foreach {
+ result =>
+ currentUpdateMessages = result.commitMessages
+ logRewrite(table, latestBeforeCommit, result)
+ }
+ }
+
+ while (true) {
+ try {
+ writer.commit(currentUpdateMessages ++ otherMessages, operation)
+ return
+ } catch {
+ case conflict: RuntimeException if isRowIdExistenceConflict(conflict)
=>
+ val elapsedBeforeRewrite = System.currentTimeMillis() - startMillis
+ if (
+ elapsedBeforeRewrite > options.commitTimeout() ||
+ retryCount >= options.commitMaxRetries()
+ ) {
+ throw conflict
+ }
+
+ val latestSnapshot = table.snapshotManager().latestSnapshot()
+ if (latestSnapshot == null) {
+ throw conflict
+ }
+
+ val rewriteResult =
+ try {
+ rewriter.rewrite(sparkSession, latestSnapshot,
currentUpdateMessages)
+ } catch {
+ case rewriteError: RuntimeException =>
+ throw new RuntimeException(
+ s"${conflict.getMessage} ${rewriteError.getMessage}",
+ conflict)
+ }
+ if (rewriteResult.isEmpty) {
+ throw conflict
+ }
+
+ currentUpdateMessages = rewriteResult.get.commitMessages
+ val elapsedMillis = System.currentTimeMillis() - startMillis
+ if (elapsedMillis > options.commitTimeout()) {
+ throw conflict
+ }
+
+ logRewrite(table, latestSnapshot, rewriteResult.get)
+ retryWaiter.retryWait(retryCount)
+ retryCount += 1
+ }
+ }
+ }
+
+ private def isRowIdExistenceConflict(error: Throwable): Boolean = {
+ ExceptionUtils.findThrowable(error,
classOf[RowIdExistenceConflictException]).isPresent
+ }
+
+ private def logRewrite(
+ table: FileStoreTable,
+ snapshot: Snapshot,
+ result: DataEvolutionRowIdConflictRewriter.RewriteResult): Unit = {
+ LOG.info(
+ "Rewrote {} stale row-id file(s) against snapshot {} before committing
to table {}.",
+ Int.box(result.rewrittenFileCount),
+ Long.box(snapshot.id()),
+ table.name()
+ )
+ }
+}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala
index 479515ca46..e739036303 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala
@@ -287,9 +287,16 @@ case class MergeIntoPaimonDataEvolutionTable(
if (readSnapshot != null) {
writer.rowIdCheckConflict(readSnapshot.id())
}
- writer.commit(
- matchedResult.commitMessages ++ deleteCommit ++ insertCommit,
- Snapshot.Operation.MERGE)
+ DataEvolutionRowIdConflictCommitter.commit(
+ sparkSession,
+ table,
+ targetRelation,
+ writer,
+ matchedResult.commitMessages,
+ deleteCommit ++ insertCommit,
+ if (readSnapshot == null) -1L else readSnapshot.id(),
+ Snapshot.Operation.MERGE
+ )
} finally {
targetActionCleanup()
if (persistSourceDss.isDefined) {
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/RowTrackingTestBase.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/RowTrackingTestBase.scala
index ba66e09697..7c40937434 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/RowTrackingTestBase.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/RowTrackingTestBase.scala
@@ -18,11 +18,13 @@
package org.apache.paimon.spark.sql
-import org.apache.paimon.Snapshot.CommitKind
+import org.apache.paimon.Snapshot.{CommitKind, Operation}
import org.apache.paimon.errors.ErrorMessages
import org.apache.paimon.globalindex.IndexedSplit
import org.apache.paimon.spark.PaimonMetrics.RESULTED_TABLE_FILES
import org.apache.paimon.spark.PaimonSparkTestBase
+import org.apache.paimon.spark.catalyst.analysis.PaimonRelation
+import org.apache.paimon.spark.commands.{DataEvolutionPaimonWriter,
DataEvolutionRowIdConflictCommitter, PaimonSparkWriter}
import org.apache.paimon.spark.read.PaimonSplitScan
import org.apache.paimon.table.source.DataSplit
@@ -32,6 +34,7 @@ import org.apache.spark.sql.connector.metric.CustomTaskMetric
import org.apache.spark.sql.execution.QueryExecution
import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
+import org.apache.spark.sql.functions.{col, udf}
import org.apache.spark.sql.paimon.Utils
import org.apache.spark.sql.util.QueryExecutionListener
@@ -132,6 +135,60 @@ abstract class RowTrackingTestBase extends
PaimonSparkTestBase with AdaptiveSpar
}
}
+ test("Data Evolution: rebase staged merge updates after concurrent compact")
{
+ withTable("t") {
+ sql(s"""
+ |CREATE TABLE t (id INT, b INT) TBLPROPERTIES (
+ | 'row-tracking.enabled' = 'true',
+ | 'compaction.min.file-num' = '2',
+ | 'commit.max-retries' = '0',
+ | 'data-evolution.row-id-conflict-rewrite.max-size' = '0 B',
+ | 'data-evolution.enabled' = 'true')
+ |""".stripMargin)
+ sql("INSERT INTO t VALUES (1, 10)")
+ sql("INSERT INTO t VALUES (2, 20)")
+
+ val table = loadTable("t")
+ val readSnapshot = table.latestSnapshot().get()
+ val dataSplits = table
+ .newSnapshotReader()
+ .withSnapshot(readSnapshot)
+ .read()
+ .splits()
+ .asScala
+ .collect { case split: DataSplit => split }
+ .toSeq
+ val firstRowIds = dataSplits
+ .flatMap(_.dataFiles().asScala)
+ .map(_.firstRowId().longValue())
+ .sorted
+ val firstRowId = udf((rowId: Long) => firstRowIds.takeWhile(_ <=
rowId).last)
+ val stagedRows = sql("SELECT b + 1 AS b, _ROW_ID FROM t")
+ .withColumn("_FIRST_ROW_ID", firstRowId(col("_ROW_ID")))
+ .select("b", "_FIRST_ROW_ID", "_ROW_ID")
+ val stagedUpdates =
+ DataEvolutionPaimonWriter(table,
dataSplits).writePartialFields(stagedRows, Seq("b"))
+
+ sql("CALL sys.compact(table => 't')").collect()
+
+ val writer = PaimonSparkWriter(table)
+ writer.rowIdCheckConflict(readSnapshot.id())
+ val targetRelation =
+
PaimonRelation.getPaimonRelation(spark.table("t").queryExecution.analyzed)
+ DataEvolutionRowIdConflictCommitter.commit(
+ spark,
+ table,
+ targetRelation,
+ writer,
+ stagedUpdates,
+ Nil,
+ readSnapshot.id(),
+ Operation.MERGE)
+
+ checkAnswer(sql("SELECT id, b FROM t ORDER BY id"), Seq(Row(1, 11),
Row(2, 21)))
+ }
+ }
+
test("Data Evolution: concurrent merge and merge") {
withTable("s", "t") {
sql(s"""
