[core] Fix FileStore tag manager ignoring the table branch

paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java

@@ -89,7 +89,6 @@
 import java.util.Comparator;
 import java.util.List;
 
-import static org.apache.paimon.catalog.Identifier.DEFAULT_MAIN_BRANCH;
 import static org.apache.paimon.partition.PartitionExpireStrategy.createPartitionExpireStrategy;
 import static org.apache.paimon.utils.Preconditions.checkArgument;
 
@@ -366,7 +365,7 @@ public ChangelogDeletion newChangelogDeletion() {
 
     @Override
     public TagManager newTagManager() {
-        return new TagManager(fileIO, options.path(), DEFAULT_MAIN_BRANCH, options);
+        return new TagManager(fileIO, options.path(), options.branch(), options);
     }
 
     @Override

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

@@ -694,6 +694,57 @@ public void testBranchStreamingReadWrite() throws Exception {
                                 "+2|22|202|binary|varbinary|mapKey:mapVal|multiset"));
     }
 
+    @Test
+    public void testExpireSnapshotsKeepsBranchTag() 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(true, 0));
+        }
+
+        table.createBranch(BRANCH_NAME);
+        FileStoreTable branchTable = table.switchToBranch(BRANCH_NAME);
+        try (StreamTableWrite write = branchTable.newWrite(commitUser);
+                StreamTableCommit commit = branchTable.newCommit(commitUser)) {
+            write.write(rowData(1, 20, 200L));
+            commit.commit(0, write.prepareCommit(true, 0));
+        }
+        long taggedSnapshotId = branchTable.snapshotManager().latestSnapshotId();
+        branchTable.createTag("branch_tag", taggedSnapshotId);
+        try (StreamTableWrite write = branchTable.newWrite(commitUser);
+                StreamTableCommit commit = branchTable.newCommit(commitUser)) {
+            write.write(rowData(1, 30, 300L));
+            commit.commit(1, write.prepareCommit(true, 1));
+            write.write(rowData(1, 40, 400L));
+            commit.commit(2, write.prepareCommit(true, 2));
+        }
+
+        // the store tag manager must resolve tags in the branch tag directory
+        assertThat(branchTable.store().newTagManager().tagDirectory())
+                .isEqualTo(branchTable.tagManager().tagDirectory());
+
+        Options options = new Options();
+        options.set(CoreOptions.SNAPSHOT_NUM_RETAINED_MIN, 1);
+        options.set(CoreOptions.SNAPSHOT_NUM_RETAINED_MAX, 1);
+        options.set(SNAPSHOT_EXPIRE_LIMIT, Integer.MAX_VALUE);
+        branchTable.copy(options.toMap()).newCommit("").expireSnapshots();
+        assertThat(branchTable.snapshotManager().earliestSnapshotId())
+                .isGreaterThan(taggedSnapshotId);
+
+        // files of the tagged snapshot must survive expiration on the branch
+        FileStoreTable tagTable =
+                branchTable.copy(
+                        Collections.singletonMap(CoreOptions.SCAN_TAG_NAME.key(), "branch_tag"));
+        ReadBuilder readBuilder = tagTable.newReadBuilder();
+        assertThat(
+                        getResult(
+                                readBuilder.newRead(),
+                                readBuilder.newScan().plan().splits(),
+                                BATCH_ROW_TO_STRING))
+                .containsExactlyInAnyOrder("1|20|200|binary|varbinary|mapKey:mapVal|multiset");
+    }
+
     @Test
     public void testStreamingProjection() throws Exception {
         writeData();