[core] Prevent tag deletion when referenced by branches

This closes #6272.

When a branch is created from a tag, the tag's snapshot information is copied
to the branch directory. However, deleting the tag could cause the snapshot
expiration process to clean up data files that are still needed by the branch,
making the branch unqueryable.

This change adds a check in deleteTag() to verify if any branches were created
from the tag before allowing deletion. If branches reference the tag, the
deletion is blocked with an error message listing the referencing branches.

Changes:
- Added branchesCreatedFromTag() method to BranchManager interface with
  default empty implementation for backward compatibility
- Implemented branchesCreatedFromTag() in FileSystemBranchManager to check
  each branch's tag directory for the specified tag
- Added validation in AbstractFileStoreTable.deleteTag() to prevent deletion
  of tags that are still referenced by branches
- Added testDeleteTagReferencedByBranch() test case to verify the fix

Author: userzhy

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

@@ -646,6 +646,14 @@ public void replaceTag(
 
     @Override
     public void deleteTag(String tagName) {
+        List<String> referencingBranches = branchManager().branchesCreatedFromTag(tagName);
+        if (!referencingBranches.isEmpty()) {
+            throw new IllegalStateException(
+                    String.format(
+                            "Cannot delete tag '%s' because it is still referenced by branches: %s. "
+                                    + "Please delete these branches first.",
+                            tagName, referencingBranches));
+        }
         tagManager()
                 .deleteTag(
                         tagName,

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

@@ -42,6 +42,16 @@ public interface BranchManager {
 
     List<String> branches();
 
+    /**
+     * Get all branches that were created based on the given tag.
+     *
+     * @param tagName the name of the tag to check
+     * @return list of branch names that reference the given tag
+     */
+    default List<String> branchesCreatedFromTag(String tagName) {
+        return java.util.Collections.emptyList();
+    }
+
     default boolean branchExists(String branchName) {
         return branches().contains(branchName);
     }

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

@@ -215,6 +215,18 @@ public List<String> branches() {
         }
     }
 
+    @Override
+    public List<String> branchesCreatedFromTag(String tagName) {
+        List<String> result = new java.util.ArrayList<>();
+        for (String branchName : branches()) {
+            TagManager branchTagManager = tagManager.copyWithBranch(branchName);
+            if (branchTagManager.tagExists(tagName)) {
+                result.add(branchName);
+            }
+        }
+        return result;
+    }
+
     private void copySchemasToBranch(String branchName, long schemaId) throws IOException {
         for (int i = 0; i <= schemaId; i++) {
             if (schemaManager.schemaExists(i)) {

paimon-core/src/test/java/org/apache/paimon/operation/LocalOrphanFilesCleanTest.java

@@ -185,6 +185,9 @@ public void normallyRemoving(Path dataPath) throws Throwable {
         expireOptions.set(CoreOptions.SNAPSHOT_NUM_RETAINED_MAX, snapshotCount - expired);
         table.copy(expireOptions.toMap()).newCommit("").expireSnapshots();
 
+        // delete branch1 first before deleting tags
+        table.deleteBranch("branch1");
+
         // randomly delete tags
         List<String> deleteTags = Collections.emptyList();
         deleteTags = randomlyPick(allTags);
@@ -288,6 +291,9 @@ public void testNormallyRemovingMixedWithExternalPath() throws Throwable {
         expireOptions.set(CoreOptions.SNAPSHOT_NUM_RETAINED_MAX, snapshotCount - expired);
         table.copy(expireOptions.toMap()).newCommit("").expireSnapshots();
 
+        // delete branch1 first before deleting tags
+        table.deleteBranch("branch1");
+
         // randomly delete tags
         List<String> deleteTags = Collections.emptyList();
         deleteTags = randomlyPick(allTags);

paimon-core/src/test/java/org/apache/paimon/rest/RESTSimpleTableTest.java

@@ -32,6 +32,8 @@
 
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
 
 import java.io.IOException;
 import java.util.ArrayList;
@@ -106,4 +108,9 @@ protected FileStoreTable createBranchTable(String branch) throws Exception {
                         new Identifier(
                                 identifier.getDatabaseName(), identifier.getTableName(), branch));
     }
+
+    @Test
+    @Disabled("REST catalog does not support branchesCreatedFromTag yet")
+    @Override
+    public void testDeleteTagReferencedByBranch() {}
 }

paimon-core/src/test/java/org/apache/paimon/table/SimpleTableTestBase.java

@@ -1174,6 +1174,46 @@ public void testDeleteBranch() throws Exception {
         table.deleteBranch("fallback");
     }
 
+    @Test
+    public void testDeleteTagReferencedByBranch() throws Exception {
+        FileStoreTable table = createFileStoreTable();
+
+        try (StreamTableWrite write = table.newWrite(commitUser);
+                StreamTableCommit commit = table.newCommit(commitUser)) {
+            write.write(rowData(1, 10, 100L));
+            commit.commit(0, write.prepareCommit(false, 1));
+        }
+
+        table.createTag("tag1", 1);
+        table.createBranch("branch1", "tag1");
+
+        // verify that deleting a tag referenced by a branch fails
+        assertThatThrownBy(() -> table.deleteTag("tag1"))
+                .satisfies(
+                        anyCauseMatches(
+                                IllegalStateException.class,
+                                "Cannot delete tag 'tag1' because it is still referenced by branches: [branch1]"));
+
+        // create another branch from the same tag
+        table.createBranch("branch2", "tag1");
+
+        // verify that deleting the tag still fails and shows both branches
+        assertThatThrownBy(() -> table.deleteTag("tag1"))
+                .satisfies(
+                        anyCauseMatches(
+                                IllegalStateException.class,
+                                "Cannot delete tag 'tag1' because it is still referenced by branches:"));
+
+        // delete both branches
+        table.deleteBranch("branch1");
+        table.deleteBranch("branch2");
+
+        // verify that deleting the tag succeeds after branches are deleted
+        table.deleteTag("tag1");
+        TagManager tagManager = new TagManager(table.fileIO(), table.location());
+        assertThat(tagManager.tagExists("tag1")).isFalse();
+    }
+
     @Test
     public void testFastForward() throws Exception {
         FileStoreTable table = createFileStoreTable();