apache · ArnavBalyan · Jul 26, 2026
diff --git a/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java b/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java
@@ -570,7 +570,8 @@ public TagAutoManager newTagAutoManager(FileStoreTable table) {
                 snapshotManager(),
                 newTagManager(),
                 newTagDeletion(),
-                createTagCallbacks(table));
+                createTagCallbacks(table),
+                table.branchManager());
     }
 
     @Override

diff --git a/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java b/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java
@@ -666,6 +666,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,
@@ -744,7 +752,8 @@ public TagManager tagManager() {
     public BranchManager branchManager() {
         if (catalogEnvironment.catalogLoader() != null
                 && catalogEnvironment.supportsVersionManagement()) {
-            return new CatalogBranchManager(catalogEnvironment.catalogLoader(), identifier());
+            return new CatalogBranchManager(
+                    catalogEnvironment.catalogLoader(), identifier(), fileIO, path);
         }
         return new FileSystemBranchManager(
                 fileIO,

diff --git a/paimon-core/src/main/java/org/apache/paimon/tag/TagAutoManager.java b/paimon-core/src/main/java/org/apache/paimon/tag/TagAutoManager.java
@@ -21,6 +21,7 @@
 import org.apache.paimon.CoreOptions;
 import org.apache.paimon.operation.TagDeletion;
 import org.apache.paimon.table.sink.TagCallback;
+import org.apache.paimon.utils.BranchManager;
 import org.apache.paimon.utils.SnapshotManager;
 import org.apache.paimon.utils.TagManager;
 
@@ -54,7 +55,8 @@ public static TagAutoManager create(
             SnapshotManager snapshotManager,
             TagManager tagManager,
             TagDeletion tagDeletion,
-            List<TagCallback> callbacks) {
+            List<TagCallback> callbacks,
+            BranchManager branchManager) {
 
         TagTimeExtractor extractor = TagTimeExtractor.createForAutoTag(options);
 
@@ -64,7 +66,8 @@ public static TagAutoManager create(
                         : TagAutoCreation.create(
                                 options, snapshotManager, tagManager, tagDeletion, callbacks),
                 options.tagTimeExpireEnabled()
-                        ? TagTimeExpire.create(snapshotManager, tagManager, tagDeletion, callbacks)
+                        ? TagTimeExpire.create(
+                                snapshotManager, tagManager, tagDeletion, callbacks, branchManager)
                         : null);
     }
 

diff --git a/paimon-core/src/main/java/org/apache/paimon/tag/TagTimeExpire.java b/paimon-core/src/main/java/org/apache/paimon/tag/TagTimeExpire.java
@@ -21,6 +21,7 @@
 import org.apache.paimon.fs.FileStatus;
 import org.apache.paimon.operation.TagDeletion;
 import org.apache.paimon.table.sink.TagCallback;
+import org.apache.paimon.utils.BranchManager;
 import org.apache.paimon.utils.DateTimeUtils;
 import org.apache.paimon.utils.Pair;
 import org.apache.paimon.utils.SnapshotManager;
@@ -44,18 +45,21 @@ public class TagTimeExpire {
     private final TagManager tagManager;
     private final TagDeletion tagDeletion;
     private final List<TagCallback> callbacks;
+    private final BranchManager branchManager;
 
     private LocalDateTime olderThanTime;
 
     private TagTimeExpire(
             SnapshotManager snapshotManager,
             TagManager tagManager,
             TagDeletion tagDeletion,
-            List<TagCallback> callbacks) {
+            List<TagCallback> callbacks,
+            BranchManager branchManager) {
         this.snapshotManager = snapshotManager;
         this.tagManager = tagManager;
         this.tagDeletion = tagDeletion;
         this.callbacks = callbacks;
+        this.branchManager = branchManager;
     }
 
     public List<String> expire() {
@@ -88,6 +92,15 @@ public List<String> expire() {
                             && LocalDateTime.now().isAfter(createTime.plus(timeRetained));
             boolean isOlderThan = olderThanTime != null && olderThanTime.isAfter(createTime);
             if (isReachTimeRetained || isOlderThan) {
+                List<String> referencingBranches = branchManager.branchesCreatedFromTag(tagName);
+                if (!referencingBranches.isEmpty()) {
+                    LOG.warn(
+                            "Skip expiring tag {} because it is still referenced by branches: {}. "
+                                    + "Delete these branches to allow the tag to expire.",
+                            tagName,
+                            referencingBranches);
+                    continue;
+                }
                 LOG.info(
                         "Delete tag {}, because its existence time has reached its timeRetained of {} or"
                                 + " its createTime {} is olderThan olderThanTime {}.",
@@ -111,7 +124,9 @@ public static TagTimeExpire create(
             SnapshotManager snapshotManager,
             TagManager tagManager,
             TagDeletion tagDeletion,
-            List<TagCallback> callbacks) {
-        return new TagTimeExpire(snapshotManager, tagManager, tagDeletion, callbacks);
+            List<TagCallback> callbacks,
+            BranchManager branchManager) {
+        return new TagTimeExpire(
+                snapshotManager, tagManager, tagDeletion, callbacks, branchManager);
     }
 }
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
@@ -22,6 +22,7 @@
 
 import javax.annotation.Nullable;
 
+import java.util.Collections;
 import java.util.List;
 
 import static org.apache.paimon.catalog.Identifier.DEFAULT_MAIN_BRANCH;
@@ -65,6 +66,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 Collections.emptyList();
+    }
+
     default boolean branchExists(String branchName) {
         return branches().contains(branchName);
     }

diff --git a/paimon-core/src/main/java/org/apache/paimon/utils/CatalogBranchManager.java b/paimon-core/src/main/java/org/apache/paimon/utils/CatalogBranchManager.java
@@ -21,20 +21,26 @@
 import org.apache.paimon.catalog.Catalog;
 import org.apache.paimon.catalog.CatalogLoader;
 import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
 
 import javax.annotation.Nullable;
 
+import java.util.ArrayList;
 import java.util.List;
 
 /** A {@link BranchManager} implementation to manage branches via catalog. */
 public class CatalogBranchManager implements BranchManager {
 
     private final CatalogLoader catalogLoader;
     private final Identifier identifier;
+    private final TagManager tagManager;
 
-    public CatalogBranchManager(CatalogLoader catalogLoader, Identifier identifier) {
+    public CatalogBranchManager(
+            CatalogLoader catalogLoader, Identifier identifier, FileIO fileIO, Path tablePath) {
         this.catalogLoader = catalogLoader;
         this.identifier = identifier;
+        this.tagManager = new TagManager(fileIO, tablePath);
     }
 
     private void executePost(ThrowingConsumer<Catalog, Exception> func) {
@@ -122,4 +128,15 @@ public void renameBranch(String fromBranch, String toBranch) {
     public List<String> branches() {
         return executeGet(catalog -> catalog.listBranches(identifier));
     }
+
+    @Override
+    public List<String> branchesCreatedFromTag(String tagName) {
+        List<String> result = new ArrayList<>();
+        for (String branchName : branches()) {
+            if (tagManager.copyWithBranch(branchName).tagExists(tagName)) {
+                result.add(branchName);
+            }
+        }
+        return result;
+    }
 }
diff --git a/paimon-core/src/main/java/org/apache/paimon/utils/FileSystemBranchManager.java b/paimon-core/src/main/java/org/apache/paimon/utils/FileSystemBranchManager.java
@@ -458,6 +458,17 @@ public List<String> branches() {
         }
     }
 
+    @Override
+    public List<String> branchesCreatedFromTag(String tagName) {
+        List<String> result = new ArrayList<>();
+        for (String branchName : branches()) {
+            if (tagManager.copyWithBranch(branchName).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)) {

diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/LocalOrphanFilesCleanTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/LocalOrphanFilesCleanTest.java
@@ -211,6 +211,10 @@ public void normallyRemoving(Path dataPath) throws Throwable {
         expireOptions.set(CoreOptions.SNAPSHOT_NUM_RETAINED_MAX, snapshotCount - expired);
         table.copy(expireOptions.toMap()).newCommit("").expireSnapshots();
 
+        table.deleteBranch("branch1");
+        String branchPathPrefix = branchPath(tablePath, "branch1").toString();
+        manuallyAddedFiles.removeIf(path -> path.toString().startsWith(branchPathPrefix));
+
         // randomly delete tags
         List<String> deleteTags = Collections.emptyList();
         deleteTags = randomlyPick(allTags);
@@ -314,6 +318,10 @@ public void testNormallyRemovingMixedWithExternalPath() throws Throwable {
         expireOptions.set(CoreOptions.SNAPSHOT_NUM_RETAINED_MAX, snapshotCount - expired);
         table.copy(expireOptions.toMap()).newCommit("").expireSnapshots();
 
+        table.deleteBranch("branch1");
+        String branchPathPrefix = branchPath(tablePath, "branch1").toString();
+        manuallyAddedFiles.removeIf(path -> path.toString().startsWith(branchPathPrefix));
+
         // randomly delete tags
         List<String> deleteTags = Collections.emptyList();
         deleteTags = randomlyPick(allTags);

diff --git a/paimon-core/src/test/java/org/apache/paimon/table/SimpleTableTestBase.java b/paimon-core/src/test/java/org/apache/paimon/table/SimpleTableTestBase.java
@@ -83,6 +83,7 @@
 import org.junit.jupiter.params.provider.ValueSource;
 
 import java.io.IOException;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
@@ -1274,6 +1275,63 @@ 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");
+
+        assertThatThrownBy(() -> table.deleteTag("tag1"))
+                .satisfies(
+                        anyCauseMatches(
+                                IllegalStateException.class,
+                                "Cannot delete tag 'tag1' because it is still referenced by branches: [branch1]"));
+
+        table.createBranch("branch2", "tag1");
+        assertThatThrownBy(() -> table.deleteTag("tag1"))
+                .satisfies(
+                        anyCauseMatches(
+                                IllegalStateException.class,
+                                "Cannot delete tag 'tag1' because it is still referenced by branches:"));
+
+        table.deleteBranch("branch1");
+        table.deleteBranch("branch2");
+
+        table.deleteTag("tag1");
+        assertThat(table.tagManager().tagExists("tag1")).isFalse();
+    }
+
+    @Test
+    public void testExpireTagsSkipsTagReferencedByBranch() 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, Duration.ofMillis(1));
+        table.createBranch("branch1", "tag1");
+        Thread.sleep(10);
+
+        List<String> expired = table.store().newTagAutoManager(table).getTagTimeExpire().expire();
+        assertThat(expired).isEmpty();
+        assertThat(table.tagManager().tagExists("tag1")).isTrue();
+
+        table.deleteBranch("branch1");
+        expired = table.store().newTagAutoManager(table).getTagTimeExpire().expire();
+        assertThat(expired).containsExactly("tag1");
+        assertThat(table.tagManager().tagExists("tag1")).isFalse();
+    }
+
     @Test
     public void testFastForward() throws Exception {
         FileStoreTable table = createFileStoreTable();