diff --git a/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java b/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java index 7399e057783c..5857fedb94e9 100644 --- a/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java +++ b/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 diff --git a/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java index 6a7a722e855b..7ad1de197256 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java +++ b/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();