[core] Prevent tag deletion when referenced by branches

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

@@ -656,6 +656,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,14 @@ 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");
+
+        // deleteBranch also removes the manually added files in branch directory,
+        // so we need to remove them from manuallyAddedFiles to avoid validation failure
+        String branchPathPrefix = branchPath(tablePath, "branch1").toString();
+        manuallyAddedFiles.removeIf(path -> path.toString().startsWith(branchPathPrefix));
+
         // randomly delete tags
         List<String> deleteTags = Collections.emptyList();
         deleteTags = randomlyPick(allTags);
@@ -288,6 +296,14 @@ 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");
+
+        // deleteBranch also removes the manually added files in branch directory,
+        // so we need to remove them from manuallyAddedFiles to avoid validation failure
+        String branchPathPrefix = branchPath(tablePath, "branch1").toString();
+        manuallyAddedFiles.removeIf(path -> path.toString().startsWith(branchPathPrefix));
+
         // 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();