[core] Remove branch merge from catalog API

Author: JunRuiLee

paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java

@@ -631,12 +631,6 @@ public void fastForward(Identifier identifier, String branch) throws BranchNotEx
         throw new UnsupportedOperationException();
     }
 
-    @Override
-    public void mergeBranch(Identifier identifier, String sourceBranch, String targetBranch)
-            throws BranchNotExistException {
-        throw new UnsupportedOperationException();
-    }
-
     @Override
     public List<String> listBranches(Identifier identifier) throws TableNotExistException {
         throw new UnsupportedOperationException();

paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java

@@ -920,19 +920,6 @@ void renameBranch(Identifier identifier, String fromBranch, String toBranch)
      */
     void fastForward(Identifier identifier, String branch) throws BranchNotExistException;
 
-    /**
-     * Merge source branch into target branch.
-     *
-     * @param identifier path of the table, cannot be system or branch name.
-     * @param sourceBranch the source branch name
-     * @param targetBranch the target branch name
-     * @throws BranchNotExistException if the source or target branch doesn't exist
-     * @throws UnsupportedOperationException if the catalog does not {@link
-     *     #supportsVersionManagement()}
-     */
-    void mergeBranch(Identifier identifier, String sourceBranch, String targetBranch)
-            throws BranchNotExistException;
-
     /**
      * List all branches of the table.
      *

paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java

@@ -263,12 +263,6 @@ public void fastForward(Identifier identifier, String branch) throws BranchNotEx
         wrapped.fastForward(identifier, branch);
     }
 
-    @Override
-    public void mergeBranch(Identifier identifier, String sourceBranch, String targetBranch)
-            throws BranchNotExistException {
-        wrapped.mergeBranch(identifier, sourceBranch, targetBranch);
-    }
-
     @Override
     public List<String> listBranches(Identifier identifier) throws TableNotExistException {
         return wrapped.listBranches(identifier);

paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java

@@ -783,12 +783,6 @@ public void fastForward(Identifier identifier, String branch) throws BranchNotEx
         }
     }
 
-    @Override
-    public void mergeBranch(Identifier identifier, String sourceBranch, String targetBranch)
-            throws BranchNotExistException {
-        throw new UnsupportedOperationException("Branch merge is not supported via REST catalog.");
-    }
-
     @Override
     public List<String> listBranches(Identifier identifier) throws TableNotExistException {
         try {

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

@@ -55,10 +55,10 @@
 import org.apache.paimon.table.source.snapshot.TimeTravelUtil;
 import org.apache.paimon.tag.TagAutoManager;
 import org.apache.paimon.utils.BranchManager;
+import org.apache.paimon.utils.BranchMergeHandler;
 import org.apache.paimon.utils.CatalogBranchManager;
 import org.apache.paimon.utils.ChangelogManager;
 import org.apache.paimon.utils.DVMetaCache;
-import org.apache.paimon.utils.DefaultBranchMergeHandler;
 import org.apache.paimon.utils.FileSystemBranchManager;
 import org.apache.paimon.utils.Preconditions;
 import org.apache.paimon.utils.SegmentsCache;
@@ -760,7 +760,7 @@ public BranchManager branchManager() {
                 snapshotManager(),
                 tagManager(),
                 schemaManager(),
-                new DefaultBranchMergeHandler(this::switchToBranch));
+                new BranchMergeHandler(this::switchToBranch));
     }
 
     @Override

paimon-core/src/main/java/org/apache/paimon/utils/BranchMergeHandler.java

@@ -18,18 +18,127 @@
 
 package org.apache.paimon.utils;
 
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.io.CompactIncrement;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataIncrement;
 import org.apache.paimon.manifest.FileEntry;
+import org.apache.paimon.manifest.ManifestCommittable;
 import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.manifest.ManifestFile;
+import org.apache.paimon.manifest.ManifestList;
+import org.apache.paimon.operation.FileStoreCommit;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.CommitMessageImpl;
 
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
+import java.util.UUID;
+import java.util.function.Function;
 
-/** Handler for branch merge data operations (manifest reading, committing). */
-public interface BranchMergeHandler {
+import static org.apache.paimon.utils.Preconditions.checkArgument;
 
-    /** Read all active data files from the given branch. */
-    Map<FileEntry.Identifier, ManifestEntry> readBranchFiles(String branch);
+/** Branch merge handler backed by {@link FileStoreTable}. */
+public class BranchMergeHandler {
 
-    /** Commit the given files to the target branch. */
-    void commit(String targetBranch, List<ManifestEntry> filesToMerge);
+    private final Function<String, FileStoreTable> branchTableFactory;
+
+    public BranchMergeHandler(Function<String, FileStoreTable> branchTableFactory) {
+        this.branchTableFactory = branchTableFactory;
+    }
+
+    public Map<FileEntry.Identifier, ManifestEntry> readBranchFiles(String branch) {
+        FileStoreTable branchTable = branchTableFactory.apply(branch);
+        Snapshot snapshot = branchTable.snapshotManager().latestSnapshot();
+        checkArgument(
+                snapshot != null,
+                "Cannot read branch '%s', because it does not have any snapshot.",
+                branch);
+        ManifestList manifestList = branchTable.store().manifestListFactory().create();
+        ManifestFile manifestFile = branchTable.store().manifestFileFactory().create();
+        Map<FileEntry.Identifier, ManifestEntry> files = new LinkedHashMap<>();
+        FileEntry.mergeEntries(manifestFile, manifestList.readDataManifests(snapshot), files, null);
+        return files;
+    }
+
+    public void commit(String targetBranch, List<ManifestEntry> filesToMerge) {
+        FileStoreTable branchTable = branchTableFactory.apply(targetBranch);
+        boolean rowTrackingEnabled =
+                new CoreOptions(branchTable.schema().options()).rowTrackingEnabled();
+
+        Map<MergeKey, List<DataFileMeta>> grouped = new LinkedHashMap<>();
+        for (ManifestEntry entry : filesToMerge) {
+            DataFileMeta file = prepareFileForTargetCommit(entry.file(), rowTrackingEnabled);
+            grouped.computeIfAbsent(
+                            new MergeKey(
+                                    entry.partition().copy(), entry.bucket(), entry.totalBuckets()),
+                            k -> new ArrayList<>())
+                    .add(file);
+        }
+
+        String commitUser = UUID.randomUUID().toString();
+        ManifestCommittable committable = new ManifestCommittable(0);
+        for (Map.Entry<MergeKey, List<DataFileMeta>> e : grouped.entrySet()) {
+            MergeKey key = e.getKey();
+            CommitMessageImpl message =
+                    new CommitMessageImpl(
+                            key.partition,
+                            key.bucket,
+                            key.totalBuckets,
+                            new DataIncrement(
+                                    e.getValue(), Collections.emptyList(), Collections.emptyList()),
+                            CompactIncrement.emptyIncrement());
+            committable.addFileCommittable(message);
+        }
+
+        try (FileStoreCommit commit = branchTable.store().newCommit(commitUser, branchTable)) {
+            commit.appendCommitCheckConflict(true).commit(committable, true);
+        }
+    }
+
+    private DataFileMeta prepareFileForTargetCommit(DataFileMeta file, boolean rowTrackingEnabled) {
+        if (rowTrackingEnabled && file.firstRowId() != null) {
+            // Source files already have row ids assigned in their branch. Clear them so the
+            // target branch commit path assigns fresh, non-overlapping row ids.
+            return file.newFirstRowId(null);
+        }
+        return file;
+    }
+
+    private static class MergeKey {
+        final BinaryRow partition;
+        final int bucket;
+        final int totalBuckets;
+
+        MergeKey(BinaryRow partition, int bucket, int totalBuckets) {
+            this.partition = partition;
+            this.bucket = bucket;
+            this.totalBuckets = totalBuckets;
+        }
+
+        @Override
+        public boolean equals(Object o) {
+            if (this == o) {
+                return true;
+            }
+            if (!(o instanceof MergeKey)) {
+                return false;
+            }
+            MergeKey that = (MergeKey) o;
+            return bucket == that.bucket
+                    && totalBuckets == that.totalBuckets
+                    && Objects.equals(partition, that.partition);
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(partition, bucket, totalBuckets);
+        }
+    }
 }

paimon-core/src/main/java/org/apache/paimon/utils/CatalogBranchManager.java

@@ -110,11 +110,7 @@ public void fastForward(String branchName) {
 
     @Override
     public void mergeBranch(String sourceBranch, String targetBranch) {
-        executePost(
-                catalog -> {
-                    BranchManager.mergeValidate(sourceBranch, targetBranch);
-                    catalog.mergeBranch(identifier, sourceBranch, targetBranch);
-                });
+        throw new UnsupportedOperationException("Branch merge is not supported via catalog.");
     }
 
     @Override