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 524dd4962c89..4867535ee00f 100644 --- 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 index cf8388fc55ee..800993309f91 100644 --- 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 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 index 31ca385d1e34..4dae2f660cb6 100644 --- 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 callbacks) { + List 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 index 0b57dfa884e9..9e12e7e36bf2 100644 --- 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,6 +45,7 @@ public class TagTimeExpire { private final TagManager tagManager; private final TagDeletion tagDeletion; private final List callbacks; + private final BranchManager branchManager; private LocalDateTime olderThanTime; @@ -51,11 +53,13 @@ private TagTimeExpire( SnapshotManager snapshotManager, TagManager tagManager, TagDeletion tagDeletion, - List callbacks) { + List callbacks, + BranchManager branchManager) { this.snapshotManager = snapshotManager; this.tagManager = tagManager; this.tagDeletion = tagDeletion; this.callbacks = callbacks; + this.branchManager = branchManager; } public List expire() { @@ -88,6 +92,15 @@ public List expire() { && LocalDateTime.now().isAfter(createTime.plus(timeRetained)); boolean isOlderThan = olderThanTime != null && olderThanTime.isAfter(createTime); if (isReachTimeRetained || isOlderThan) { + List 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 callbacks) { - return new TagTimeExpire(snapshotManager, tagManager, tagDeletion, callbacks); + List 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 index 085f04747ca1..6670a335217e 100644 --- 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 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 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 index 9f97cf3a097d..9390d4585285 100644 --- 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,9 +21,12 @@ 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. */ @@ -31,10 +34,13 @@ 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 func) { @@ -122,4 +128,15 @@ public void renameBranch(String fromBranch, String toBranch) { public List branches() { return executeGet(catalog -> catalog.listBranches(identifier)); } + + @Override + public List branchesCreatedFromTag(String tagName) { + List 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 index 9466ccd851cc..8007bb73e6a4 100644 --- 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 branches() { } } + @Override + public List branchesCreatedFromTag(String tagName) { + List 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 index cd6480d7e74c..feb0777dbafb 100644 --- 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 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 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 index 8c7266939f89..df276871f1ea 100644 --- 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 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();