[iceberg] Fix Iceberg metadata after a rollback

paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java

@@ -24,6 +24,7 @@
 import org.apache.paimon.consumer.ConsumerManager;
 import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.fs.Path;
+import org.apache.paimon.iceberg.IcebergCommitCallback;
 import org.apache.paimon.manifest.IndexManifestEntry;
 import org.apache.paimon.manifest.ManifestEntry;
 import org.apache.paimon.manifest.ManifestFileMeta;
@@ -518,6 +519,7 @@ private Optional<TableSchema> tryTimeTravel(Options options) {
 
     @Override
     public void rollbackTo(long snapshotId) {
+        IcebergCommitCallback.markRetirePendingForRollback(this);
         SnapshotManager snapshotManager = snapshotManager();
         try {
             snapshotManager.rollback(Instant.snapshot(snapshotId));
@@ -543,6 +545,7 @@ public void rollbackTo(long snapshotId) {
 
     @Override
     public void rollbackTo(String tagName) {
+        IcebergCommitCallback.markRetirePendingForRollback(this);
         SnapshotManager snapshotManager = snapshotManager();
         try {
             snapshotManager.rollback(Instant.tag(tagName));

paimon-core/src/test/java/org/apache/paimon/iceberg/RecordingIcebergMetadataCommitter.java

@@ -0,0 +1,78 @@
+/*
+ * 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.iceberg;
+
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.iceberg.metadata.IcebergMetadata;
+import org.apache.paimon.table.FileStoreTable;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+/** An {@link IcebergMetadataCommitter} recording every commit, for tests. */
+public class RecordingIcebergMetadataCommitter implements IcebergMetadataCommitter {
+
+    public static final List<Object> COMMITS = Collections.synchronizedList(new ArrayList<>());
+    public static final List<Object> BASES = Collections.synchronizedList(new ArrayList<>());
+    public static volatile boolean failNextCommit = false;
+
+    private static void maybeFail() {
+        if (failNextCommit) {
+            failNextCommit = false;
+            throw new RuntimeException("injected catalog failure");
+        }
+    }
+
+    @Override
+    public String identifier() {
+        return "hive";
+    }
+
+    @Override
+    public void commitMetadata(Path newMetadataPath, @Nullable Path baseMetadataPath) {
+        maybeFail();
+        COMMITS.add(newMetadataPath);
+        BASES.add(baseMetadataPath);
+    }
+
+    @Override
+    public void commitMetadata(
+            IcebergMetadata newIcebergMetadata, @Nullable IcebergMetadata baseIcebergMetadata) {
+        maybeFail();
+        COMMITS.add(newIcebergMetadata);
+        BASES.add(baseIcebergMetadata);
+    }
+
+    /** Registered under hadoop-catalog: no real committer exists there, so no ambiguity. */
+    public static class Factory implements IcebergMetadataCommitterFactory {
+
+        @Override
+        public String identifier() {
+            return IcebergOptions.StorageType.HADOOP_CATALOG.toString();
+        }
+
+        @Override
+        public IcebergMetadataCommitter create(FileStoreTable table) {
+            return new RecordingIcebergMetadataCommitter();
+        }
+    }
+}

paimon-core/src/test/resources/META-INF/services/org.apache.paimon.factories.Factory

@@ -16,3 +16,4 @@
 org.apache.paimon.mergetree.compact.aggregate.TestCustomAggFactory
 org.apache.paimon.mergetree.compact.aggregate.TestMapOnlyAggFactory
 org.apache.paimon.rest.auth.CustomTestDLFTokenLoaderFactory
+org.apache.paimon.iceberg.RecordingIcebergMetadataCommitter$Factory

paimon-iceberg/src/main/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitter.java

@@ -71,6 +71,8 @@
  */
 public class IcebergRestMetadataCommitter implements IcebergMetadataCommitter {
 
+    private static final String PAIMON_COMMIT_IDENTITY = "paimon-commit-identity";
+
     private static final Logger LOG = LoggerFactory.getLogger(IcebergRestMetadataCommitter.class);
 
     private static final String REST_CATALOG_NAME = "rest-catalog";
@@ -160,6 +162,23 @@ private void commitMetadataImpl(
 
                 TableMetadata metadata = ((BaseTable) icebergTable).operations().current();
 
+                org.apache.iceberg.Snapshot catalogCurrent = metadata.currentSnapshot();
+                org.apache.iceberg.Snapshot newCurrent = newMetadata.currentSnapshot();
+                if (catalogCurrent != null
+                        && newCurrent != null
+                        && catalogCurrent.snapshotId() == newCurrent.snapshotId()
+                        && java.util.Objects.equals(
+                                catalogCurrent.summary().get(PAIMON_COMMIT_IDENTITY),
+                                newCurrent.summary().get(PAIMON_COMMIT_IDENTITY))) {
+                    // an idempotent retry: the catalog is already at this snapshot; rebuilding
+                    // through updatesForIncorrectBase would drop and recreate the table
+                    LOG.info(
+                            "Iceberg table {} is already at snapshot {}, nothing to commit.",
+                            icebergTableIdentifier,
+                            newCurrent.snapshotId());
+                    return;
+                }
+
                 if (metadata.currentSnapshot() == null) {
                     // Table exists in the REST catalog but has no snapshots yet. This happens
                     // when a previous createTable() or recreateTable() succeeded but the
@@ -229,19 +248,16 @@ private TableMetadata.Builder updatesForCorrectBase(
 
         } else {
             // add new schema if needed
-            Preconditions.checkArgument(
-                    newMetadata.currentSchemaId() >= schemaId,
-                    "the new metadata has correct base, but the schemaId(%s) in iceberg table "
-                            + "is greater than currentSchemaId(%s) in new metadata.",
-                    schemaId,
-                    newMetadata.currentSchemaId());
-            if (newMetadata.currentSchemaId() != schemaId) {
+            if (newMetadata.currentSchemaId() > schemaId) {
                 addAndSetCurrentSchema(
                         newMetadata.schemas().stream()
                                 .filter(schema -> schema.schemaId() > schemaId)
                                 .collect(Collectors.toList()),
                         newMetadata.currentSchemaId(),
                         updateBuilder);
+            } else if (newMetadata.currentSchemaId() < schemaId) {
+                // a rollback moved the current schema back; only the pointer moves
+                updateBuilder.setCurrentSchema(newMetadata.currentSchemaId());
             }
 
             // add snapshot
@@ -468,6 +484,21 @@ private static boolean checkBase(
             return false;
         }
 
+        // the same numeric id can belong to a rolled-back timeline; extending from it would
+        // keep the abandoned history in the catalog
+        IcebergSnapshot baseCurrent = baseIcebergMetadata.currentSnapshot();
+        if (baseCurrent != null
+                && currentMetadata.currentSnapshot().snapshotId() == baseCurrent.snapshotId()) {
+            String catalogIdentity =
+                    currentMetadata.currentSnapshot().summary().get(PAIMON_COMMIT_IDENTITY);
+            String baseIdentity = baseCurrent.summary().get(PAIMON_COMMIT_IDENTITY);
+            if (catalogIdentity != null
+                    && baseIdentity != null
+                    && !catalogIdentity.equals(baseIdentity)) {
+                return false;
+            }
+        }
+
         // if the iceberg table is existed, check whether the current metadata of the table is the
         // base of the new table metadata, we use current snapshot id to check.
         // Note: callers must ensure currentMetadata.currentSnapshot() is non-null before calling
@@ -603,7 +634,15 @@ private IcebergMetadata adjustMetadataForRest(IcebergMetadata newIcebergMetadata
                                             snapshot.sequenceNumber(),
                                             snapshot.snapshotId(),
                                             snapshot.parentSnapshotId(),
-                                            snapshot.timestampMs(),
+                                            // a slow rebuild must not trip Iceberg's
+                                            // one-minute update-timestamp window
+                                            snapshot.snapshotId()
+                                                            == newIcebergMetadata
+                                                                    .currentSnapshotId()
+                                                    ? Math.max(
+                                                            snapshot.timestampMs(),
+                                                            System.currentTimeMillis() - 59_000L)
+                                                    : snapshot.timestampMs(),
                                             snapshot.summary(),
                                             snapshot.manifestList(),
                                             remappedSchemaId,