fix

JingsongLi · JingsongLi · commit 71a597175727 · 2025-03-17T19:56:33.000+08:00
diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/Identifier.java b/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.FileSystemBranchManager;
+import org.apache.paimon.utils.BranchManager;
 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 ? FileSystemBranchManager.DEFAULT_MAIN_BRANCH : branch;
+        return branch == null ? BranchManager.DEFAULT_MAIN_BRANCH : branch;
     }
 
     @JsonIgnore
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/OrphanFilesClean.java b/paimon-core/src/main/java/org/apache/paimon/operation/OrphanFilesClean.java
@@ -30,10 +30,10 @@
 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.ChangelogManager;
 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;
@@ -108,7 +108,7 @@ public OrphanFilesClean(FileStoreTable table, long olderThanMillis, boolean dryR
 
     protected List<String> validBranches() {
         List<String> branches = table.branchManager().branches();
-        branches.add(FileSystemBranchManager.DEFAULT_MAIN_BRANCH);
+        branches.add(BranchManager.DEFAULT_MAIN_BRANCH);
 
         List<String> abnormalBranches = new ArrayList<>();
         for (String branch : branches) {
diff --git a/paimon-core/src/main/java/org/apache/paimon/utils/BranchManager.java b/paimon-core/src/main/java/org/apache/paimon/utils/BranchManager.java
@@ -40,6 +40,10 @@ public interface BranchManager {
 
     List<String> branches();
 
+    default boolean branchExists(String branchName) {
+        return branches().contains(branchName);
+    }
+
     /** Return the path string of a branch. */
     static String branchPath(Path tablePath, String branch) {
         return isMainBranch(branch)
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/FileStoreTableTestBase.java b/paimon-core/src/test/java/org/apache/paimon/table/FileStoreTableTestBase.java
@@ -66,7 +66,7 @@
 import org.apache.paimon.types.DataTypes;
 import org.apache.paimon.types.RowKind;
 import org.apache.paimon.types.RowType;
-import org.apache.paimon.utils.FileSystemBranchManager;
+import org.apache.paimon.utils.BranchManager;
 import org.apache.paimon.utils.SnapshotManager;
 import org.apache.paimon.utils.TagManager;
 import org.apache.paimon.utils.TraceableFileIO;
@@ -1177,7 +1177,7 @@ public void testCreateBranch() throws Exception {
         table.createBranch("test-branch", "test-tag");
 
         // verify that branch file exist
-        FileSystemBranchManager branchManager = (FileSystemBranchManager) table.branchManager();
+        BranchManager branchManager = table.branchManager();
         assertThat(branchManager.branchExists("test-branch")).isTrue();
 
         // verify test-tag in test-branch is equal to snapshot 2
@@ -1261,7 +1261,7 @@ public void testDeleteBranch() throws Exception {
         table.deleteBranch("branch1");
 
         // verify that branch file not exist
-        FileSystemBranchManager branchManager = (FileSystemBranchManager) table.branchManager();
+        BranchManager branchManager = table.branchManager();
         assertThat(branchManager.branchExists("branch1")).isFalse();
 
         assertThatThrownBy(() -> table.deleteBranch("branch1"))
@@ -1927,7 +1927,7 @@ protected void generateBranch(FileStoreTable table) throws Exception {
         table.createBranch(BRANCH_NAME, "tag1");
 
         // verify that branch1 file exist
-        FileSystemBranchManager branchManager = (FileSystemBranchManager) table.branchManager();
+        BranchManager branchManager = table.branchManager();
         assertThat(branchManager.branchExists(BRANCH_NAME)).isTrue();
 
         // Verify branch1 and the main branch have the same data
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/system/AggregationFieldsTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/system/AggregationFieldsTableTest.java
@@ -32,7 +32,7 @@
 import org.apache.paimon.table.FileStoreTable;
 import org.apache.paimon.table.TableTestBase;
 import org.apache.paimon.types.DataTypes;
-import org.apache.paimon.utils.FileSystemBranchManager;
+import org.apache.paimon.utils.BranchManager;
 
 import org.apache.paimon.shade.guava30.com.google.common.collect.Multimap;
 
@@ -91,7 +91,7 @@ public void testBranchAggregationFieldsRecord() throws Exception {
         FileStoreTable table = (FileStoreTable) catalog.getTable(identifier(tableName));
         table.createBranch("b1");
         // verify that branch file exist
-        FileSystemBranchManager branchManager = (FileSystemBranchManager) table.branchManager();
+        BranchManager branchManager = table.branchManager();
         assertThat(branchManager.branchExists("b1")).isTrue();
 
         SchemaManager schemaManagerBranch = schemaManager.copyWithBranch("b1");
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/system/OptionsTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/system/OptionsTableTest.java
@@ -32,7 +32,7 @@
 import org.apache.paimon.table.FileStoreTable;
 import org.apache.paimon.table.TableTestBase;
 import org.apache.paimon.types.DataTypes;
-import org.apache.paimon.utils.FileSystemBranchManager;
+import org.apache.paimon.utils.BranchManager;
 
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -86,7 +86,7 @@ public void testBranchOptionsTable() throws Exception {
         FileStoreTable table = (FileStoreTable) catalog.getTable(identifier(tableName));
         table.createBranch("b1");
         // verify that branch file exist
-        FileSystemBranchManager branchManager = (FileSystemBranchManager) table.branchManager();
+        BranchManager branchManager = table.branchManager();
         assertThat(branchManager.branchExists("b1")).isTrue();
         SchemaManager schemaManagerBranch = schemaManager.copyWithBranch("b1");
         Map<String, String> newOptions = new HashMap<>();
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FileStoreITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FileStoreITCase.java
@@ -33,8 +33,8 @@
 import org.apache.paimon.table.FileStoreTable;
 import org.apache.paimon.table.FileStoreTableFactory;
 import org.apache.paimon.utils.BlockingIterator;
+import org.apache.paimon.utils.BranchManager;
 import org.apache.paimon.utils.FailingFileIO;
-import org.apache.paimon.utils.FileSystemBranchManager;
 
 import org.apache.flink.api.common.eventtime.WatermarkStrategy;
 import org.apache.flink.api.common.functions.MapFunction;
@@ -139,7 +139,7 @@ public FileStoreITCase(boolean isBatch) {
 
     @BeforeAll
     public static void before() {
-        branch = FileSystemBranchManager.DEFAULT_MAIN_BRANCH;
+        branch = BranchManager.DEFAULT_MAIN_BRANCH;
     }
 
     @Parameters(name = "isBatch-{0}")
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/BranchActionITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/BranchActionITCase.java
@@ -26,7 +26,7 @@
 import org.apache.paimon.types.DataType;
 import org.apache.paimon.types.DataTypes;
 import org.apache.paimon.types.RowType;
-import org.apache.paimon.utils.FileSystemBranchManager;
+import org.apache.paimon.utils.BranchManager;
 import org.apache.paimon.utils.SnapshotManager;
 import org.apache.paimon.utils.TagManager;
 
@@ -76,7 +76,7 @@ void testCreateAndDeleteBranch() throws Exception {
                         "CALL sys.create_tag('%s.%s', 'tag2', 2, '5 d')", database, tableName));
         assertThat(tagManager.tagExists("tag2")).isTrue();
 
-        FileSystemBranchManager branchManager = (FileSystemBranchManager) table.branchManager();
+        BranchManager branchManager = table.branchManager();
         executeSQL(
                 String.format(
                         "CALL sys.create_branch('%s.%s', 'branch_name', 'tag2')",
@@ -157,7 +157,7 @@ void testCreateAndDeleteEmptyBranch() throws Exception {
         writeData(rowData(2L, BinaryString.fromString("Hello")));
         writeData(rowData(3L, BinaryString.fromString("Paimon")));
 
-        FileSystemBranchManager branchManager = (FileSystemBranchManager) table.branchManager();
+        BranchManager branchManager = table.branchManager();
         executeSQL(
                 String.format(
                         "CALL sys.create_branch('%s.%s', 'empty_branch_name')",
@@ -244,7 +244,7 @@ void testFastForward() throws Exception {
         assertThat(tagManager.tagExists("tag3")).isTrue();
 
         // Create branch_name branch
-        FileSystemBranchManager branchManager = (FileSystemBranchManager) table.branchManager();
+        BranchManager branchManager = table.branchManager();
         executeSQL(
                 String.format(
                         "CALL sys.create_branch('%s.%s', 'branch_name', 'tag2')",
diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/AlterBranchProcedureTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/AlterBranchProcedureTest.scala
@@ -19,7 +19,6 @@
 package org.apache.paimon.spark.procedure
 
 import org.apache.paimon.spark.PaimonSparkTestBase
-import org.apache.paimon.utils.FileSystemBranchManager
 
 import org.apache.spark.sql.{Dataset, Row}
 import org.apache.spark.sql.execution.streaming.MemoryStream
@@ -68,7 +67,7 @@ class AlterBranchProcedureTest extends PaimonSparkTestBase with StreamTest {
           checkAnswer(query(), Row(1, "a") :: Row(2, "b2") :: Nil)
 
           val table = loadTable("T")
-          val branchManager = table.branchManager().asInstanceOf[FileSystemBranchManager]
+          val branchManager = table.branchManager()
 
           // create branch with tag
           checkAnswer(
diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/BranchProcedureTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/BranchProcedureTest.scala
@@ -19,7 +19,6 @@
 package org.apache.paimon.spark.procedure
 
 import org.apache.paimon.spark.PaimonSparkTestBase
-import org.apache.paimon.utils.FileSystemBranchManager
 
 import org.apache.spark.sql.{Dataset, Row}
 import org.apache.spark.sql.execution.streaming.MemoryStream
@@ -84,7 +83,7 @@ class BranchProcedureTest extends PaimonSparkTestBase with StreamTest {
                 "CALL paimon.sys.create_branch(table => 'test.T', branch => 'test_branch', tag => 'test_tag')"),
               Row(true) :: Nil)
             val table = loadTable("T")
-            val branchManager = table.branchManager().asInstanceOf[FileSystemBranchManager]
+            val branchManager = table.branchManager()
             assert(branchManager.branchExists("test_branch"))
 
             // query from branch
