This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 80c8f1f781ad65b3e3f9e33ecb6aaa49bbba9039 Author: Eunbin Son <[email protected]> AuthorDate: Thu Sep 17 14:08:29 2026 +0900 [core] Fix FileStore tag manager ignoring the table branch (#9868) --- .../java/org/apache/paimon/AbstractFileStore.java | 3 +- .../paimon/table/PrimaryKeySimpleTableTest.java | 51 ++++++++++++++++++++++ 2 files changed, 52 insertions(+), 2 deletions(-) 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 86839355a7..c8719aae32 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.Collections; 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; @@ -369,7 +368,7 @@ abstract class AbstractFileStore<T> implements FileStore<T> { @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 12339f35b9..140f549d50 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 @@ -720,6 +720,57 @@ public class PrimaryKeySimpleTableTest extends SimpleTableTestBase { "+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();
