[core] Fast forward branch should refresh table in CachingCatalog

Author: JingsongLi

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

@@ -37,10 +37,14 @@
 import org.apache.paimon.table.sink.BatchTableCommit;
 import org.apache.paimon.table.system.SystemTableLoader;
 import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.BranchManager;
+import org.apache.paimon.utils.FileSystemBranchManager;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import javax.annotation.Nullable;
+
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.HashMap;
@@ -456,6 +460,56 @@ public Table getTable(Identifier identifier) throws TableNotExistException {
                 lockContext().orElse(null));
     }
 
+    @Override
+    public void createBranch(Identifier identifier, String branch, @Nullable String fromTag)
+            throws TableNotExistException, BranchAlreadyExistException, TagNotExistException {
+        BranchManager branchManager = fileSystemBranchManager(identifier);
+        if (fromTag == null) {
+            branchManager.createBranch(branch);
+        } else {
+            branchManager.createBranch(branch, fromTag);
+        }
+    }
+
+    @Override
+    public void dropBranch(Identifier identifier, String branch) throws BranchNotExistException {
+        try {
+            fileSystemBranchManager(identifier).dropBranch(branch);
+        } catch (TableNotExistException e) {
+            throw new BranchNotExistException(identifier, branch, e);
+        }
+    }
+
+    @Override
+    public void fastForward(Identifier identifier, String branch) throws BranchNotExistException {
+        try {
+            fileSystemBranchManager(identifier).fastForward(branch);
+        } catch (TableNotExistException e) {
+            throw new BranchNotExistException(identifier, branch, e);
+        }
+    }
+
+    @Override
+    public List<String> listBranches(Identifier identifier) throws TableNotExistException {
+        return fileSystemBranchManager(identifier).branches();
+    }
+
+    private BranchManager fileSystemBranchManager(Identifier identifier)
+            throws TableNotExistException {
+        Table originTable = getTable(identifier);
+        if (!(originTable instanceof FileStoreTable)) {
+            throw new UnsupportedOperationException(
+                    "Unsupported table type: " + originTable.getClass());
+        }
+        FileStoreTable table = (FileStoreTable) originTable;
+        return new FileSystemBranchManager(
+                table.fileIO(),
+                table.location(),
+                table.snapshotManager(),
+                table.tagManager(),
+                table.schemaManager());
+    }
+
     /**
      * Create a {@link FormatTable} identified by the given {@link Identifier}.
      *

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

@@ -289,6 +289,12 @@ public void alterPartitions(Identifier identifier, List<Partition> partitions)
         }
     }
 
+    @Override
+    public void fastForward(Identifier identifier, String branch) throws BranchNotExistException {
+        super.fastForward(identifier, branch);
+        invalidateTable(identifier);
+    }
+
     @Override
     public void invalidateTable(Identifier identifier) {
         tableCache.invalidate(identifier);

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

@@ -463,6 +463,49 @@ default void renameView(Identifier fromView, Identifier toView, boolean ignoreIf
         throw new UnsupportedOperationException();
     }
 
+    // ======================= branch methods ===============================
+
+    /**
+     * Create a new branch for this table. By default, an empty branch will be created using the
+     * latest schema. If you provide {@code #fromTag}, a branch will be created from the tag and the
+     * data files will be inherited from it.
+     *
+     * @param identifier path of the table, cannot be system or branch name.
+     * @param branch the branch name
+     * @param fromTag from the tag
+     * @throws TableNotExistException if the table in identifier doesn't exist
+     * @throws BranchAlreadyExistException if the branch already exists
+     * @throws TagNotExistException if the tag doesn't exist
+     */
+    void createBranch(Identifier identifier, String branch, @Nullable String fromTag)
+            throws TableNotExistException, BranchAlreadyExistException, TagNotExistException;
+
+    /**
+     * Drop the branch for this table.
+     *
+     * @param identifier path of the table, cannot be system or branch name.
+     * @param branch the branch name
+     * @throws BranchNotExistException if the branch doesn't exist
+     */
+    void dropBranch(Identifier identifier, String branch) throws BranchNotExistException;
+
+    /**
+     * Fast-forward a branch to main branch.
+     *
+     * @param identifier path of the table, cannot be system or branch name.
+     * @param branch the branch name
+     * @throws BranchNotExistException if the branch doesn't exist
+     */
+    void fastForward(Identifier identifier, String branch) throws BranchNotExistException;
+
+    /**
+     * List all branches of the table.
+     *
+     * @param identifier path of the table, cannot be system or branch name.
+     * @throws TableNotExistException if the table in identifier doesn't exist
+     */
+    List<String> listBranches(Identifier identifier) throws TableNotExistException;
+
     // ======================= repair methods ===============================
 
     /**
@@ -772,4 +815,85 @@ public Identifier identifier() {
             return identifier;
         }
     }
+
+    /** Exception for trying to create a branch that already exists. */
+    class BranchAlreadyExistException extends Exception {
+
+        private static final String MSG = "Branch %s in table %s already exists.";
+
+        private final Identifier identifier;
+        private final String branch;
+
+        public BranchAlreadyExistException(Identifier identifier, String branch) {
+            this(identifier, branch, null);
+        }
+
+        public BranchAlreadyExistException(Identifier identifier, String branch, Throwable cause) {
+            super(String.format(MSG, branch, identifier.getFullName()), cause);
+            this.identifier = identifier;
+            this.branch = branch;
+        }
+
+        public Identifier identifier() {
+            return identifier;
+        }
+
+        public String branch() {
+            return branch;
+        }
+    }
+
+    /** Exception for trying to operate on a branch that doesn't exist. */
+    class BranchNotExistException extends Exception {
+
+        private static final String MSG = "Branch %s in table %s doesn't exist.";
+
+        private final Identifier identifier;
+        private final String branch;
+
+        public BranchNotExistException(Identifier identifier, String branch) {
+            this(identifier, branch, null);
+        }
+
+        public BranchNotExistException(Identifier identifier, String branch, Throwable cause) {
+            super(String.format(MSG, branch, identifier.getFullName()), cause);
+            this.identifier = identifier;
+            this.branch = branch;
+        }
+
+        public Identifier identifier() {
+            return identifier;
+        }
+
+        public String branch() {
+            return branch;
+        }
+    }
+
+    /** Exception for trying to operate on a tag that doesn't exist. */
+    class TagNotExistException extends Exception {
+
+        private static final String MSG = "Tag %s in table %s doesn't exist.";
+
+        private final Identifier identifier;
+        private final String tag;
+
+        public TagNotExistException(Identifier identifier, String tag) {
+            this(identifier, tag, null);
+        }
+
+        public TagNotExistException(Identifier identifier, String tag, Throwable cause) {
+            super(String.format(MSG, tag, identifier.getFullName()), cause);
+            this.identifier = identifier;
+            this.tag = tag;
+        }
+
+        public Identifier identifier() {
+            return identifier;
+        }
+
+        public String tag() {
+            return tag;
+        }
+    }
 }

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

@@ -195,8 +195,7 @@ public static Table loadTable(
                         catalog.catalogLoader(),
                         lockFactory,
                         lockContext,
-                        catalog instanceof SupportsSnapshots,
-                        catalog instanceof SupportsBranches);
+                        catalog instanceof SupportsSnapshots);
         Path path = new Path(schema.options().get(PATH.key()));
         FileStoreTable table =
                 FileStoreTableFactory.create(dataFileIO.apply(path), path, schema, catalogEnv);

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

@@ -25,6 +25,8 @@
 import org.apache.paimon.table.Table;
 import org.apache.paimon.view.View;
 
+import javax.annotation.Nullable;
+
 import java.util.List;
 import java.util.Map;
 
@@ -211,6 +213,27 @@ public PagedList<Partition> listPartitionsPaged(
         return wrapped.listPartitionsPaged(identifier, maxResults, pageToken);
     }
 
+    @Override
+    public void createBranch(Identifier identifier, String branch, @Nullable String fromTag)
+            throws TableNotExistException, BranchAlreadyExistException, TagNotExistException {
+        wrapped.createBranch(identifier, branch, fromTag);
+    }
+
+    @Override
+    public void dropBranch(Identifier identifier, String branch) throws BranchNotExistException {
+        wrapped.dropBranch(identifier, branch);
+    }
+
+    @Override
+    public void fastForward(Identifier identifier, String branch) throws BranchNotExistException {
+        wrapped.fastForward(identifier, branch);
+    }
+
+    @Override
+    public List<String> listBranches(Identifier identifier) throws TableNotExistException {
+        return wrapped.listBranches(identifier);
+    }
+
     @Override
     public void repairCatalog() {
         wrapped.repairCatalog();