[core] Only remove tag metadata when deleting tags (#10823)

Author: JingsongLi

docs/docs/concepts/spec/manifest.md

@@ -61,7 +61,8 @@ skip manifests before opening them.
 | `_EXTRA_FILES` | ARRAY of STRING, nullable | Names of additional files in the manifest directory; defaults to null. |
 
 Each extra file belongs exclusively to one manifest. It is retained and cleaned up together with
-that manifest during snapshot, tag, or changelog deletion.
+that manifest during snapshot or changelog expiration and orphan file cleanup. Tag deletion retains
+both the manifest and its extra files for later orphan file cleanup.
 
 ### Manifest Sidecar
 

docs/docs/maintenance/manage-snapshots.mdx

@@ -398,9 +398,10 @@ changelog producer.
 
 ## Remove Orphan Files
 
-Paimon files are deleted physically only when expiring snapshots. However, it is possible that some unexpected errors occurred
-when deleting files, so that there may exist files that are not used by Paimon snapshots (so-called "orphan files"). You can
-submit a `remove_orphan_files` job to clean them:
+Snapshot expiration physically deletes unused files. Deleting a tag only removes its tag file,
+and unexpected errors during file deletion can also leave unused files behind. Files that are no longer
+referenced by any snapshot, tag, or changelog are called "orphan files". You can submit a `remove_orphan_files` job
+to clean them:
 
 <Tabs groupId="remove-orphan-files">
 

docs/docs/maintenance/manage-tags.mdx

@@ -177,6 +177,10 @@ CALL sys.create_tag(table => 'test.t', tag => 'test_tag');
 
 You can delete a tag by its name.
 
+Deleting a tag only removes the tag file. Its manifest, index, and data files are retained.
+Use [Remove Orphan Files](./manage-snapshots#remove-orphan-files) to reclaim files that are no longer
+referenced by any snapshot, tag, or changelog, subject to the orphan file cleanup retention period.
+
 <Tabs groupId="delete-tag">
 
 <TabItem value="flink-sql" label="Flink SQL">

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

@@ -46,7 +46,6 @@
 import org.apache.paimon.operation.NormalPartitionExpire;
 import org.apache.paimon.operation.PartitionExpire;
 import org.apache.paimon.operation.SnapshotDeletion;
-import org.apache.paimon.operation.TagDeletion;
 import org.apache.paimon.operation.commit.CommitRollback;
 import org.apache.paimon.operation.commit.ConflictDetection;
 import org.apache.paimon.partition.PartitionExpireStrategy;
@@ -367,20 +366,6 @@ public TagManager newTagManager() {
         return new TagManager(fileIO, options.path(), options.branch(), options);
     }
 
-    @Override
-    public TagDeletion newTagDeletion() {
-        return new TagDeletion(
-                fileIO,
-                pathFactory(),
-                manifestFileFactory().create(),
-                manifestListFactory().create(),
-                newIndexFileHandler(),
-                newStatsFileHandler(),
-                options.cleanEmptyDirectories(),
-                options.fileOperationThreadNum(),
-                options.scanManifestParallelism());
-    }
-
     public abstract Comparator<InternalRow> newKeyComparator();
 
     @Override
@@ -573,11 +558,7 @@ private PartitionModification newPartitionModificationForBranch(String branchNam
     @Override
     public TagAutoManager newTagAutoManager(FileStoreTable table) {
         return TagAutoManager.create(
-                options,
-                snapshotManager(),
-                newTagManager(),
-                newTagDeletion(),
-                createTagCallbacks(table));
+                options, snapshotManager(), newTagManager(), createTagCallbacks(table));
     }
 
     @Override

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

@@ -30,7 +30,6 @@
 import org.apache.paimon.operation.PartitionExpire;
 import org.apache.paimon.operation.SnapshotDeletion;
 import org.apache.paimon.operation.SplitRead;
-import org.apache.paimon.operation.TagDeletion;
 import org.apache.paimon.partition.PartitionExpireStrategy;
 import org.apache.paimon.service.ServiceManager;
 import org.apache.paimon.stats.StatsFileHandler;
@@ -100,8 +99,6 @@ public interface FileStore<T> {
 
     TagManager newTagManager();
 
-    TagDeletion newTagDeletion();
-
     @Nullable
     PartitionExpire newPartitionExpire(String commitUser, FileStoreTable table);
 

paimon-core/src/main/java/org/apache/paimon/operation/TagDeletion.java

@@ -769,12 +769,7 @@ public void replaceTag(
 
     @Override
     public void deleteTag(String tagName) {
-        tagManager()
-                .deleteTag(
-                        tagName,
-                        store().newTagDeletion(),
-                        snapshotManager(),
-                        store().createTagCallbacks(this));
+        tagManager().deleteTag(tagName, store().createTagCallbacks(this));
     }
 
     @Override

paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java

@@ -143,7 +143,7 @@ default String uuid() {
     @Experimental
     void replaceTag(String tagName, @Nullable Long fromSnapshotId, @Nullable Duration timeRetained);
 
-    /** Delete a tag by name. */
+    /** Delete a tag by name, leaving its referenced files to orphan file cleanup. */
     @Experimental
     void deleteTag(String tagName);
 

paimon-core/src/main/java/org/apache/paimon/table/Table.java

@@ -78,10 +78,8 @@ public String createReadTag(long snapshotId) {
     public void deleteReadTag(String tagName) {
         try {
             if (tagManager.tagExists(tagName)) {
-                // Directly delete the tag metadata file instead of using TagManager.deleteTag(),
-                // which would also scan and delete unreferenced data files — too heavyweight for a
-                // read-path cleanup. Any orphan data files left behind will be reclaimed by
-                // OrphanFilesClean.
+                // Read protection tags do not need deletion callbacks. Any orphan files left
+                // behind will be reclaimed by OrphanFilesClean.
                 snapshotManager.fileIO().deleteQuietly(tagManager.tagPath(tagName));
                 LOG.info("Deleted batch read protection tag '{}'.", tagName);
             }

paimon-core/src/main/java/org/apache/paimon/tag/BatchReadTagCreator.java

@@ -21,7 +21,6 @@
 import org.apache.paimon.CoreOptions;
 import org.apache.paimon.Snapshot;
 import org.apache.paimon.annotation.VisibleForTesting;
-import org.apache.paimon.operation.TagDeletion;
 import org.apache.paimon.table.sink.TagCallback;
 import org.apache.paimon.tag.TagTimeExtractor.ProcessTimeExtractor;
 import org.apache.paimon.tag.TagTimeExtractor.WatermarkExtractor;
@@ -53,7 +52,6 @@ public class TagAutoCreation {
 
     private final SnapshotManager snapshotManager;
     private final TagManager tagManager;
-    private final TagDeletion tagDeletion;
     private final TagTimeExtractor timeExtractor;
     private final TagPeriodHandler periodHandler;
     private final Duration delay;
@@ -70,7 +68,6 @@ public class TagAutoCreation {
     private TagAutoCreation(
             SnapshotManager snapshotManager,
             TagManager tagManager,
-            TagDeletion tagDeletion,
             TagTimeExtractor timeExtractor,
             TagPeriodHandler periodHandler,
             Duration delay,
@@ -82,7 +79,6 @@ private TagAutoCreation(
             ZoneId sinkProcessTimeZone) {
         this.snapshotManager = snapshotManager;
         this.tagManager = tagManager;
-        this.tagDeletion = tagDeletion;
         this.timeExtractor = timeExtractor;
         this.periodHandler = periodHandler;
         this.delay = delay;
@@ -196,7 +192,7 @@ private void tryToCreateTags(Snapshot snapshot) {
                                 "Delete tag {}, because the number of auto-created tags reached numRetainedMax of {}.",
                                 toDeleteTag,
                                 numRetainedMax);
-                        tagManager.deleteTag(toDeleteTag, tagDeletion, snapshotManager, callbacks);
+                        tagManager.deleteTag(toDeleteTag, callbacks);
                         i++;
                         if (i == toDelete) {
                             break;
@@ -225,7 +221,6 @@ public static TagAutoCreation create(
             CoreOptions options,
             SnapshotManager snapshotManager,
             TagManager tagManager,
-            TagDeletion tagDeletion,
             List<TagCallback> callbacks) {
         TagTimeExtractor extractor = TagTimeExtractor.createForAutoTag(options);
         if (extractor == null) {
@@ -234,7 +229,6 @@ public static TagAutoCreation create(
         return new TagAutoCreation(
                 snapshotManager,
                 tagManager,
-                tagDeletion,
                 extractor,
                 TagPeriodHandler.create(options),
                 options.tagCreationDelay(),