[core] Persist row ID reassignment plans and mark snapshots

docs/docs/multimodal-table/data-evolution-maintenance.mdx

@@ -187,6 +187,11 @@ are not permanent application identifiers. The `reassign_row_id` procedure is
 documented in the [Spark](../spark/procedures) and
 [Flink](../flink/procedures) procedure references.
 
+Each reassignment snapshot records a `row-id-reassign.plan` property referencing a
+versioned plan file in the table's `manifest/` directory. The file stores the row-ID
+mappings applied to each affected partition. It is retained while its snapshot or a
+tag referencing that snapshot is retained.
+
 ## File Sizing
 
 `target-file-size` controls normal-file sizing. `blob.target-file-size` and

paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassigner.java

@@ -427,6 +427,14 @@ private CommitAssignmentResult commitAssignment(
         Pair<String, Long> deltaManifestList = manifestList.write(Collections.emptyList());
         RewrittenIndexManifest rewrittenIndexManifest = rewriteIndexManifest(assignment);
 
+        Map<String, String> properties =
+                SerializationAssignment.writeProperties(
+                        table,
+                        assignment.snapshot,
+                        assignment.rowIdMappings,
+                        assignment.firstAssignedRowId,
+                        assignment.nextRowId);
+
         boolean success;
         try (FileStoreCommitImpl commit =
                 (FileStoreCommitImpl) table.store().newCommit(commitUser, table)) {
@@ -438,7 +446,11 @@ private CommitAssignmentResult commitAssignment(
                             baseManifestList,
                             deltaManifestList,
                             rewrittenIndexManifest.indexManifest,
-                            assignment.nextRowId);
+                            assignment.nextRowId,
+                            properties);
+        }
+        if (!success) {
+            SerializationAssignment.deletePlan(table, properties);
         }
         return new CommitAssignmentResult(
                 success, rewrittenDataManifests.fileCount, rewrittenIndexManifest.indexFileCount);

paimon-core/src/main/java/org/apache/paimon/append/dataevolution/RowRangeMappingIndex.java

@@ -18,15 +18,19 @@
 
 package org.apache.paimon.append.dataevolution;
 
+import org.apache.paimon.io.DataInputView;
+import org.apache.paimon.io.DataOutputView;
 import org.apache.paimon.utils.Range;
 
+import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Comparator;
 import java.util.List;
 import java.util.Optional;
 
 import static org.apache.paimon.utils.Preconditions.checkArgument;
+import static org.apache.paimon.utils.SerializationUtils.readCount;
 
 /** Index for row-range mappings. */
 final class RowRangeMappingIndex {
@@ -166,6 +170,28 @@ boolean overlaps(Range oldRange) {
         return index < oldStarts.length && oldStarts[index] <= oldRange.to;
     }
 
+    void serialize(DataOutputView out) throws IOException {
+        out.writeInt(oldStarts.length);
+        for (int i = 0; i < oldStarts.length; i++) {
+            out.writeLong(oldStarts[i]);
+            out.writeLong(oldEnds[i]);
+            out.writeLong(Math.addExact(newStarts[i], newStartOffset));
+        }
+    }
+
+    static RowRangeMappingIndex deserialize(DataInputView in) throws IOException {
+        int size = readCount(in, "row-id mappings");
+        long[] oldStarts = new long[size];
+        long[] oldEnds = new long[size];
+        long[] newStarts = new long[size];
+        for (int i = 0; i < size; i++) {
+            oldStarts[i] = in.readLong();
+            oldEnds[i] = in.readLong();
+            newStarts[i] = in.readLong();
+        }
+        return createFromOwnedArrays(oldStarts, oldEnds, newStarts);
+    }
+
     private static int lowerBound(long[] sorted, long target) {
         int left = 0;
         int right = sorted.length;

paimon-core/src/main/java/org/apache/paimon/append/dataevolution/SerializationAssignment.java

@@ -0,0 +1,209 @@
+/*
+ * 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.append.dataevolution;
+
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.io.DataInputViewStreamWrapper;
+import org.apache.paimon.io.DataOutputViewStreamWrapper;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.FileStorePathFactory;
+import org.apache.paimon.utils.Range;
+
+import javax.annotation.Nullable;
+
+import java.io.BufferedInputStream;
+import java.io.BufferedOutputStream;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Optional;
+import java.util.UUID;
+import java.util.zip.CRC32;
+import java.util.zip.CheckedInputStream;
+import java.util.zip.CheckedOutputStream;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+import static org.apache.paimon.utils.SerializationUtils.deserializeBinaryRow;
+import static org.apache.paimon.utils.SerializationUtils.readCount;
+import static org.apache.paimon.utils.SerializationUtils.serializeBinaryRow;
+
+/** Persisted row-id mappings and allocation bounds copied from a reassignment attempt. */
+public final class SerializationAssignment {
+
+    /** A snapshot-local marker and reference to the plan in the manifest directory. */
+    public static final String PLAN_FILE_PROPERTY = "row-id-reassign.plan";
+
+    private static final int VERSION = 1;
+    private static final String FILE_PREFIX = "row-id-reassign-plan-";
+
+    private final Map<BinaryRow, RowRangeMappingIndex> rowIdMappings;
+    private final long firstAssignedRowId;
+    private final long nextRowId;
+
+    private SerializationAssignment(
+            Map<BinaryRow, RowRangeMappingIndex> rowIdMappings,
+            long firstAssignedRowId,
+            long nextRowId) {
+        checkArgument(!rowIdMappings.isEmpty(), "Reassignment mappings must not be empty.");
+        checkArgument(
+                firstAssignedRowId >= 0 && nextRowId > firstAssignedRowId,
+                "Invalid assigned row-id range [%s, %s).",
+                firstAssignedRowId,
+                nextRowId);
+        this.rowIdMappings = Collections.unmodifiableMap(new LinkedHashMap<>(rowIdMappings));
+        this.firstAssignedRowId = firstAssignedRowId;
+        this.nextRowId = nextRowId;
+    }
+
+    public long firstAssignedRowId() {
+        return firstAssignedRowId;
+    }
+
+    public long nextRowId() {
+        return nextRowId;
+    }
+
+    /** Returns a mapping only when the entire range maps to a contiguous range. */
+    public Optional<Range> map(BinaryRow partition, Range range) {
+        RowRangeMappingIndex mapping = rowIdMappings.get(partition);
+        return mapping == null ? Optional.empty() : mapping.map(range);
+    }
+
+    public boolean overlaps(BinaryRow partition, Range range) {
+        RowRangeMappingIndex mapping = rowIdMappings.get(partition);
+        return mapping != null && mapping.overlaps(range);
+    }
+
+    @Nullable
+    public static String planFile(Snapshot snapshot) {
+        return snapshot.properties() == null ? null : snapshot.properties().get(PLAN_FILE_PROPERTY);
+    }
+
+    /** Reassignment describes a transition and must not be inherited by another commit. */
+    @Nullable
+    public static Map<String, String> withoutPlan(@Nullable Map<String, String> properties) {
+        if (properties == null || !properties.containsKey(PLAN_FILE_PROPERTY)) {
+            return properties;
+        }
+        Map<String, String> result = new HashMap<>(properties);
+        result.remove(PLAN_FILE_PROPERTY);
+        return result.isEmpty() ? null : result;
+    }
+
+    /** Persists the assignment and adds its reference to this commit's snapshot properties. */
+    static Map<String, String> writeProperties(
+            FileStoreTable table,
+            Snapshot snapshot,
+            Map<BinaryRow, RowRangeMappingIndex> rowIdMappings,
+            long firstAssignedRowId,
+            long nextRowId) {
+        String planFile;
+        try {
+            planFile =
+                    new SerializationAssignment(rowIdMappings, firstAssignedRowId, nextRowId)
+                            .write(table.fileIO(), table.store().pathFactory());
+        } catch (IOException e) {
+            throw new UncheckedIOException("Failed to persist row-id reassignment plan.", e);
+        }
+        Map<String, String> properties =
+                snapshot.properties() == null
+                        ? new HashMap<>()
+                        : new HashMap<>(snapshot.properties());
+        properties.put(PLAN_FILE_PROPERTY, planFile);
+        return properties;
+    }
+
+    /** Only call after a definitively rejected commit, never when its outcome is uncertain. */
+    static void deletePlan(FileStoreTable table, Map<String, String> properties) {
+        table.fileIO()
+                .deleteQuietly(
+                        table.store()
+                                .pathFactory()
+                                .toManifestFilePath(properties.get(PLAN_FILE_PROPERTY)));
+    }
+
+    /** Streams the effective mappings without materializing another copy of the plan. */
+    private String write(FileIO fileIO, FileStorePathFactory pathFactory) throws IOException {
+        String fileName = FILE_PREFIX + UUID.randomUUID();
+        Path path = pathFactory.toManifestFilePath(fileName);
+        try (DataOutputViewStreamWrapper out =
+                new DataOutputViewStreamWrapper(
+                        new BufferedOutputStream(fileIO.newOutputStream(path, false)))) {
+            CRC32 checksum = new CRC32();
+            DataOutputViewStreamWrapper payload =
+                    new DataOutputViewStreamWrapper(new CheckedOutputStream(out, checksum));
+            payload.writeInt(VERSION);
+            payload.writeLong(firstAssignedRowId);
+            payload.writeLong(nextRowId);
+            payload.writeInt(rowIdMappings.size());
+            for (Map.Entry<BinaryRow, RowRangeMappingIndex> entry : rowIdMappings.entrySet()) {
+                serializeBinaryRow(entry.getKey(), payload);
+                entry.getValue().serialize(payload);
+            }
+            payload.flush();
+            out.writeLong(checksum.getValue());
+        } catch (IOException | RuntimeException e) {
+            fileIO.deleteQuietly(path);
+            throw e;
+        }
+        return fileName;
+    }
+
+    public static SerializationAssignment read(
+            FileIO fileIO, FileStorePathFactory pathFactory, String fileName) throws IOException {
+        try (DataInputViewStreamWrapper in =
+                new DataInputViewStreamWrapper(
+                        new BufferedInputStream(
+                                fileIO.newInputStream(pathFactory.toManifestFilePath(fileName))))) {
+            CRC32 checksum = new CRC32();
+            DataInputViewStreamWrapper payload =
+                    new DataInputViewStreamWrapper(new CheckedInputStream(in, checksum));
+            int version = payload.readInt();
+            if (version != VERSION) {
+                throw new IOException("Unsupported row-id reassignment plan version: " + version);
+            }
+            long firstAssignedRowId = payload.readLong();
+            long nextRowId = payload.readLong();
+            int partitions = readCount(payload, "reassignment partitions");
+            Map<BinaryRow, RowRangeMappingIndex> mappings = new LinkedHashMap<>();
+            for (int i = 0; i < partitions; i++) {
+                BinaryRow partition = deserializeBinaryRow(payload);
+                if (mappings.put(partition, RowRangeMappingIndex.deserialize(payload)) != null) {
+                    throw new IOException("Duplicate partition in row-id reassignment plan.");
+                }
+            }
+            long actualChecksum = checksum.getValue();
+            if (in.readLong() != actualChecksum) {
+                throw new IOException("Row-id reassignment plan checksum mismatch.");
+            }
+            if (in.read() != -1) {
+                throw new IOException("Unexpected trailing bytes in row-id reassignment plan.");
+            }
+            return new SerializationAssignment(mappings, firstAssignedRowId, nextRowId);
+        } catch (IllegalArgumentException e) {
+            throw new IOException("Invalid row-id reassignment plan " + fileName, e);
+        }
+    }
+}

paimon-core/src/main/java/org/apache/paimon/operation/FileDeletionBase.java

@@ -19,6 +19,7 @@
 package org.apache.paimon.operation;
 
 import org.apache.paimon.Snapshot;
+import org.apache.paimon.append.dataevolution.SerializationAssignment;
 import org.apache.paimon.data.BinaryRow;
 import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.fs.Path;
@@ -381,6 +382,11 @@ protected List<Runnable> planManifestsCleaner(
         collectUnusedIndexManifests(snapshot, skippingSet, indexFiles, indexManifests);
         collectUnusedStatisticsManifests(snapshot, skippingSet, statistics);
 
+        String reassignPlan = SerializationAssignment.planFile(snapshot);
+        if (reassignPlan != null && skippingSet.add(reassignPlan)) {
+            manifests.add(reassignPlan);
+        }
+
         List<Runnable> tasks = new ArrayList<>();
         for (String manifest : manifests) {
             tasks.add(() -> manifestFile.delete(manifest));
@@ -617,6 +623,11 @@ private Set<String> manifestSkippingSet(Snapshot skippingSnapshot) {
                     .forEach(skippingSet::add);
         }
 
+        String reassignPlan = SerializationAssignment.planFile(skippingSnapshot);
+        if (reassignPlan != null) {
+            skippingSet.add(reassignPlan);
+        }
+
         // statistics
         if (skippingSnapshot.statistics() != null) {
             skippingSet.add(skippingSnapshot.statistics());

paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java

@@ -22,6 +22,7 @@
 import org.apache.paimon.Snapshot;
 import org.apache.paimon.Snapshot.CommitKind;
 import org.apache.paimon.annotation.VisibleForTesting;
+import org.apache.paimon.append.dataevolution.SerializationAssignment;
 import org.apache.paimon.catalog.SnapshotCommit;
 import org.apache.paimon.data.BinaryRow;
 import org.apache.paimon.data.InternalRow;
@@ -1394,6 +1395,24 @@ public boolean replaceManifestList(
             Pair<String, Long> deltaManifestList,
             @Nullable String indexManifest,
             @Nullable Long nextRowId) {
+        return replaceManifestList(
+                latest,
+                totalRecordCount,
+                baseManifestList,
+                deltaManifestList,
+                indexManifest,
+                nextRowId,
+                SerializationAssignment.withoutPlan(latest.properties()));
+    }
+
+    public boolean replaceManifestList(
+            Snapshot latest,
+            long totalRecordCount,
+            Pair<String, Long> baseManifestList,
+            Pair<String, Long> deltaManifestList,
+            @Nullable String indexManifest,
+            @Nullable Long nextRowId,
+            @Nullable Map<String, String> properties) {
         Snapshot newSnapshot =
                 new Snapshot(
                         latest.id() + 1,
@@ -1416,7 +1435,7 @@ public boolean replaceManifestList(
                         latest.watermark(),
                         latest.statistics(),
                         // if empty properties, just set to null
-                        latest.properties(),
+                        properties == null || properties.isEmpty() ? null : properties,
                         nextRowId,
                         null);
 
@@ -1502,7 +1521,7 @@ public boolean rollbackToAsLatest(Snapshot targetSnapshot) {
                         null,
                         targetSnapshot.watermark(),
                         targetSnapshot.statistics(),
-                        targetSnapshot.properties(),
+                        SerializationAssignment.withoutPlan(targetSnapshot.properties()),
                         nextRowId,
                         null);
 
@@ -1652,7 +1671,7 @@ private boolean compactManifestOnce() {
                         null,
                         latestSnapshot.watermark(),
                         latestSnapshot.statistics(),
-                        latestSnapshot.properties(),
+                        SerializationAssignment.withoutPlan(latestSnapshot.properties()),
                         latestSnapshot.nextRowId(),
                         null);