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 47a220fed7 [spark][python] Preserve rows when rebasing stale updates
(#9096)
47a220fed7 is described below
commit 47a220fed78461ae2cb7a71e06a521f4520026de
Author: Jingsong Lee <[email protected]>
AuthorDate: Fri Aug 7 17:47:31 2026 +0800
[spark][python] Preserve rows when rebasing stale updates (#9096)
---
.../pypaimon/tests/table_upsert_by_key_test.py | 60 ++++++++++++++++++++
paimon-python/pypaimon/write/file_store_commit.py | 4 +-
.../DataEvolutionRowIdConflictRewriter.scala | 64 ++++++++++++----------
.../paimon/spark/sql/RowTrackingTestBase.scala | 61 ++++++++++++++++++++-
4 files changed, 158 insertions(+), 31 deletions(-)
diff --git a/paimon-python/pypaimon/tests/table_upsert_by_key_test.py
b/paimon-python/pypaimon/tests/table_upsert_by_key_test.py
index feb98ae3c6..7f06fcab42 100644
--- a/paimon-python/pypaimon/tests/table_upsert_by_key_test.py
+++ b/paimon-python/pypaimon/tests/table_upsert_by_key_test.py
@@ -620,6 +620,66 @@ class _TableUpsertByKeyTestBase(DataEvolutionTestBase):
}
self.assertEqual(99, rows[2])
+ def test_compaction_rewrite_rejects_update_after_compaction(self):
+ table = self._create_table()
+ self._write_arrow(table, pa.Table.from_pydict({
+ 'id': [1, 2],
+ 'name': ['Alice', 'Bob'],
+ 'age': [25, 30],
+ 'city': ['NYC', 'LA'],
+ }, schema=self.pa_schema))
+ self._write_arrow(table, pa.Table.from_pydict({
+ 'id': [3, 4],
+ 'name': ['Carol', 'Dave'],
+ 'age': [35, 40],
+ 'city': ['Chicago', 'Houston'],
+ }, schema=self.pa_schema))
+
+ wb = self._make_write_builder(table)
+ update = wb.new_update().with_update_type(['age'])
+ commit_identifier = self._next_commit_id()
+ stale_messages = self._apply_upsert(
+ update,
+ pa.Table.from_pydict({
+ 'id': [2],
+ 'name': ['ignored'],
+ 'age': [31],
+ 'city': ['ignored'],
+ }, schema=self.pa_schema),
+ ['id'],
+ commit_identifier,
+ )
+
+ self._compact_all_data_files(table)
+ self._upsert(
+ table,
+ pa.Table.from_pydict({
+ 'id': [2],
+ 'name': ['ignored'],
+ 'age': [99],
+ 'city': ['ignored'],
+ }, schema=self.pa_schema),
+ ['id'],
+ update_cols=['age'],
+ )
+
+ commit = wb.new_commit()
+ with self.assertRaisesRegex(
+ RuntimeError,
+ "multiple 'MERGE INTO' operations have encountered conflicts"):
+ self._apply_commit(
+ commit,
+ stale_messages,
+ commit_identifier,
+ )
+ commit.close()
+
+ rows = {
+ row['id']: row['age']
+ for row in self._read_all(table).to_pylist()
+ }
+ self.assertEqual(99, rows[2])
+
def test_large_table_upsert(self):
"""Upsert that touches a wide selection of rows in a 200-row table."""
table = self._create_table()
diff --git a/paimon-python/pypaimon/write/file_store_commit.py
b/paimon-python/pypaimon/write/file_store_commit.py
index 28fbd441a9..b1f212283d 100644
--- a/paimon-python/pypaimon/write/file_store_commit.py
+++ b/paimon-python/pypaimon/write/file_store_commit.py
@@ -851,7 +851,9 @@ class FileStoreCommit:
)
)
if non_compaction_conflict is not None:
- return None
+ raise CommitConflictError(
+ str(non_compaction_conflict)
+ ) from non_compaction_conflict
try:
return RowIdConflictRewriter(
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
index 5512c735fb..ebb9402b1a 100644
---
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
@@ -106,30 +106,12 @@ private[spark] class DataEvolutionRowIdConflictRewriter(
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)
+ rewriteFiles(sparkSession, columnNames, files, currentSplits)
}
val candidateKeys = candidates.map(staged => fileKey(staged.message,
staged.file)).toSet
@@ -142,8 +124,7 @@ private[spark] class DataEvolutionRowIdConflictRewriter(
sparkSession: SparkSession,
columnNames: Seq[String],
stagedFiles: Seq[StagedFile],
- affectedSplits: Seq[DataSplit],
- firstRowIds: immutable.IndexedSeq[Long]): Seq[CommitMessage] = {
+ currentSplits: Seq[DataSplit]): Seq[CommitMessage] = {
val stagedSplits = stagedFiles.map(
staged =>
DataSplit
@@ -160,6 +141,23 @@ private[spark] class DataEvolutionRowIdConflictRewriter(
.withDataFiles(java.util.Collections.singletonList(staged.file))
.rawConvertible(true)
.build())
+ val affectedSplits = currentSplits.flatMap(
+ split => {
+ val filtered = split.filterDataFile(
+ file =>
+ isNormalRowIdFile(file) && stagedFiles.exists(
+ staged =>
+ sameBucket(split, staged.message) &&
+
file.nonNullRowIdRange().hasIntersection(staged.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 relationAttributes = (targetRelation.output ++
targetRelation.metadataOutput).collect {
case attribute: AttributeReference => attribute
@@ -172,14 +170,24 @@ private[spark] class DataEvolutionRowIdConflictRewriter(
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)
+ def readRows(splits: Seq[DataSplit]) = {
+ val relation = createNewScanPlan(splits, targetRelation)
+ val readPlan =
+ SparkShimLoader.shim.copyDataSourceV2Relation(relation,
relation.table, readOutput)
+ createDataset(sparkSession, readPlan)
+ .select((columnNames.map(quotedColumn) :+ quotedColumn(ROW_ID_NAME)):
_*)
+ }
+
+ val stagedRows = readRows(stagedSplits)
+ val currentRows = readRows(affectedSplits)
+ // A new compacted row-id range may contain rows outside the staged file.
Preserve their
+ // latest values and only replace rows for which the staged update has a
value.
+ val mergedRows = currentRows
+ .join(stagedRows.select(quotedColumn(ROW_ID_NAME)), Seq(ROW_ID_NAME),
"left_anti")
+ .unionByName(stagedRows)
.select((columnNames.map(quotedColumn) :+ quotedColumn(ROW_ID_NAME)): _*)
+ val firstRowIdUdf = udf((rowId: Long) => floorBinarySearch(firstRowIds,
rowId))
+ val rewrittenRows = mergedRows
.withColumn(FIRST_ROW_ID_NAME, firstRowIdUdf(quotedColumn(ROW_ID_NAME)))
.repartition(col(FIRST_ROW_ID_NAME))
.sortWithinPartitions(FIRST_ROW_ID_NAME, ROW_ID_NAME)
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 7c40937434..60527e913b 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
@@ -163,7 +163,7 @@ abstract class RowTrackingTestBase extends
PaimonSparkTestBase with AdaptiveSpar
.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")
+ val stagedRows = sql("SELECT b + 1 AS b, _ROW_ID FROM t WHERE id = 1")
.withColumn("_FIRST_ROW_ID", firstRowId(col("_ROW_ID")))
.select("b", "_FIRST_ROW_ID", "_ROW_ID")
val stagedUpdates =
@@ -185,7 +185,64 @@ abstract class RowTrackingTestBase extends
PaimonSparkTestBase with AdaptiveSpar
readSnapshot.id(),
Operation.MERGE)
- checkAnswer(sql("SELECT id, b FROM t ORDER BY id"), Seq(Row(1, 11),
Row(2, 21)))
+ checkAnswer(sql("SELECT id, b FROM t ORDER BY id"), Seq(Row(1, 11),
Row(2, 20)))
+ }
+ }
+
+ test("Data Evolution: rebase rejects same-column update 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.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 WHERE id = 1")
+ .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()
+ sql("UPDATE t SET b = 99 WHERE id = 1").collect()
+
+ val writer = PaimonSparkWriter(table)
+ writer.rowIdCheckConflict(readSnapshot.id())
+ val targetRelation =
+
PaimonRelation.getPaimonRelation(spark.table("t").queryExecution.analyzed)
+ val exception = intercept[RuntimeException] {
+ DataEvolutionRowIdConflictCommitter.commit(
+ spark,
+ table,
+ targetRelation,
+ writer,
+ stagedUpdates,
+ Nil,
+ readSnapshot.id(),
+ Operation.MERGE)
+ }
+
+ assert(hasMessage(exception,
ErrorMessages.DATA_EVOLUTION_ROW_ID_CONFLICT_MESSAGE))
+ checkAnswer(sql("SELECT id, b FROM t ORDER BY id"), Seq(Row(1, 99),
Row(2, 20)))
}
}
