[spark] Fix inconsistent snapshots for data evolution V1 UPDATE

paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseActionITCase.java

@@ -1028,6 +1028,22 @@ public void testSyncMultipleShards() throws Exception {
                     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 void testSyncMultipleShards() throws Exception {
             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]",

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)

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)

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 = {

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
+  }
+}

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",