[core][spark]: support rewrite_manifest procedure

paimon-core/src/main/java/org/apache/paimon/manifest/ManifestEntrySortKey.java

@@ -0,0 +1,143 @@
+/*
+ * 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.manifest;
+
+import org.apache.paimon.codegen.CodeGenUtils;
+import org.apache.paimon.codegen.RecordComparator;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.utils.SerializationUtils;
+
+import java.io.Serializable;
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * A lightweight, serializable sort key for {@link ManifestEntry}. Entries are ordered by {@code
+ * partition -> bucket -> level -> fileName}, which co-locates the ADD and DELETE of the same file
+ * (they share the same key) so they always land in the same Spark partition after {@code sortByKey}
+ * and can be cancelled during the manifest rewrite. The {@code partition -> bucket -> level} prefix
+ * also keeps {@link ManifestFileMeta} statistics (partitionStats / minBucket / maxBucket / minLevel
+ * / maxLevel) compact for scan pruning.
+ *
+ * <p>The partition is kept as serialized bytes and deserialized lazily on the first comparison; the
+ * generated {@link RecordComparator} is also created lazily per executor to avoid being serialized
+ * across the shuffle. This class is intentionally small so that it can be used as the key of a
+ * Spark {@code sortByKey} shuffle without moving the whole {@link ManifestEntry} payload.
+ *
+ * <p><b>Kryo compatibility:</b> the partition is stored as a plain {@code byte[]} rather than a
+ * {@link BinaryRow} because this key travels through Spark's shuffle, where the serializer is
+ * {@code KryoSerializer} (Paimon's Spark test base randomly picks Kryo). {@code BinaryRow} only
+ * implements Java serialization (its {@code BinarySection.writeObject/readObject} callbacks), and
+ * Kryo does not invoke those callbacks while also skipping the {@code transient segments} field —
+ * so a {@code BinaryRow} key ends up with {@code null} segments after a Kryo shuffle and NPEs on
+ * the first comparison. {@code byte[]} and {@code String} are transparent to both Kryo and Java
+ * serialization, and the {@link BinaryRow} / {@link RecordComparator} are rebuilt lazily on the
+ * executor after the shuffle.
+ */
+public class ManifestEntrySortKey implements Serializable, Comparable<ManifestEntrySortKey> {
+
+    private static final long serialVersionUID = 1L;
+
+    private final byte[] partitionBytes;
+    private final int bucket;
+    private final int level;
+    private final String fileName;
+
+    private final List<DataType> partitionFieldTypes;
+
+    private transient BinaryRow partition;
+    private transient RecordComparator partitionComparator;
+
+    public ManifestEntrySortKey(
+            BinaryRow partition,
+            int bucket,
+            int level,
+            String fileName,
+            List<DataType> partitionFieldTypes) {
+        this.partitionBytes = SerializationUtils.serializeBinaryRow(partition);
+        this.bucket = bucket;
+        this.level = level;
+        this.fileName = fileName;
+        this.partitionFieldTypes = partitionFieldTypes;
+    }
+
+    @Override
+    public int compareTo(ManifestEntrySortKey other) {
+        // 1. partition
+        int cmp = partitionComparator().compare(partition(), other.partition());
+        if (cmp != 0) {
+            return cmp;
+        }
+        // 2. bucket
+        cmp = Integer.compare(bucket, other.bucket);
+        if (cmp != 0) {
+            return cmp;
+        }
+        // 3. level
+        cmp = Integer.compare(level, other.level);
+        if (cmp != 0) {
+            return cmp;
+        }
+        // 4. fileName — co-locates ADD and DELETE of the same file (same key)
+        return fileName.compareTo(other.fileName);
+    }
+
+    private BinaryRow partition() {
+        if (partition == null) {
+            partition = SerializationUtils.deserializeBinaryRow(partitionBytes);
+        }
+        return partition;
+    }
+
+    private RecordComparator partitionComparator() {
+        if (partitionComparator == null) {
+            partitionComparator = CodeGenUtils.newRecordComparator(partitionFieldTypes);
+        }
+        return partitionComparator;
+    }
+
+    public int bucket() {
+        return bucket;
+    }
+
+    public int level() {
+        return level;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        ManifestEntrySortKey that = (ManifestEntrySortKey) o;
+        return bucket == that.bucket
+                && level == that.level
+                && Objects.deepEquals(partitionBytes, that.partitionBytes)
+                && Objects.equals(fileName, that.fileName);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(bucket, level, fileName, java.util.Arrays.hashCode(partitionBytes));
+    }
+}

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

@@ -23,6 +23,7 @@
 import org.apache.paimon.disk.IOManager;
 import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.manifest.ManifestCommittable;
+import org.apache.paimon.manifest.ManifestFileMeta;
 import org.apache.paimon.operation.metrics.CommitMetrics;
 import org.apache.paimon.stats.Statistics;
 import org.apache.paimon.table.sink.CommitMessage;
@@ -83,6 +84,16 @@ FileStoreCommit rowIdCheckConflictForMaterializeDvCompaction(
     /** Compact the manifest entries only. */
     void compactManifest();
 
+    /**
+     * Replace the manifest entries with the given rewritten manifests. The {@code removedManifests}
+     * are the manifests the caller read and sorted (used for conflict detection); the {@code
+     * addedManifests} are the sorted rewrite result produced by the caller. The commit reuses the
+     * optimistic concurrency mode of {@link #compactManifest()}: on conflict, new delta manifests
+     * added by other commits are appended to the tail of {@code addedManifests}.
+     */
+    void replaceManifest(
+            List<ManifestFileMeta> removedManifests, List<ManifestFileMeta> addedManifests);
+
     /** Roll back to the target snapshot and materialize it as the latest snapshot. */
     boolean rollbackToAsLatest(Snapshot targetSnapshot);
 

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

@@ -1675,6 +1675,98 @@ public void compactManifest() {
         }
     }
 
+    @Override
+    public void replaceManifest(
+            List<ManifestFileMeta> removedManifests, List<ManifestFileMeta> addedManifests) {
+        int retryCount = 0;
+        long startMillis = System.currentTimeMillis();
+        Set<String> removedPathSet =
+                removedManifests.stream()
+                        .map(ManifestFileMeta::fileName)
+                        .collect(Collectors.toSet());
+        while (true) {
+            Snapshot latestSnapshot = snapshotManager.latestSnapshot();
+            if (latestSnapshot == null) {
+                throw new RuntimeException("Cannot replace manifests: the table has no snapshot.");
+            }
+
+            List<ManifestFileMeta> currentBase = manifestList.readDataManifests(latestSnapshot);
+            Set<String> currentPathSet =
+                    currentBase.stream()
+                            .map(ManifestFileMeta::fileName)
+                            .collect(Collectors.toSet());
+
+            if (!currentPathSet.containsAll(removedPathSet)) {
+                cleanUpRewrittenManifests(addedManifests);
+                throw new RuntimeException(
+                        "Manifest conflict: the current snapshot does not contain all the "
+                                + "manifests to replace. Another manifest rewrite may have "
+                                + "happened concurrently; please retry.");
+            }
+
+            List<ManifestFileMeta> manifestsKept =
+                    currentBase.stream()
+                            .filter(m -> !removedPathSet.contains(m.fileName()))
+                            .collect(Collectors.toList());
+
+            List<ManifestFileMeta> manifestsToCommit = new ArrayList<>(manifestsKept);
+            manifestsToCommit.addAll(addedManifests);
+
+            Pair<String, Long> baseManifestList = manifestList.write(manifestsToCommit);
+            Pair<String, Long> deltaManifestList = manifestList.write(emptyList());
+
+            Snapshot newSnapshot =
+                    new Snapshot(
+                            latestSnapshot.id() + 1,
+                            latestSnapshot.schemaId(),
+                            baseManifestList.getLeft(),
+                            baseManifestList.getRight(),
+                            deltaManifestList.getLeft(),
+                            deltaManifestList.getRight(),
+                            null,
+                            null,
+                            latestSnapshot.indexManifest(),
+                            commitUser,
+                            Long.MAX_VALUE,
+                            CommitKind.COMPACT,
+                            System.currentTimeMillis(),
+                            latestSnapshot.totalRecordCount(),
+                            0L,
+                            null,
+                            latestSnapshot.watermark(),
+                            latestSnapshot.statistics(),
+                            latestSnapshot.properties(),
+                            latestSnapshot.nextRowId(),
+                            null);
+
+            if (commitSnapshotImpl(latestSnapshot, newSnapshot, emptyList())) {
+                return;
+            }
+
+            manifestList.delete(deltaManifestList.getLeft());
+            manifestList.delete(baseManifestList.getLeft());
+
+            if (System.currentTimeMillis() - startMillis > options.commitTimeout()
+                    || retryCount >= options.commitMaxRetries()) {
+                cleanUpRewrittenManifests(addedManifests);
+                throw new RuntimeException(
+                        String.format(
+                                "Commit failed after %s millis with %s retries, there maybe exist commit conflicts between multiple jobs.",
+                                options.commitTimeout(), retryCount));
+            }
+
+            retryWaiter.retryWait(retryCount);
+            retryCount++;
+        }
+    }
+
+    /** Delete the rewritten manifest files produced by this rewrite. */
+    private void cleanUpRewrittenManifests(List<ManifestFileMeta> addedManifests) {
+        for (ManifestFileMeta manifest : addedManifests) {
+            manifestFile.delete(manifest.fileName());
+        }
+    }
+
     private boolean compactManifestOnce() {
         Snapshot latestSnapshot = snapshotManager.latestSnapshot();
 

paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java

@@ -27,6 +27,7 @@
 import org.apache.paimon.fs.FileStatus;
 import org.apache.paimon.fs.Path;
 import org.apache.paimon.fs.TwoPhaseOutputStream;
+import org.apache.paimon.manifest.ManifestFileMeta;
 import org.apache.paimon.metrics.MetricRegistry;
 import org.apache.paimon.options.CatalogOptions;
 import org.apache.paimon.options.Options;
@@ -318,6 +319,12 @@ public void compactManifests() {
         throw new UnsupportedOperationException();
     }
 
+    @Override
+    public void replaceManifests(
+            List<ManifestFileMeta> removedManifests, List<ManifestFileMeta> addedManifests) {
+        throw new UnsupportedOperationException();
+    }
+
     @Override
     public TableCommit withMetricRegistry(MetricRegistry registry) {
         throw new UnsupportedOperationException();

paimon-core/src/main/java/org/apache/paimon/table/sink/BatchTableCommit.java

@@ -21,6 +21,7 @@
 import org.apache.paimon.Snapshot;
 import org.apache.paimon.Snapshot.CommitKind;
 import org.apache.paimon.annotation.Public;
+import org.apache.paimon.manifest.ManifestFileMeta;
 import org.apache.paimon.stats.Statistics;
 
 import java.util.List;
@@ -73,6 +74,14 @@ public interface BatchTableCommit extends TableCommit {
     /** Compact the manifest entries. Generates a snapshot with {@link CommitKind#COMPACT}. */
     void compactManifests();
 
+    /**
+     * Replace the manifest entries with the given rewritten manifests. The {@code removedManifests}
+     * are the manifests the caller read and sorted; the {@code addedManifests} are the sorted
+     * rewrite result. Generates a snapshot with {@link CommitKind#COMPACT}.
+     */
+    void replaceManifests(
+            List<ManifestFileMeta> removedManifests, List<ManifestFileMeta> addedManifests);
+
     /** Set the logical operation type (e.g. WRITE, DELETE, MERGE) recorded in the snapshot. */
     default BatchTableCommit withOperation(Snapshot.Operation operation) {
         return this;

paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java

@@ -28,6 +28,7 @@
 import org.apache.paimon.io.DataFileMeta;
 import org.apache.paimon.io.DataFilePathFactory;
 import org.apache.paimon.manifest.ManifestCommittable;
+import org.apache.paimon.manifest.ManifestFileMeta;
 import org.apache.paimon.metrics.MetricRegistry;
 import org.apache.paimon.operation.FileStoreCommit;
 import org.apache.paimon.operation.PartitionExpire;
@@ -235,6 +236,12 @@ public void compactManifests() {
         commit.compactManifest();
     }
 
+    @Override
+    public void replaceManifests(
+            List<ManifestFileMeta> removedManifests, List<ManifestFileMeta> addedManifests) {
+        commit.replaceManifest(removedManifests, addedManifests);
+    }
+
     public boolean rollbackToAsLatest(Tag targetTag) {
         checkCommitted();
         boolean success = commit.rollbackToAsLatest(targetTag.trimToSnapshot());