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 db5b308f77 [spark] Fix inconsistent snapshots for data evolution V1
UPDATE (#9597)
db5b308f77 is described below
commit db5b308f77a5b4468740ca6a2ab706ea10f55bc8
Author: zhoulii <[email protected]>
AuthorDate: Mon Sep 7 22:45:09 2026 +0800
[spark] Fix inconsistent snapshots for data evolution V1 UPDATE (#9597)
---
.../cdc/mysql/MySqlSyncDatabaseActionITCase.java | 18 +++++
.../MergeIntoPaimonDataEvolutionTable.scala | 14 ++--
.../MergeIntoPaimonDataEvolutionTable.scala | 14 ++--
.../UpdatePaimonDataEvolutionTableCommand.scala | 3 +-
.../commands/DataEvolutionUpdateSnapshotTest.scala | 81 ++++++++++++++++++++++
.../paimon/spark/sql/RowTrackingTestBase.scala | 45 +++++++++++-
6 files changed, 162 insertions(+), 13 deletions(-)
diff --git
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseActionITCase.java
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseActionITCase.java
index 52464368b3..e0b2344cf7 100644
---
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseActionITCase.java
+++
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseActionITCase.java
@@ -1028,6 +1028,22 @@ public class MySqlSyncDatabaseActionITCase extends
MySqlActionITCaseBase {
rowType,
Collections.singletonList("k"));
+ // Use records from both shards as a barrier before altering t2.
Waiting for t1 alone
+ // does not guarantee that the t2 snapshot readers have loaded
their schemas.
+ statement.executeUpdate("INSERT INTO database_shard_1.t2 VALUES
(-1, -1.1)");
+ statement.executeUpdate("INSERT INTO database_shard_2.t2 VALUES
(-2, -2.2)");
+ table = getFileStoreTable("t2");
+ rowType =
+ RowType.of(
+ new DataType[] {DataTypes.BIGINT().notNull(),
DataTypes.DOUBLE()},
+ new String[] {"k", "v1"});
+ waitForResult(
+ client,
+ Arrays.asList("+I[-1, -1.1]", "+I[-2, -2.2]"),
+ table,
+ rowType,
+ Collections.singletonList("k"));
+
// test schema evolution of t2
statement.executeUpdate("ALTER TABLE database_shard_1.t2 ADD
COLUMN v2 INT");
statement.executeUpdate("ALTER TABLE database_shard_2.t2 ADD
COLUMN v3 VARCHAR(10)");
@@ -1048,6 +1064,8 @@ public class MySqlSyncDatabaseActionITCase extends
MySqlActionITCaseBase {
waitForResult(
client,
Arrays.asList(
+ "+I[-1, -1.1, NULL, NULL]",
+ "+I[-2, -2.2, NULL, NULL]",
"+I[1, 1.1, 1, NULL]",
"+I[2, 2.2, 2, NULL]",
"+I[3, 3.3, NULL, db2_3]",
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 88deb9817c..c6db56419b 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
@@ -73,7 +73,8 @@ case class MergeIntoPaimonDataEvolutionTable(
matchedCondition: Expression,
matchedActions: Seq[MergeAction],
notMatchedActions: Seq[MergeAction],
- notMatchedBySourceActions: Seq[MergeAction])
+ notMatchedBySourceActions: Seq[MergeAction],
+ allowPinnedSelfMergeShortcut: Boolean = false)
extends PaimonRowLevelCommand
with Logging {
@@ -157,7 +158,7 @@ case class MergeIntoPaimonDataEvolutionTable(
private def passthroughSourceRelation(plan: LogicalPlan):
Option[DataSourceV2Relation] = {
EliminateSubqueryAliases(plan) match {
- case relation: DataSourceV2Relation if
isPaimonRelationWithoutTimeTravel(relation) =>
+ case relation: DataSourceV2Relation if
isEligibleSelfMergeSource(relation) =>
Some(relation)
case Project(projectList, child) if isPassthroughProject(projectList,
child) =>
passthroughSourceRelation(child)
@@ -166,10 +167,13 @@ case class MergeIntoPaimonDataEvolutionTable(
}
}
- private def isPaimonRelationWithoutTimeTravel(relation:
DataSourceV2Relation): Boolean =
+ private def isEligibleSelfMergeSource(relation: DataSourceV2Relation):
Boolean =
relation.table match {
case sparkTable: SparkTable =>
-
!TimeTravelUtil.hasTimeTravelOptions(Options.fromMap(sparkTable.getTable.options()))
+
!TimeTravelUtil.hasTimeTravelOptions(Options.fromMap(sparkTable.getTable.options()))
||
+ // Reference equality proves that both sides use the exact same pinned
table and snapshot.
+ // Keep this exception scoped to callers that explicitly construct
such a self-merge.
+ (allowPinnedSelfMergeShortcut && (sparkTable eq targetSparkTable))
case _ => false
}
@@ -286,7 +290,7 @@ case class MergeIntoPaimonDataEvolutionTable(
}
private def invokeMergeInto(sparkSession: SparkSession): Unit = {
- val readSnapshot = table.snapshotManager().latestSnapshot()
+ val readSnapshot = TimeTravelUtil.tryTravelOrLatest(table)
val snapshotReader = table.newSnapshotReader()
if (readSnapshot != null) {
snapshotReader.withSnapshot(readSnapshot)
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 7bbc1487b6..f40db70fc1 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
@@ -73,7 +73,8 @@ case class MergeIntoPaimonDataEvolutionTable(
matchedCondition: Expression,
matchedActions: Seq[MergeAction],
notMatchedActions: Seq[MergeAction],
- notMatchedBySourceActions: Seq[MergeAction])
+ notMatchedBySourceActions: Seq[MergeAction],
+ allowPinnedSelfMergeShortcut: Boolean = false)
extends PaimonRowLevelCommand
with Logging {
@@ -157,7 +158,7 @@ case class MergeIntoPaimonDataEvolutionTable(
private def passthroughSourceRelation(plan: LogicalPlan):
Option[DataSourceV2Relation] = {
EliminateSubqueryAliases(plan) match {
- case relation: DataSourceV2Relation if
isPaimonRelationWithoutTimeTravel(relation) =>
+ case relation: DataSourceV2Relation if
isEligibleSelfMergeSource(relation) =>
Some(relation)
case Project(projectList, child) if isPassthroughProject(projectList,
child) =>
passthroughSourceRelation(child)
@@ -166,10 +167,13 @@ case class MergeIntoPaimonDataEvolutionTable(
}
}
- private def isPaimonRelationWithoutTimeTravel(relation:
DataSourceV2Relation): Boolean =
+ private def isEligibleSelfMergeSource(relation: DataSourceV2Relation):
Boolean =
relation.table match {
case sparkTable: SparkTable =>
-
!TimeTravelUtil.hasTimeTravelOptions(Options.fromMap(sparkTable.getTable.options()))
+
!TimeTravelUtil.hasTimeTravelOptions(Options.fromMap(sparkTable.getTable.options()))
||
+ // Reference equality proves that both sides use the exact same pinned
table and snapshot.
+ // Keep this exception scoped to callers that explicitly construct
such a self-merge.
+ (allowPinnedSelfMergeShortcut && (sparkTable eq targetSparkTable))
case _ => false
}
@@ -286,7 +290,7 @@ case class MergeIntoPaimonDataEvolutionTable(
}
private def invokeMergeInto(sparkSession: SparkSession): Unit = {
- val readSnapshot = table.snapshotManager().latestSnapshot()
+ val readSnapshot = TimeTravelUtil.tryTravelOrLatest(table)
val snapshotReader = table.newSnapshotReader()
if (readSnapshot != null) {
snapshotReader.withSnapshot(readSnapshot)
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/UpdatePaimonDataEvolutionTableCommand.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/UpdatePaimonDataEvolutionTableCommand.scala
index daaef27a85..d3506ecec2 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/UpdatePaimonDataEvolutionTableCommand.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/UpdatePaimonDataEvolutionTableCommand.scala
@@ -84,7 +84,8 @@ case class UpdatePaimonDataEvolutionTableCommand(
matchedCondition,
Seq(updateAction),
Nil,
- Nil).run(sparkSession)
+ Nil,
+ allowPinnedSelfMergeShortcut = true).run(sparkSession)
}
private def deterministicUpdate: Boolean = {
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/commands/DataEvolutionUpdateSnapshotTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/commands/DataEvolutionUpdateSnapshotTest.scala
new file mode 100644
index 0000000000..7615c94c31
--- /dev/null
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/commands/DataEvolutionUpdateSnapshotTest.scala
@@ -0,0 +1,81 @@
+/*
+ * 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.errors.ErrorMessages
+import org.apache.paimon.spark.PaimonSparkTestBase
+
+import org.apache.spark.sql.catalyst.QueryPlanningTracker
+
+import scala.util.Try
+
+class DataEvolutionUpdateSnapshotTest extends PaimonSparkTestBase {
+
+ test("V1 update detects a concurrent update after its snapshot is pinned") {
+ withSparkSQLConf(
+ "spark.paimon.write.use-v2-write" -> "false",
+ "spark.paimon.write.data-evolution.update-conflict-retry.max-attempts"
-> "1"
+ ) {
+ withTable("t") {
+ sql(
+ "CREATE TABLE t (id INT, status STRING) TBLPROPERTIES " +
+ "('row-tracking.enabled' = 'true', 'data-evolution.enabled' =
'true')")
+ sql("INSERT INTO t VALUES (1, 'pending')")
+
+ val parsed = spark.sessionState.sqlParser.parsePlan(
+ "UPDATE t SET status = 'done' WHERE status = 'pending'")
+ val updateCommand = spark.sessionState.analyzer
+ .executeAndCheck(parsed, new QueryPlanningTracker)
+ .asInstanceOf[UpdatePaimonDataEvolutionTableCommand]
+
+ // Materialize the same pinned table produced at the beginning of
runOnce, then commit a
+ // conflicting update before MergeIntoPaimonDataEvolutionTable
resolves its target snapshot.
+ val (pinnedTable, pinnedRelation) =
+ MergeIntoPaimonDataEvolutionTable.withMatchedUpdateScanOptions(
+ updateCommand.v2Table,
+ updateCommand.relation)
+ val pinnedUpdate = updateCommand.copy(v2Table = pinnedTable, relation
= pinnedRelation)
+
+ sql("UPDATE t SET status = 'cancelled' WHERE id = 1")
+
+ val result = Try(pinnedUpdate.run(spark))
+ val finalStatus = sql("SELECT status FROM t WHERE id =
1").head().getString(0)
+ val detectedConflict = result.failed.toOption.exists(
+ hasMessage(_, ErrorMessages.DATA_EVOLUTION_ROW_ID_CONFLICT_MESSAGE))
+
+ assert(
+ detectedConflict && finalStatus == "cancelled",
+ s"Expected a row-id conflict and final status 'cancelled', but got "
+
+ s"failure=${result.failed.toOption.map(_.toString)},
finalStatus=$finalStatus"
+ )
+ }
+ }
+ }
+
+ private def hasMessage(throwable: Throwable, expected: String): Boolean = {
+ var current = throwable
+ while (current != null) {
+ if (Option(current.getMessage).exists(_.contains(expected))) {
+ return true
+ }
+ current = current.getCause
+ }
+ false
+ }
+}
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 834eb9c372..a0a552baf4 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
@@ -1691,7 +1691,13 @@ abstract class RowTrackingTestBase extends
PaimonSparkTestBase with AdaptiveSpar
"CREATE TABLE t (id INT, b INT, c INT) TBLPROPERTIES
('row-tracking.enabled' = 'true', 'data-evolution.enabled' = 'true')")
sql("INSERT INTO t SELECT /*+ REPARTITION(1) */ id, id AS b, id AS c
FROM range(2, 4)")
- sql("UPDATE t SET b = 22 WHERE id = 2")
+ val (mergeRowsPlans, _) =
+ executeMergeIntoAndCollectPlans("UPDATE t SET b = 22 WHERE id = 2")
+ assert(
+ mergeRowsPlans.exists(_.collectFirst { case _: Join => true
}.nonEmpty),
+ s"Expected conditional UPDATE to use the general MERGE plan, but
got: " +
+ mergeRowsPlans.mkString("\n")
+ )
checkAnswer(
sql("SELECT *, _ROW_ID, _SEQUENCE_NUMBER FROM t ORDER BY id"),
Seq(Row(2, 22, 2, 0, 2), Row(3, 3, 3, 1, 2))
@@ -1794,7 +1800,8 @@ abstract class RowTrackingTestBase extends
PaimonSparkTestBase with AdaptiveSpar
"CREATE TABLE t (id INT, b INT, c INT) TBLPROPERTIES
('row-tracking.enabled' = 'true', 'data-evolution.enabled' = 'true')")
sql("INSERT INTO t SELECT /*+ REPARTITION(1) */ id, id AS b, id AS c
FROM range(2, 4)")
- sql("UPDATE t SET b = 22")
+ val (mergeRowsPlans, _) = executeMergeIntoAndCollectPlans("UPDATE t
SET b = 22")
+ assertSelfMergeShortcut(mergeRowsPlans)
checkAnswer(
sql("SELECT *, _ROW_ID, _SEQUENCE_NUMBER FROM t ORDER BY id"),
Seq(Row(2, 22, 2, 0, 2), Row(3, 22, 3, 1, 2))
@@ -1803,6 +1810,40 @@ abstract class RowTrackingTestBase extends
PaimonSparkTestBase with AdaptiveSpar
}
}
+ test("Data Evolution: V1 update with user-specified snapshot uses self-merge
shortcut") {
+ withSparkSQLConf("spark.paimon.write.use-v2-write" -> "false") {
+ withTable("t") {
+ sql(
+ "CREATE TABLE t (id INT, b INT) TBLPROPERTIES
('row-tracking.enabled' = 'true', 'data-evolution.enabled' = 'true')")
+ sql("INSERT INTO t VALUES (1, 10), (2, 20)")
+ val snapshotId = loadTable("t").snapshotManager().latestSnapshotId()
+ sql("INSERT INTO t VALUES (3, 30)")
+
+ var mergeRowsPlans = Seq.empty[LogicalPlan]
+ withSparkSQLConf("spark.paimon.scan.snapshot-id" ->
snapshotId.toString) {
+ mergeRowsPlans = executeMergeIntoAndCollectPlans("UPDATE t SET b =
100")._1
+ }
+ assertSelfMergeShortcut(mergeRowsPlans)
+
+ checkAnswer(
+ sql("SELECT id, b FROM t ORDER BY id"),
+ Seq(Row(1, 100), Row(2, 100), Row(3, 30)))
+ }
+ }
+ }
+
+ private def assertSelfMergeShortcut(mergeRowsPlans: Seq[LogicalPlan]): Unit
= {
+ assert(mergeRowsPlans.nonEmpty, "Expected a MergeRows plan for V1 UPDATE.")
+ assert(
+ mergeRowsPlans.forall(_.collectFirst {
+ case p: Join => p
+ case p: Sort => p
+ case p: RepartitionByExpression => p
+ }.isEmpty),
+ s"Found unexpected Join/Sort/Exchange in plans:
${mergeRowsPlans.mkString("\n")}"
+ )
+ }
+
test("Data Evolution: V1 update retries concurrent update conflicts") {
withSparkSQLConf(
"spark.paimon.write.use-v2-write" -> "false",
