[core] Validate fast-forward from the current branch (#5413)

Author: liyubin117

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

@@ -77,7 +77,7 @@ static void validateBranch(String branchName) {
                 branchName);
     }
 
-    static void fastForwardValidate(String branchName) {
+    static void fastForwardValidate(String branchName, String currentBranch) {
         checkArgument(
                 !branchName.equals(DEFAULT_MAIN_BRANCH),
                 "Branch name '%s' do not use in fast-forward.",
@@ -86,5 +86,9 @@ static void fastForwardValidate(String branchName) {
                 !StringUtils.isNullOrWhitespaceOnly(branchName),
                 "Branch name '%s' is blank.",
                 branchName);
+        checkArgument(
+                !branchName.equals(currentBranch),
+                "Fast-forward from the current branch '%s' is not allowed.",
+                branchName);
     }
 }

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

@@ -89,7 +89,8 @@ public void dropBranch(String branchName) {
     public void fastForward(String branchName) {
         executePost(
                 catalog -> {
-                    BranchManager.fastForwardValidate(branchName);
+                    BranchManager.fastForwardValidate(
+                            branchName, identifier.getBranchNameOrDefault());
                     catalog.fastForward(identifier, branchName);
                 });
     }

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

@@ -138,7 +138,7 @@ private boolean fileExists(Path path) {
 
     @Override
     public void fastForward(String branchName) {
-        BranchManager.fastForwardValidate(branchName);
+        BranchManager.fastForwardValidate(branchName, snapshotManager.branch());
         checkArgument(branchExists(branchName), "Branch name '%s' doesn't exist.", branchName);
 
         Long earliestSnapshotId = snapshotManager.copyWithBranch(branchName).earliestSnapshotId();

paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BranchSqlITCase.java

@@ -280,6 +280,12 @@ public void testBranchFastForward() throws Exception {
                 .containsExactlyInAnyOrder("+I[1, 10, hunter]", "+I[2, 10, hunterX]");
 
         checkSnapshots(snapshotManager, 1, 2);
+
+        sql("alter table T set ('branch'='test')");
+
+        assertThatThrownBy(() -> sql("CALL sys.fast_forward('default.T', 'test')"))
+                .hasMessageContaining(
+                        "Fast-forward from the current branch 'test' is not allowed.");
     }
 
     @Test