[core] Introduce SupportsBranches to put branch management to Catalog

Author: JingsongLi

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

@@ -59,7 +59,7 @@
 import static org.apache.paimon.catalog.CatalogUtils.validateAutoCreateClose;
 import static org.apache.paimon.options.CatalogOptions.LOCK_ENABLED;
 import static org.apache.paimon.options.CatalogOptions.LOCK_TYPE;
-import static org.apache.paimon.utils.BranchManager.DEFAULT_MAIN_BRANCH;
+import static org.apache.paimon.utils.FileSystemBranchManager.DEFAULT_MAIN_BRANCH;
 import static org.apache.paimon.utils.Preconditions.checkArgument;
 
 /** Common implementation of {@link Catalog}. */
@@ -366,16 +366,14 @@ protected abstract void alterTableImpl(Identifier identifier, List<SchemaChange>
 
     @Override
     public Table getTable(Identifier identifier) throws TableNotExistException {
-        SnapshotCommit.Factory commitFactory =
-                new RenamingSnapshotCommit.Factory(
-                        lockFactory().orElse(null), lockContext().orElse(null));
         return CatalogUtils.loadTable(
                 this,
                 identifier,
                 p -> fileIO(),
                 this::fileIO,
                 this::loadTableMetadata,
-                commitFactory);
+                lockFactory().orElse(null),
+                lockContext().orElse(null));
     }
 
     /**

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

@@ -0,0 +1,67 @@
+/*
+ * 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.catalog;
+
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.partition.Partition;
+import org.apache.paimon.utils.SnapshotManager;
+
+import java.util.List;
+
+/** A {@link SnapshotCommit} using {@link Catalog} to commit. */
+public class CatalogSnapshotCommit implements SnapshotCommit {
+
+    private final SupportsSnapshots supportsSnapshots;
+    private final Identifier identifier;
+
+    public CatalogSnapshotCommit(SupportsSnapshots supportsSnapshots, Identifier identifier) {
+        this.supportsSnapshots = supportsSnapshots;
+        this.identifier = identifier;
+    }
+
+    @Override
+    public boolean commit(Snapshot snapshot, String branch, List<Partition> statistics)
+            throws Exception {
+        Identifier newIdentifier =
+                new Identifier(identifier.getDatabaseName(), identifier.getTableName(), branch);
+        return supportsSnapshots.commitSnapshot(newIdentifier, snapshot, statistics);
+    }
+
+    @Override
+    public void close() throws Exception {
+        supportsSnapshots.close();
+    }
+
+    /** Factory to create {@link CatalogSnapshotCommit}. */
+    public static class Factory implements SnapshotCommit.Factory {
+
+        private static final long serialVersionUID = 1L;
+
+        private final CatalogLoader catalogLoader;
+
+        public Factory(CatalogLoader catalogLoader) {
+            this.catalogLoader = catalogLoader;
+        }
+
+        @Override
+        public SnapshotCommit create(Identifier identifier, SnapshotManager snapshotManager) {
+            return new CatalogSnapshotCommit((SupportsSnapshots) catalogLoader.load(), identifier);
+        }
+    }
+}

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

@@ -39,6 +39,8 @@
 import org.apache.paimon.utils.InternalRowPartitionComputer;
 import org.apache.paimon.utils.Preconditions;
 
+import javax.annotation.Nullable;
+
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
@@ -169,7 +171,8 @@ public static Table loadTable(
             Function<Path, FileIO> internalFileIO,
             Function<Path, FileIO> externalFileIO,
             TableMetadata.Loader metadataLoader,
-            SnapshotCommit.Factory commitFactory)
+            @Nullable CatalogLockFactory lockFactory,
+            @Nullable CatalogLockContext lockContext)
             throws Catalog.TableNotExistException {
         if (SYSTEM_DATABASE_NAME.equals(identifier.getDatabaseName())) {
             return CatalogUtils.createGlobalSystemTable(identifier.getTableName(), catalog);
@@ -190,8 +193,10 @@ public static Table loadTable(
                         identifier,
                         metadata.uuid(),
                         catalog.catalogLoader(),
-                        commitFactory,
-                        catalog instanceof SupportsSnapshots);
+                        lockFactory,
+                        lockContext,
+                        catalog instanceof SupportsSnapshots,
+                        catalog instanceof SupportsBranches);
         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/Identifier.java

@@ -22,7 +22,7 @@
 import org.apache.paimon.types.DataField;
 import org.apache.paimon.types.DataTypes;
 import org.apache.paimon.types.RowType;
-import org.apache.paimon.utils.BranchManager;
+import org.apache.paimon.utils.FileSystemBranchManager;
 import org.apache.paimon.utils.Preconditions;
 import org.apache.paimon.utils.StringUtils;
 
@@ -139,7 +139,7 @@ public String getTableName() {
     @JsonIgnore
     public String getBranchNameOrDefault() {
         String branch = getBranchName();
-        return branch == null ? BranchManager.DEFAULT_MAIN_BRANCH : branch;
+        return branch == null ? FileSystemBranchManager.DEFAULT_MAIN_BRANCH : branch;
     }
 
     @JsonIgnore

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

@@ -0,0 +1,73 @@
+/*
+ * 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.catalog;
+
+import javax.annotation.Nullable;
+
+import java.util.List;
+
+/** A {@link Catalog} supports creating and dropping table branches. */
+public interface SupportsBranches extends Catalog {
+
+    /**
+     * 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 DatabaseNotExistException if the database in identifier doesn't exist
+     */
+    void createBranch(Identifier identifier, String branch, @Nullable String fromTag)
+            throws TableNotExistException, DatabaseNotExistException;
+
+    /**
+     * Drop the branch for this table.
+     *
+     * @param identifier path of the table, cannot be system or branch name.
+     * @param branch the branch name
+     * @throws TableNotExistException if the table in identifier doesn't exist
+     * @throws DatabaseNotExistException if the database in identifier doesn't exist
+     */
+    void dropBranch(Identifier identifier, String branch)
+            throws TableNotExistException, DatabaseNotExistException;
+
+    /**
+     * 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 TableNotExistException if the table in identifier doesn't exist
+     * @throws DatabaseNotExistException if the database in identifier doesn't exist
+     */
+    void fastForward(Identifier identifier, String branch)
+            throws TableNotExistException, DatabaseNotExistException;
+
+    /**
+     * 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
+     * @throws DatabaseNotExistException if the database in identifier doesn't exist
+     */
+    List<String> listBranches(Identifier identifier)
+            throws TableNotExistException, DatabaseNotExistException;
+}

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

@@ -19,11 +19,25 @@
 package org.apache.paimon.catalog;
 
 import org.apache.paimon.Snapshot;
+import org.apache.paimon.partition.Partition;
 
+import java.util.List;
 import java.util.Optional;
 
-/** A {@link Catalog} supports loading table snapshots. */
-public interface SupportsSnapshots {
+/** A {@link Catalog} supports committing and loading table snapshots. */
+public interface SupportsSnapshots extends Catalog {
+
+    /**
+     * Commit the {@link Snapshot} for table identified by the given {@link Identifier}.
+     *
+     * @param identifier Path of the table
+     * @param snapshot Snapshot to be committed
+     * @param statistics statistics information of this change
+     * @return Success or not
+     * @throws Catalog.TableNotExistException if the target does not exist
+     */
+    boolean commitSnapshot(Identifier identifier, Snapshot snapshot, List<Partition> statistics)
+            throws Catalog.TableNotExistException;
 
     /**
      * Return the snapshot of table identified by the given {@link Identifier}.

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

@@ -30,9 +30,9 @@
 import org.apache.paimon.manifest.ManifestList;
 import org.apache.paimon.schema.SchemaManager;
 import org.apache.paimon.table.FileStoreTable;
-import org.apache.paimon.utils.BranchManager;
 import org.apache.paimon.utils.DateTimeUtils;
 import org.apache.paimon.utils.FileStorePathFactory;
+import org.apache.paimon.utils.FileSystemBranchManager;
 import org.apache.paimon.utils.Pair;
 import org.apache.paimon.utils.Preconditions;
 import org.apache.paimon.utils.SnapshotManager;
@@ -103,7 +103,7 @@ public OrphanFilesClean(FileStoreTable table, long olderThanMillis, boolean dryR
 
     protected List<String> validBranches() {
         List<String> branches = table.branchManager().branches();
-        branches.add(BranchManager.DEFAULT_MAIN_BRANCH);
+        branches.add(FileSystemBranchManager.DEFAULT_MAIN_BRANCH);
 
         List<String> abnormalBranches = new ArrayList<>();
         for (String branch : branches) {

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

@@ -283,7 +283,8 @@ public Table getTable(Identifier identifier) throws TableNotExistException {
                 path -> fileIOForData(path, identifier),
                 this::fileIOFromOptions,
                 this::loadTableMetadata,
-                new RESTSnapshotCommitFactory(catalogLoader()));
+                null,
+                null);
     }
 
     private FileIO fileIOForData(Path path, Identifier identifier) {
@@ -340,15 +341,26 @@ public Optional<Snapshot> loadSnapshot(Identifier identifier) throws TableNotExi
         return Optional.of(response.getSnapshot());
     }
 
+    @Override
     public boolean commitSnapshot(
-            Identifier identifier, Snapshot snapshot, List<Partition> statistics) {
+            Identifier identifier, Snapshot snapshot, List<Partition> statistics)
+            throws TableNotExistException {
         CommitTableRequest request = new CommitTableRequest(identifier, snapshot, statistics);
-        CommitTableResponse response =
-                client.post(
-                        resourcePaths.commitTable(identifier.getDatabaseName()),
-                        request,
-                        CommitTableResponse.class,
-                        restAuthFunction);
+        CommitTableResponse response;
+
+        try {
+            response =
+                    client.post(
+                            resourcePaths.commitTable(identifier.getDatabaseName()),
+                            request,
+                            CommitTableResponse.class,
+                            restAuthFunction);
+        } catch (NoSuchResourceException e) {
+            throw new TableNotExistException(identifier);
+        } catch (ForbiddenException e) {
+            throw new TableNoPermissionException(identifier, e);
+        }
+
         return response.isSuccess();
     }