This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 3304abe262 [core] Only remove tag metadata when deleting tags (#10823)
3304abe262 is described below
commit 3304abe2624b069f5242bae7b2ba9fe377bd930d
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Oct 27 12:58:02 2026 +0800
[core] Only remove tag metadata when deleting tags (#10823)
---
docs/docs/concepts/spec/manifest.md | 3 +-
docs/docs/maintenance/manage-snapshots.mdx | 7 +-
docs/docs/maintenance/manage-tags.mdx | 4 +
.../java/org/apache/paimon/AbstractFileStore.java | 21 +---
.../src/main/java/org/apache/paimon/FileStore.java | 3 -
.../org/apache/paimon/operation/TagDeletion.java | 124 --------------------
.../paimon/table/AbstractFileStoreTable.java | 7 +-
.../main/java/org/apache/paimon/table/Table.java | 2 +-
.../org/apache/paimon/tag/BatchReadTagCreator.java | 6 +-
.../org/apache/paimon/tag/TagAutoCreation.java | 8 +-
.../java/org/apache/paimon/tag/TagAutoManager.java | 7 +-
.../org/apache/paimon/tag/TagBatchCreation.java | 20 +---
.../java/org/apache/paimon/tag/TagTimeExpire.java | 17 +--
.../java/org/apache/paimon/utils/TagManager.java | 127 ++------------------
.../DataEvolutionRowIdReassignerTest.java | 3 +
.../paimon/operation/ExpireSnapshotsTest.java | 14 +--
.../apache/paimon/operation/FileDeletionTest.java | 129 +++++++++++++++------
.../operation/LocalOrphanFilesCleanTest.java | 6 +-
.../paimon/table/IndexFileExpireTableTest.java | 9 +-
.../org/apache/paimon/tag/TagAutoManagerTest.java | 1 -
.../procedure/RollbackToAsLatestProcedure.java | 6 +-
.../sink/AutoTagForSavepointCommitterOperator.java | 15 +--
...utoTagForSavepointCommitterOperatorFactory.java | 6 -
.../org/apache/paimon/flink/sink/FlinkSink.java | 1 -
.../paimon/flink/sink/RowAppendTableSink.java | 4 -
.../paimon/flink/sink/SavepointTagUtils.java | 9 +-
.../flink/sink/coordinator/SavepointTagger.java | 7 +-
.../AutoTagForSavepointCommitterOperatorTest.java | 2 -
.../paimon/flink/sink/SavepointTagUtilsTest.java | 2 -
.../CommittingWriteOperatorCoordinatorTest.java | 1 -
.../sink/coordinator/SavepointTaggerTest.java | 1 -
.../procedure/RollbackToAsLatestProcedure.java | 6 +-
32 files changed, 154 insertions(+), 424 deletions(-)
diff --git a/docs/docs/concepts/spec/manifest.md
b/docs/docs/concepts/spec/manifest.md
index 6f72db9067..a23106c874 100644
--- a/docs/docs/concepts/spec/manifest.md
+++ b/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
diff --git a/docs/docs/maintenance/manage-snapshots.mdx
b/docs/docs/maintenance/manage-snapshots.mdx
index 205894d7b7..656a8a76ee 100644
--- a/docs/docs/maintenance/manage-snapshots.mdx
+++ b/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">
diff --git a/docs/docs/maintenance/manage-tags.mdx
b/docs/docs/maintenance/manage-tags.mdx
index 490f264a41..54b1ab780c 100644
--- a/docs/docs/maintenance/manage-tags.mdx
+++ b/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">
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 9e554ce8e2..8663945778 100644
--- a/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java
+++ b/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java
@@ -46,7 +46,6 @@ import org.apache.paimon.operation.ManifestsReader;
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 @@ abstract class AbstractFileStore<T> implements
FileStore<T> {
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 @@ abstract class AbstractFileStore<T> implements
FileStore<T> {
@Override
public TagAutoManager newTagAutoManager(FileStoreTable table) {
return TagAutoManager.create(
- options,
- snapshotManager(),
- newTagManager(),
- newTagDeletion(),
- createTagCallbacks(table));
+ options, snapshotManager(), newTagManager(),
createTagCallbacks(table));
}
@Override
diff --git a/paimon-core/src/main/java/org/apache/paimon/FileStore.java
b/paimon-core/src/main/java/org/apache/paimon/FileStore.java
index d9b143f75f..a948d1505e 100644
--- a/paimon-core/src/main/java/org/apache/paimon/FileStore.java
+++ b/paimon-core/src/main/java/org/apache/paimon/FileStore.java
@@ -30,7 +30,6 @@ import org.apache.paimon.operation.FileStoreWrite;
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);
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/TagDeletion.java
b/paimon-core/src/main/java/org/apache/paimon/operation/TagDeletion.java
deleted file mode 100644
index e6f9d9d673..0000000000
--- a/paimon-core/src/main/java/org/apache/paimon/operation/TagDeletion.java
+++ /dev/null
@@ -1,124 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.paimon.operation;
-
-import org.apache.paimon.Snapshot;
-import org.apache.paimon.data.BinaryRow;
-import org.apache.paimon.fs.FileIO;
-import org.apache.paimon.fs.Path;
-import org.apache.paimon.index.IndexFileHandler;
-import org.apache.paimon.io.DataFilePathFactory;
-import org.apache.paimon.manifest.ExpireFileEntry;
-import org.apache.paimon.manifest.ManifestFile;
-import org.apache.paimon.manifest.ManifestFileMeta;
-import org.apache.paimon.manifest.ManifestList;
-import org.apache.paimon.stats.StatsFileHandler;
-import org.apache.paimon.utils.DataFilePathFactories;
-import org.apache.paimon.utils.FileStorePathFactory;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import javax.annotation.Nullable;
-
-import java.io.IOException;
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.function.Predicate;
-
-/** Delete tag files. */
-public class TagDeletion extends FileDeletionBase<Snapshot> {
-
- private static final Logger LOG =
LoggerFactory.getLogger(TagDeletion.class);
-
- public TagDeletion(
- FileIO fileIO,
- FileStorePathFactory pathFactory,
- ManifestFile manifestFile,
- ManifestList manifestList,
- IndexFileHandler indexFileHandler,
- StatsFileHandler statsFileHandler,
- boolean cleanEmptyDirectories,
- int fileOperationThreadNum,
- @Nullable Integer manifestReadParallelism) {
- super(
- fileIO,
- pathFactory,
- manifestFile,
- manifestList,
- indexFileHandler,
- statsFileHandler,
- cleanEmptyDirectories,
- fileOperationThreadNum,
- manifestReadParallelism);
- }
-
- @Override
- public void cleanDeletedDataFiles(Snapshot taggedSnapshot,
Predicate<ExpireFileEntry> skipper) {
- Collection<ExpireFileEntry> manifestEntries;
- try {
- List<ManifestFileMeta> manifests =
- tryReadManifestList(taggedSnapshot.baseManifestList());
-
manifests.addAll(tryReadManifestList(taggedSnapshot.deltaManifestList()));
- manifestEntries = readMergedDataFiles(manifests);
- } catch (IOException e) {
- LOG.info("Skip data file clean for the tag of id {}.",
taggedSnapshot.id(), e);
- return;
- }
-
- Set<Path> dataFileToDelete = new HashSet<>();
- DataFilePathFactories factories = new
DataFilePathFactories(pathFactory);
- for (ExpireFileEntry entry : manifestEntries) {
- DataFilePathFactory dataFilePathFactory =
- factories.get(entry.partition(), entry.bucket());
- if (!skipper.test(entry)) {
- dataFileToDelete.add(dataFilePathFactory.toPath(entry));
- for (String file : entry.extraFiles()) {
-
dataFileToDelete.add(dataFilePathFactory.toAlignedPath(file, entry));
- }
-
- recordDeletionBuckets(entry);
- }
- }
- executeAll(dataFileToDelete, fileIO::deleteQuietly);
- }
-
- public void cleanUnusedDataFiles(Snapshot taggedSnapshot,
Predicate<ExpireFileEntry> skipper) {
- cleanDeletedDataFiles(taggedSnapshot, skipper);
- }
-
- @Override
- public void cleanUnusedManifests(Snapshot taggedSnapshot, Set<String>
skippingSet) {
- // doesn't clean changelog files because they are handled by
SnapshotDeletion
- executeAll(planManifestsCleaner(taggedSnapshot, skippingSet, true,
false));
- }
-
- public Predicate<ExpireFileEntry> dataFileSkipper(List<Snapshot>
fromSnapshots)
- throws Exception {
- Map<BinaryRow, Map<Integer, Set<String>>> skipped = new HashMap<>();
- for (Snapshot snapshot : fromSnapshots) {
- addMergedDataFiles(skipped, snapshot);
- }
- return entry -> containsDataFile(skipped, entry);
- }
-}
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 cb4349a59e..459713687e 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
@@ -769,12 +769,7 @@ abstract class AbstractFileStoreTable implements
FileStoreTable {
@Override
public void deleteTag(String tagName) {
- tagManager()
- .deleteTag(
- tagName,
- store().newTagDeletion(),
- snapshotManager(),
- store().createTagCallbacks(this));
+ tagManager().deleteTag(tagName, store().createTagCallbacks(this));
}
@Override
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/Table.java
b/paimon-core/src/main/java/org/apache/paimon/table/Table.java
index b4fe2bc237..846054b9f4 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/Table.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/Table.java
@@ -143,7 +143,7 @@ public interface Table extends Serializable {
@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);
diff --git
a/paimon-core/src/main/java/org/apache/paimon/tag/BatchReadTagCreator.java
b/paimon-core/src/main/java/org/apache/paimon/tag/BatchReadTagCreator.java
index 8637709b31..04b7e1b036 100644
--- a/paimon-core/src/main/java/org/apache/paimon/tag/BatchReadTagCreator.java
+++ b/paimon-core/src/main/java/org/apache/paimon/tag/BatchReadTagCreator.java
@@ -78,10 +78,8 @@ public class BatchReadTagCreator {
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);
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/tag/TagAutoCreation.java
b/paimon-core/src/main/java/org/apache/paimon/tag/TagAutoCreation.java
index 47a186cbd4..b6dd51f360 100644
--- a/paimon-core/src/main/java/org/apache/paimon/tag/TagAutoCreation.java
+++ b/paimon-core/src/main/java/org/apache/paimon/tag/TagAutoCreation.java
@@ -21,7 +21,6 @@ package org.apache.paimon.tag;
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 @@ public class 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 @@ public class TagAutoCreation {
"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 class TagAutoCreation {
CoreOptions options,
SnapshotManager snapshotManager,
TagManager tagManager,
- TagDeletion tagDeletion,
List<TagCallback> callbacks) {
TagTimeExtractor extractor =
TagTimeExtractor.createForAutoTag(options);
if (extractor == null) {
@@ -234,7 +229,6 @@ public class TagAutoCreation {
return new TagAutoCreation(
snapshotManager,
tagManager,
- tagDeletion,
extractor,
TagPeriodHandler.create(options),
options.tagCreationDelay(),
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 31ca385d1e..d649c7b99c 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
@@ -19,7 +19,6 @@
package org.apache.paimon.tag;
import org.apache.paimon.CoreOptions;
-import org.apache.paimon.operation.TagDeletion;
import org.apache.paimon.table.sink.TagCallback;
import org.apache.paimon.utils.SnapshotManager;
import org.apache.paimon.utils.TagManager;
@@ -53,7 +52,6 @@ public class TagAutoManager {
CoreOptions options,
SnapshotManager snapshotManager,
TagManager tagManager,
- TagDeletion tagDeletion,
List<TagCallback> callbacks) {
TagTimeExtractor extractor =
TagTimeExtractor.createForAutoTag(options);
@@ -61,10 +59,9 @@ public class TagAutoManager {
return new TagAutoManager(
extractor == null
? null
- : TagAutoCreation.create(
- options, snapshotManager, tagManager,
tagDeletion, callbacks),
+ : TagAutoCreation.create(options, snapshotManager,
tagManager, callbacks),
options.tagTimeExpireEnabled()
- ? TagTimeExpire.create(snapshotManager, tagManager,
tagDeletion, callbacks)
+ ? TagTimeExpire.create(snapshotManager, tagManager,
callbacks)
: null);
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/tag/TagBatchCreation.java
b/paimon-core/src/main/java/org/apache/paimon/tag/TagBatchCreation.java
index 2b637db1cb..d049129311 100644
--- a/paimon-core/src/main/java/org/apache/paimon/tag/TagBatchCreation.java
+++ b/paimon-core/src/main/java/org/apache/paimon/tag/TagBatchCreation.java
@@ -20,7 +20,6 @@ package org.apache.paimon.tag;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.Snapshot;
-import org.apache.paimon.operation.TagDeletion;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.utils.SnapshotManager;
import org.apache.paimon.utils.TagManager;
@@ -44,14 +43,12 @@ public class TagBatchCreation {
private final CoreOptions options;
private final TagManager tagManager;
private final SnapshotManager snapshotManager;
- private final TagDeletion tagDeletion;
private final ZoneId zoneId;
public TagBatchCreation(FileStoreTable table) {
this.table = table;
this.snapshotManager = table.snapshotManager();
this.tagManager = table.tagManager();
- this.tagDeletion = table.store().newTagDeletion();
this.options = table.coreOptions();
this.zoneId = options.sinkProcessTimeZone();
}
@@ -70,8 +67,7 @@ public class TagBatchCreation {
+
localDateTime.format(DateTimeFormatter.ofPattern("yyyy-MM-dd"));
try {
// If the tag already exists, delete the tag
- tagManager.deleteTag(
- tagName, tagDeletion, snapshotManager,
table.store().createTagCallbacks(table));
+ tagManager.deleteTag(tagName,
table.store().createTagCallbacks(table));
// Create a new tag
tagManager.createTag(
snapshot,
@@ -86,11 +82,7 @@ public class TagBatchCreation {
snapshot.id(),
e);
if (tagManager.tagExists(tagName)) {
- tagManager.deleteTag(
- tagName,
- tagDeletion,
- snapshotManager,
- table.store().createTagCallbacks(table));
+ tagManager.deleteTag(tagName,
table.store().createTagCallbacks(table));
}
}
// Expire the tag
@@ -108,17 +100,13 @@ public class TagBatchCreation {
while (tagCount > tagNumRetainedMax) {
for (List<String> tagNames : tagManager.tags().values()) {
if (tagCount - tagNames.size() > tagNumRetainedMax) {
- tagManager.deleteAllTagsOfOneSnapshot(
- tagNames, tagDeletion, snapshotManager);
+ tagManager.deleteAllTagsOfOneSnapshot(tagNames);
tagCount = tagCount - tagNames.size();
} else {
List<String> sortedTagNames =
tagManager.sortTagsOfOneSnapshot(tagNames);
for (String toBeDeleted : sortedTagNames) {
tagManager.deleteTag(
- toBeDeleted,
- tagDeletion,
- snapshotManager,
- table.store().createTagCallbacks(table));
+ toBeDeleted,
table.store().createTagCallbacks(table));
tagCount--;
if (tagCount == tagNumRetainedMax) {
break;
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 0b57dfa884..8c6d8958a6 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
@@ -19,7 +19,6 @@
package org.apache.paimon.tag;
import org.apache.paimon.fs.FileStatus;
-import org.apache.paimon.operation.TagDeletion;
import org.apache.paimon.table.sink.TagCallback;
import org.apache.paimon.utils.DateTimeUtils;
import org.apache.paimon.utils.Pair;
@@ -42,19 +41,14 @@ public class TagTimeExpire {
private final SnapshotManager snapshotManager;
private final TagManager tagManager;
- private final TagDeletion tagDeletion;
private final List<TagCallback> callbacks;
private LocalDateTime olderThanTime;
private TagTimeExpire(
- SnapshotManager snapshotManager,
- TagManager tagManager,
- TagDeletion tagDeletion,
- List<TagCallback> callbacks) {
+ SnapshotManager snapshotManager, TagManager tagManager,
List<TagCallback> callbacks) {
this.snapshotManager = snapshotManager;
this.tagManager = tagManager;
- this.tagDeletion = tagDeletion;
this.callbacks = callbacks;
}
@@ -95,7 +89,7 @@ public class TagTimeExpire {
timeRetained,
createTime,
olderThanTime);
- tagManager.deleteTag(tagName, tagDeletion, snapshotManager,
callbacks);
+ tagManager.deleteTag(tagName, callbacks);
expired.add(tagName);
}
}
@@ -108,10 +102,7 @@ public class TagTimeExpire {
}
public static TagTimeExpire create(
- SnapshotManager snapshotManager,
- TagManager tagManager,
- TagDeletion tagDeletion,
- List<TagCallback> callbacks) {
- return new TagTimeExpire(snapshotManager, tagManager, tagDeletion,
callbacks);
+ SnapshotManager snapshotManager, TagManager tagManager,
List<TagCallback> callbacks) {
+ return new TagTimeExpire(snapshotManager, tagManager, callbacks);
}
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/utils/TagManager.java
b/paimon-core/src/main/java/org/apache/paimon/utils/TagManager.java
index 8bbcf55791..040046c09a 100644
--- a/paimon-core/src/main/java/org/apache/paimon/utils/TagManager.java
+++ b/paimon-core/src/main/java/org/apache/paimon/utils/TagManager.java
@@ -25,8 +25,6 @@ import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.FileStatus;
import org.apache.paimon.fs.Path;
import org.apache.paimon.iceberg.IcebergCommitCallback;
-import org.apache.paimon.manifest.ExpireFileEntry;
-import org.apache.paimon.operation.TagDeletion;
import org.apache.paimon.table.sink.TagCallback;
import org.apache.paimon.tag.Tag;
import org.apache.paimon.tag.TagPeriodHandler;
@@ -46,7 +44,6 @@ import java.util.Collection;
import java.util.Comparator;
import java.util.List;
import java.util.Optional;
-import java.util.Set;
import java.util.SortedMap;
import java.util.TreeMap;
import java.util.function.Predicate;
@@ -210,31 +207,19 @@ public class TagManager {
}
}
- /** Make sure the tagNames are ALL tags of one snapshot. */
- public void deleteAllTagsOfOneSnapshot(
- List<String> tagNames, TagDeletion tagDeletion, SnapshotManager
snapshotManager) {
- Snapshot taggedSnapshot = getOrThrow(tagNames.get(0)).trimToSnapshot();
- List<Snapshot> taggedSnapshots;
-
- // skip file deletion if snapshot exists
- if (snapshotManager.snapshotExists(taggedSnapshot.id())) {
- tagNames.forEach(tagName ->
fileIO.deleteQuietly(tagPath(tagName)));
- return;
- } else {
- // FileIO discovers tags by tag file, so we should read all tags
before we delete tag
- taggedSnapshots = taggedSnapshots();
- tagNames.forEach(tagName ->
fileIO.deleteQuietly(tagPath(tagName)));
- }
-
- doClean(taggedSnapshot, taggedSnapshots, snapshotManager, tagDeletion);
+ /**
+ * Delete only tag files, leaving referenced files to orphan file cleanup.
Make sure the
+ * tagNames are ALL tags of one snapshot.
+ */
+ public void deleteAllTagsOfOneSnapshot(List<String> tagNames) {
+ tagNames.forEach(tagName -> fileIO.deleteQuietly(tagPath(tagName)));
}
- /** Ignore errors if the tag doesn't exist. */
- public void deleteTag(
- String tagName,
- TagDeletion tagDeletion,
- SnapshotManager snapshotManager,
- List<TagCallback> callbacks) {
+ /**
+ * Delete only the tag file, leaving referenced files to orphan file
cleanup. Ignore errors if
+ * the tag doesn't exist.
+ */
+ public void deleteTag(String tagName, List<TagCallback> callbacks) {
checkArgument(!StringUtils.isNullOrWhitespaceOnly(tagName), "Tag name
shouldn't be blank.");
Optional<Tag> tag = get(tagName);
if (!tag.isPresent()) {
@@ -242,25 +227,7 @@ public class TagManager {
return;
}
- Snapshot taggedSnapshot = tag.get().trimToSnapshot();
- List<Snapshot> taggedSnapshots;
-
- // skip file deletion if snapshot exists
- if
(snapshotManager.copyWithBranch(branch).snapshotExists(taggedSnapshot.id())) {
- deleteTagMetaFile(tagName, callbacks);
- return;
- } else {
- // FileIO discovers tags by tag file, so we should read all tags
before we delete tag
- SortedMap<Snapshot, List<String>> tags = tags();
- deleteTagMetaFile(tagName, callbacks);
- // skip data file clean if more than 1 tags are created based on
this snapshot
- if (tags.get(taggedSnapshot).size() > 1) {
- return;
- }
- taggedSnapshots = new ArrayList<>(tags.keySet());
- }
-
- doClean(taggedSnapshot, taggedSnapshots, snapshotManager, tagDeletion);
+ deleteTagMetaFile(tagName, callbacks);
}
private void deleteTagMetaFile(String tagName, List<TagCallback>
callbacks) {
@@ -274,64 +241,6 @@ public class TagManager {
}
}
- private void doClean(
- Snapshot taggedSnapshot,
- List<Snapshot> taggedSnapshots,
- SnapshotManager snapshotManager,
- TagDeletion tagDeletion) {
- // collect skipping sets from the left neighbor tag and the nearest
right neighbor (either
- // the earliest snapshot or right neighbor tag)
- List<Snapshot> skippedSnapshots = new ArrayList<>();
-
- int index = findIndex(taggedSnapshot, taggedSnapshots);
- // the left neighbor tag
- if (index - 1 >= 0) {
- skippedSnapshots.add(taggedSnapshots.get(index - 1));
- }
- // the nearest right neighbor
- Snapshot right =
snapshotManager.copyWithBranch(branch).earliestSnapshot();
- if (index + 1 < taggedSnapshots.size()) {
- Snapshot rightTag = taggedSnapshots.get(index + 1);
- right = right.id() < rightTag.id() ? right : rightTag;
- }
- skippedSnapshots.add(right);
-
- // delete data files and empty directories
- Predicate<ExpireFileEntry> dataFileSkipper = null;
- boolean success = true;
- try {
- dataFileSkipper = tagDeletion.dataFileSkipper(skippedSnapshots);
- } catch (Exception e) {
- LOG.info(
- String.format(
- "Skip cleaning data files for tag of snapshot %s
due to failed to build skipping set.",
- taggedSnapshot.id()),
- e);
- success = false;
- }
- if (success) {
- tagDeletion.cleanUnusedDataFiles(taggedSnapshot, dataFileSkipper);
- tagDeletion.cleanEmptyDirectories();
- }
-
- // delete manifests
- success = true;
- Set<String> manifestSkippingSet = null;
- try {
- manifestSkippingSet =
tagDeletion.manifestSkippingSet(skippedSnapshots);
- } catch (Exception e) {
- LOG.info(
- String.format(
- "Skip cleaning manifest files for tag of snapshot
%s due to failed to build skipping set.",
- taggedSnapshot.id()),
- e);
- success = false;
- }
- if (success) {
- tagDeletion.cleanUnusedManifests(taggedSnapshot,
manifestSkippingSet);
- }
- }
-
/** Check if a tag exists. */
public boolean tagExists(String tagName) {
Path path = tagPath(tagName);
@@ -467,18 +376,6 @@ public class TagManager {
return
tags().values().stream().flatMap(Collection::stream).collect(Collectors.toList());
}
- private int findIndex(Snapshot taggedSnapshot, List<Snapshot>
taggedSnapshots) {
- for (int i = 0; i < taggedSnapshots.size(); i++) {
- if (taggedSnapshot.id() == taggedSnapshots.get(i).id()) {
- return i;
- }
- }
- throw new RuntimeException(
- String.format(
- "Didn't find tag with snapshot id '%s'. This is
unexpected.",
- taggedSnapshot.id()));
- }
-
private void validateNoAutoTag(String tagName, Snapshot snapshot) {
if (tagPeriodHandler == null || !tagPeriodHandler.isAutoTag(tagName)) {
return;
diff --git
a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassignerTest.java
b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassignerTest.java
index b0ad384968..1b71781ed8 100644
---
a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassignerTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassignerTest.java
@@ -334,6 +334,9 @@ public class DataEvolutionRowIdReassignerTest extends
TableTestBase {
assertThat(table.fileIO().exists(plan)).isTrue();
table.deleteTag("reassign");
+ assertThat(table.tagManager().tagExists("reassign")).isFalse();
+ assertThat(table.fileIO().exists(plan)).isTrue();
+ new LocalOrphanFilesClean(table, System.currentTimeMillis() +
2000).clean();
assertThat(table.fileIO().exists(plan)).isFalse();
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/ExpireSnapshotsTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/ExpireSnapshotsTest.java
index c0a113a90e..446391fdfe 100644
---
a/paimon-core/src/test/java/org/apache/paimon/operation/ExpireSnapshotsTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/ExpireSnapshotsTest.java
@@ -257,11 +257,7 @@ public class ExpireSnapshotsTest {
// randomly delete tags
for (int id = 1; id <= latestSnapshotId; id++) {
if (random.nextBoolean()) {
- tagManager.deleteTag(
- "tag" + id,
- store.newTagDeletion(),
- snapshotManager,
- Collections.emptyList());
+ tagManager.deleteTag("tag" + id, Collections.emptyList());
}
}
@@ -480,7 +476,7 @@ public class ExpireSnapshotsTest {
}
@ParameterizedTest
- @ValueSource(strings = {"snapshot", "tag", "changelog"})
+ @ValueSource(strings = {"snapshot", "changelog"})
public void testCleanManifestExtraFilesFollowsManifestRetention(String
cleaner)
throws Exception {
ManifestFileMeta expired =
@@ -501,10 +497,8 @@ public class ExpireSnapshotsTest {
expiredSnapshot.deltaManifestList(),
deletion.manifestSkippingSet(skippingSnapshots));
} else {
- FileDeletionBase<Snapshot> deletion =
- "tag".equals(cleaner) ? store.newTagDeletion() :
snapshotDeletion;
- deletion.cleanUnusedManifests(
- expiredSnapshot,
deletion.manifestSkippingSet(skippingSnapshots));
+ snapshotDeletion.cleanUnusedManifests(
+ expiredSnapshot,
snapshotDeletion.manifestSkippingSet(skippingSnapshots));
}
assertThat(fileIO.exists(store.pathFactory().toManifestFilePath("expired"))).isFalse();
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/FileDeletionTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/FileDeletionTest.java
index 4443813d77..a67d16dfd6 100644
---
a/paimon-core/src/test/java/org/apache/paimon/operation/FileDeletionTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/FileDeletionTest.java
@@ -44,6 +44,8 @@ import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.stats.StatsFileHandler;
import org.apache.paimon.table.ExpireSnapshots;
import org.apache.paimon.table.ExpireSnapshotsImpl;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.FileStoreTableFactory;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.ChangelogManager;
import org.apache.paimon.utils.FileStorePathFactory;
@@ -73,6 +75,7 @@ import java.util.Map;
import java.util.Set;
import java.util.concurrent.ThreadLocalRandom;
import java.util.stream.Collectors;
+import java.util.stream.Stream;
import static org.apache.paimon.CoreOptions.SNAPSHOT_CLEAN_EMPTY_DIRECTORIES;
import static
org.apache.paimon.operation.FileStoreTestUtils.assertNFilesExists;
@@ -389,9 +392,14 @@ public class FileDeletionTest {
assertPathExists(fileIO,
pathFactory.toManifestListPath(tag1.deltaManifestList()));
}
- @Test
- public void testDeleteTagWithSnapshot() throws Exception {
- TestFileStore store =
createStore(TestKeyValueGenerator.GeneratorMode.NON_PARTITIONED, 3);
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testDeleteTagWithSnapshot(boolean deleteAllTags) throws
Exception {
+ TestFileStore store =
+ createStore(
+ TestKeyValueGenerator.GeneratorMode.NON_PARTITIONED,
+ 3,
+ CoreOptions.ChangelogProducer.NONE);
tagManager = new TagManager(fileIO, store.options().path());
SnapshotManager snapshotManager = store.snapshotManager();
TestKeyValueGenerator gen =
@@ -417,14 +425,14 @@ public class FileDeletionTest {
ManifestList manifestList = store.manifestListFactory().create();
Snapshot snapshot1 = snapshotManager.snapshot(1);
List<ManifestFileMeta> snapshot1Data =
manifestList.readDataManifests(snapshot1);
- Snapshot snapshot3 = snapshotManager.snapshot(3);
- List<ManifestFileMeta> snapshot3Data =
manifestList.readDataManifests(snapshot3);
-
List<String> manifestLists =
Arrays.asList(snapshot1.baseManifestList(),
snapshot1.deltaManifestList());
// create tag1
createTag(snapshot1, "tag1", store.options().tagDefaultTimeRetained());
+ if (deleteAllTags) {
+ createTag(snapshot1, "tag1-copy",
store.options().tagDefaultTimeRetained());
+ }
// expire snapshot 1, 2
store.newExpire(1, 1, Long.MAX_VALUE).expire();
@@ -441,26 +449,72 @@ public class FileDeletionTest {
assertPathExists(fileIO,
pathFactory.toManifestListPath(manifestListName));
}
- tagManager.deleteTag(
- "tag1", store.newTagDeletion(), snapshotManager,
Collections.emptyList());
+ Set<java.nio.file.Path> retainedFiles = tableFiles();
+ retainedFiles.remove(Paths.get(tagManager.tagPath("tag1").toString()));
+ if (deleteAllTags) {
+
retainedFiles.remove(Paths.get(tagManager.tagPath("tag1-copy").toString()));
+ tagManager.deleteAllTagsOfOneSnapshot(Arrays.asList("tag1",
"tag1-copy"));
+ assertThat(tagManager.tagExists("tag1-copy")).isFalse();
+ } else {
+ tagManager.deleteTag("tag1", Collections.emptyList());
+ }
+ assertThat(tagManager.tagExists("tag1")).isFalse();
+
assertThat(tableFiles()).containsExactlyInAnyOrderElementsOf(retainedFiles);
// check data files
- assertPathNotExists(fileIO, pathFactory.bucketPath(partition, 0));
+ assertPathExists(fileIO, pathFactory.bucketPath(partition, 0));
assertPathExists(fileIO, pathFactory.bucketPath(partition, 1));
assertPathExists(fileIO, pathFactory.bucketPath(partition, 2));
// check manifests
for (ManifestFileMeta manifestFileMeta : snapshot1Data) {
Path path =
pathFactory.toManifestFilePath(manifestFileMeta.fileName());
- if (snapshot3Data.contains(manifestFileMeta)) {
- assertPathExists(fileIO, path);
- } else {
- assertPathNotExists(fileIO, path);
- }
+ assertPathExists(fileIO, path);
+ }
+ for (String manifestListName : manifestLists) {
+ assertPathExists(fileIO,
pathFactory.toManifestListPath(manifestListName));
+ }
+
+ FileStoreTable table = FileStoreTableFactory.create(fileIO,
store.options().path());
+ assertThat(new
LocalOrphanFilesClean(table).clean().getDeletedFilesPath()).isEmpty();
+
assertThat(tableFiles()).containsExactlyInAnyOrderElementsOf(retainedFiles);
+ List<Path> orphanDataFiles =
+
Arrays.stream(fileIO.listStatus(pathFactory.bucketPath(partition, 0)))
+ .map(status -> status.getPath())
+ .filter(path -> path.getName().startsWith("data-"))
+ .collect(Collectors.toList());
+ assertThat(orphanDataFiles).isNotEmpty();
+
+ // Remove obsolete entries from the retained manifests so the old data
is unreferenced.
+ try (FileStoreCommitImpl commit = store.newCommit()) {
+ commit.compactManifest();
+ }
+ store.newExpire(1, 1, Long.MAX_VALUE).expire();
+ for (Path path : orphanDataFiles) {
+ assertPathExists(fileIO, path);
}
+
+ List<Path> deleted =
+ new LocalOrphanFilesClean(table, System.currentTimeMillis() +
2000)
+ .clean()
+ .getDeletedFilesPath();
+ assertThat(deleted).containsAll(orphanDataFiles);
+ assertThat(deleted.stream().map(path -> path.toUri().getPath()))
+ .contains(
+ pathFactory
+
.toManifestListPath(snapshot1.deltaManifestList())
+ .toUri()
+ .getPath());
+ assertPathNotExists(fileIO, pathFactory.bucketPath(partition, 0));
+ assertPathExists(fileIO, pathFactory.bucketPath(partition, 1));
+ assertPathExists(fileIO, pathFactory.bucketPath(partition, 2));
for (String manifestListName : manifestLists) {
assertPathNotExists(fileIO,
pathFactory.toManifestListPath(manifestListName));
}
+ for (ManifestFileMeta meta :
+
manifestList.readDataManifests(snapshotManager.latestSnapshot())) {
+ assertPathExists(fileIO,
pathFactory.toManifestFilePath(meta.fileName()));
+ }
}
@Test
@@ -518,29 +572,21 @@ public class FileDeletionTest {
assertPathExists(fileIO,
pathFactory.toManifestListPath(manifestListName));
}
- tagManager.deleteTag(
- "tag2", store.newTagDeletion(), snapshotManager,
Collections.emptyList());
+ tagManager.deleteTag("tag2", Collections.emptyList());
+ assertThat(tagManager.tagExists("tag2")).isFalse();
// check data files
assertPathExists(fileIO, pathFactory.bucketPath(partition, 0));
- assertPathNotExists(fileIO, pathFactory.bucketPath(partition, 1));
+ assertPathExists(fileIO, pathFactory.bucketPath(partition, 1));
// check manifests
- Snapshot tag1 = tagManager.getOrThrow("tag1");
- Snapshot tag3 = tagManager.getOrThrow("tag3");
- List<ManifestFileMeta> existing = manifestList.readDataManifests(tag1);
- existing.addAll(manifestList.readDataManifests(tag3));
for (ManifestFileMeta manifestFileMeta : snapshot2Data) {
Path path =
pathFactory.toManifestFilePath(manifestFileMeta.fileName());
- if (existing.contains(manifestFileMeta)) {
- assertPathExists(fileIO, path);
- } else {
- assertPathNotExists(fileIO, path);
- }
+ assertPathExists(fileIO, path);
}
for (String manifestListName : manifestLists) {
- assertPathNotExists(fileIO,
pathFactory.toManifestListPath(manifestListName));
+ assertPathExists(fileIO,
pathFactory.toManifestListPath(manifestListName));
}
}
@@ -696,8 +742,7 @@ public class FileDeletionTest {
.snapshotTimeRetain(Duration.ofMillis(Long.MAX_VALUE))
.build())
.expire();
- tagManager.deleteTag(
- "tag1", store.newTagDeletion(), snapshotManager,
Collections.emptyList());
+ tagManager.deleteTag("tag1", Collections.emptyList());
expireSnapshots
.config(
ExpireConfig.builder()
@@ -851,17 +896,27 @@ public class FileDeletionTest {
return createStore(mode, 2);
}
+ private Set<java.nio.file.Path> tableFiles() throws IOException {
+ try (Stream<java.nio.file.Path> files = Files.walk(tempDir)) {
+ return
files.filter(Files::isRegularFile).collect(Collectors.toSet());
+ }
+ }
+
private TestFileStore createStore(TestKeyValueGenerator.GeneratorMode
mode, int buckets)
throws Exception {
- ThreadLocalRandom random = ThreadLocalRandom.current();
-
- CoreOptions.ChangelogProducer changelogProducer;
- if (random.nextBoolean()) {
- changelogProducer = CoreOptions.ChangelogProducer.INPUT;
- } else {
- changelogProducer = CoreOptions.ChangelogProducer.NONE;
- }
+ return createStore(
+ mode,
+ buckets,
+ ThreadLocalRandom.current().nextBoolean()
+ ? CoreOptions.ChangelogProducer.INPUT
+ : CoreOptions.ChangelogProducer.NONE);
+ }
+ private TestFileStore createStore(
+ TestKeyValueGenerator.GeneratorMode mode,
+ int buckets,
+ CoreOptions.ChangelogProducer changelogProducer)
+ throws Exception {
RowType rowType, partitionType;
switch (mode) {
case NON_PARTITIONED:
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 49e22ee551..6659205bc3 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
@@ -410,11 +410,15 @@ public class LocalOrphanFilesCleanTest {
Map<Long, List<TestPojo>> snapshotData,
Map<Long, List<InternalRow>> changelogData)
throws Exception {
+ // Deleted tags can leave additional orphan files beyond the manually
added ones.
assertThat(deleteFiles.stream().map(p -> p.toUri().getPath()))
- .containsExactlyInAnyOrderElementsOf(
+ .containsAll(
manuallyAddedFiles.stream()
.map(p -> p.toUri().getPath())
.collect(Collectors.toList()));
+ for (Path path : deleteFiles) {
+ assertThat(fileIO.exists(path)).as("Deleted orphan file %s",
path).isFalse();
+ }
Set<Snapshot> snapshots = new HashSet<>();
table.snapshotManager().snapshots().forEachRemaining(snapshots::add);
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/IndexFileExpireTableTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/IndexFileExpireTableTest.java
index 7293e8f8be..b6f66316fe 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/IndexFileExpireTableTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/IndexFileExpireTableTest.java
@@ -130,14 +130,19 @@ public class IndexFileExpireTableTest extends
PrimaryKeyTableTestBase {
// test delete tag after expiring snapshots
table.createTag("tag3", 3);
+ Snapshot taggedSnapshot = table.tagManager().getOrThrow("tag3");
expire.expireUntil(1, 7);
+ indexFileSize = indexFileSize();
+ indexManifestSize = indexManifestSize();
table.deleteTag("tag3");
+ assertThat(table.tagManager().tagExists("tag3")).isFalse();
TagManager tagManager = new TagManager(LocalFileIO.create(),
table.location());
+ checkIndexFiles(taggedSnapshot);
checkIndexFiles(7);
checkIndexFiles(tagManager.getOrThrow("tag5"));
- assertThat(indexFileSize()).isEqualTo(4);
- assertThat(indexManifestSize()).isEqualTo(2);
+ assertThat(indexFileSize()).isEqualTo(indexFileSize);
+ assertThat(indexManifestSize()).isEqualTo(indexManifestSize);
}
@Test
diff --git
a/paimon-core/src/test/java/org/apache/paimon/tag/TagAutoManagerTest.java
b/paimon-core/src/test/java/org/apache/paimon/tag/TagAutoManagerTest.java
index d7951fd584..fb01f2d557 100644
--- a/paimon-core/src/test/java/org/apache/paimon/tag/TagAutoManagerTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/tag/TagAutoManagerTest.java
@@ -546,7 +546,6 @@ public class TagAutoManagerTest extends
PrimaryKeyTableTestBase {
table.coreOptions(),
table.snapshotManager(),
table.store().newTagManager(),
- table.store().newTagDeletion(),
Collections.emptyList());
// threshold = tagTime + 2 days (nextTag + 1 period)
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/RollbackToAsLatestProcedure.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/RollbackToAsLatestProcedure.java
index 363a55b190..e41e822a86 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/RollbackToAsLatestProcedure.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/RollbackToAsLatestProcedure.java
@@ -119,11 +119,7 @@ public class RollbackToAsLatestProcedure extends
ProcedureBase {
} catch (Exception e) {
try {
if (createdRollbackTag != null && canDeleteCreatedTag) {
- tagManager.deleteTag(
- createdRollbackTag,
- store.newTagDeletion(),
- store.snapshotManager(),
- Collections.emptyList());
+ tagManager.deleteTag(createdRollbackTag,
Collections.emptyList());
}
} catch (Exception cleanupException) {
e.addSuppressed(cleanupException);
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AutoTagForSavepointCommitterOperator.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AutoTagForSavepointCommitterOperator.java
index 48f8d65e2e..523946d877 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AutoTagForSavepointCommitterOperator.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AutoTagForSavepointCommitterOperator.java
@@ -19,7 +19,6 @@
package org.apache.paimon.flink.sink;
import org.apache.paimon.Snapshot;
-import org.apache.paimon.operation.TagDeletion;
import org.apache.paimon.table.sink.TagCallback;
import org.apache.paimon.utils.SerializableSupplier;
import org.apache.paimon.utils.SnapshotManager;
@@ -63,8 +62,6 @@ public class AutoTagForSavepointCommitterOperator<CommitT,
GlobalCommitT>
private final SerializableSupplier<TagManager> tagManagerFactory;
- private final SerializableSupplier<TagDeletion> tagDeletionFactory;
-
private final SerializableSupplier<List<TagCallback>> callbacksSupplier;
private final NavigableSet<Long> identifiersForTags;
@@ -75,8 +72,6 @@ public class AutoTagForSavepointCommitterOperator<CommitT,
GlobalCommitT>
private transient TagManager tagManager;
- private transient TagDeletion tagDeletion;
-
private transient List<TagCallback> callbacks;
private transient ListState<Long> identifiersForTagsState;
@@ -85,13 +80,11 @@ public class AutoTagForSavepointCommitterOperator<CommitT,
GlobalCommitT>
CommitterOperator<CommitT, GlobalCommitT> commitOperator,
SerializableSupplier<SnapshotManager> snapshotManagerFactory,
SerializableSupplier<TagManager> tagManagerFactory,
- SerializableSupplier<TagDeletion> tagDeletionFactory,
SerializableSupplier<List<TagCallback>> callbacksSupplier,
Duration tagTimeRetained) {
this.commitOperator = commitOperator;
this.tagManagerFactory = tagManagerFactory;
this.snapshotManagerFactory = snapshotManagerFactory;
- this.tagDeletionFactory = tagDeletionFactory;
this.callbacksSupplier = callbacksSupplier;
this.identifiersForTags = new TreeSet<>();
this.tagTimeRetained = tagTimeRetained;
@@ -105,7 +98,6 @@ public class AutoTagForSavepointCommitterOperator<CommitT,
GlobalCommitT>
} finally {
snapshotManager = snapshotManagerFactory.get();
tagManager = tagManagerFactory.get();
- tagDeletion = tagDeletionFactory.get();
callbacks = callbacksSupplier.get();
identifiersForTagsState =
@@ -152,12 +144,7 @@ public class AutoTagForSavepointCommitterOperator<CommitT,
GlobalCommitT>
commitOperator.notifyCheckpointAborted(checkpointId);
identifiersForTags.remove(checkpointId);
SavepointTagUtils.deleteTagIfMatches(
- tagManager,
- commitOperator.getCommitUser(),
- checkpointId,
- tagDeletion,
- snapshotManager,
- callbacks);
+ tagManager, commitOperator.getCommitUser(), checkpointId,
callbacks);
}
private void createTagForIdentifiers(List<Long> identifiers) {
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AutoTagForSavepointCommitterOperatorFactory.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AutoTagForSavepointCommitterOperatorFactory.java
index 1787f8e7ad..4c8f19a49c 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AutoTagForSavepointCommitterOperatorFactory.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AutoTagForSavepointCommitterOperatorFactory.java
@@ -18,7 +18,6 @@
package org.apache.paimon.flink.sink;
-import org.apache.paimon.operation.TagDeletion;
import org.apache.paimon.table.sink.TagCallback;
import org.apache.paimon.utils.SerializableSupplier;
import org.apache.paimon.utils.SnapshotManager;
@@ -48,8 +47,6 @@ public class
AutoTagForSavepointCommitterOperatorFactory<CommitT, GlobalCommitT>
private final SerializableSupplier<TagManager> tagManagerFactory;
- private final SerializableSupplier<TagDeletion> tagDeletionFactory;
-
private final SerializableSupplier<List<TagCallback>> callbacksSupplier;
private final NavigableSet<Long> identifiersForTags;
@@ -60,13 +57,11 @@ public class
AutoTagForSavepointCommitterOperatorFactory<CommitT, GlobalCommitT>
CommitterOperatorFactory<CommitT, GlobalCommitT>
commitOperatorFactory,
SerializableSupplier<SnapshotManager> snapshotManagerFactory,
SerializableSupplier<TagManager> tagManagerFactory,
- SerializableSupplier<TagDeletion> tagDeletionFactory,
SerializableSupplier<List<TagCallback>> callbacksSupplier,
Duration tagTimeRetained) {
this.commitOperatorFactory = commitOperatorFactory;
this.tagManagerFactory = tagManagerFactory;
this.snapshotManagerFactory = snapshotManagerFactory;
- this.tagDeletionFactory = tagDeletionFactory;
this.callbacksSupplier = callbacksSupplier;
this.identifiersForTags = new TreeSet<>();
this.tagTimeRetained = tagTimeRetained;
@@ -81,7 +76,6 @@ public class
AutoTagForSavepointCommitterOperatorFactory<CommitT, GlobalCommitT>
commitOperatorFactory.createStreamOperator(parameters),
snapshotManagerFactory,
tagManagerFactory,
- tagDeletionFactory,
callbacksSupplier,
tagTimeRetained);
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java
index 21e9b722c4..a1fd0cd7aa 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java
@@ -260,7 +260,6 @@ public abstract class FlinkSink<T> implements Serializable {
committerOperator,
table::snapshotManager,
table::tagManager,
- () -> table.store().newTagDeletion(),
() -> table.store().createTagCallbacks(table),
table.coreOptions().tagDefaultTimeRetained());
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/RowAppendTableSink.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/RowAppendTableSink.java
index 2e6b54fef1..87a28bb10b 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/RowAppendTableSink.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/RowAppendTableSink.java
@@ -23,7 +23,6 @@ import org.apache.paimon.flink.FlinkConnectorOptions;
import
org.apache.paimon.flink.sink.coordinator.CommittingWriteOperatorCoordinator;
import org.apache.paimon.flink.sink.coordinator.SavepointTagger;
import org.apache.paimon.manifest.ManifestCommittable;
-import org.apache.paimon.operation.TagDeletion;
import org.apache.paimon.options.Options;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.sink.TagCallback;
@@ -143,8 +142,6 @@ public class RowAppendTableSink extends
AppendTableSink<InternalRow> {
private static SavepointTagger.Factory
createSavepointTaggerFactory(FileStoreTable table) {
SerializableSupplier<SnapshotManager> snapshotManagerFactory =
table::snapshotManager;
SerializableSupplier<TagManager> tagManagerFactory =
table::tagManager;
- SerializableSupplier<TagDeletion> tagDeletionFactory =
- () -> table.store().newTagDeletion();
SerializableSupplier<List<TagCallback>> callbacksSupplier =
() -> table.store().createTagCallbacks(table);
Duration tagTimeRetained =
table.coreOptions().tagDefaultTimeRetained();
@@ -152,7 +149,6 @@ public class RowAppendTableSink extends
AppendTableSink<InternalRow> {
new SavepointTagger(
snapshotManagerFactory.get(),
tagManagerFactory.get(),
- tagDeletionFactory.get(),
callbacksSupplier.get(),
tagTimeRetained,
commitUser);
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/SavepointTagUtils.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/SavepointTagUtils.java
index b530cd864c..8905ad6ba6 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/SavepointTagUtils.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/SavepointTagUtils.java
@@ -18,10 +18,8 @@
package org.apache.paimon.flink.sink;
-import org.apache.paimon.operation.TagDeletion;
import org.apache.paimon.table.sink.TagCallback;
import org.apache.paimon.tag.Tag;
-import org.apache.paimon.utils.SnapshotManager;
import org.apache.paimon.utils.TagManager;
import java.util.List;
@@ -52,16 +50,11 @@ public class SavepointTagUtils {
TagManager tagManager,
String commitUser,
long commitIdentifier,
- TagDeletion tagDeletion,
- SnapshotManager snapshotManager,
List<TagCallback> callbacks) {
String tagName = tagNameOf(commitIdentifier);
tagManager
.get(tagName)
.filter(tag -> isSavepointTagFor(tag, commitUser,
commitIdentifier))
- .ifPresent(
- ignored ->
- tagManager.deleteTag(
- tagName, tagDeletion, snapshotManager,
callbacks));
+ .ifPresent(ignored -> tagManager.deleteTag(tagName,
callbacks));
}
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/SavepointTagger.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/SavepointTagger.java
index e762f55e01..882c96eddf 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/SavepointTagger.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/SavepointTagger.java
@@ -20,7 +20,6 @@ package org.apache.paimon.flink.sink.coordinator;
import org.apache.paimon.Snapshot;
import org.apache.paimon.flink.sink.SavepointTagUtils;
-import org.apache.paimon.operation.TagDeletion;
import org.apache.paimon.table.sink.TagCallback;
import org.apache.paimon.utils.SnapshotManager;
import org.apache.paimon.utils.TagManager;
@@ -47,7 +46,6 @@ public class SavepointTagger {
private final SnapshotManager snapshotManager;
private final TagManager tagManager;
- private final TagDeletion tagDeletion;
private final List<TagCallback> callbacks;
private final Duration tagTimeRetained;
// findSnapshotsForIdentifiers filters by commit user, so the tagger must
be bound to the user
@@ -59,13 +57,11 @@ public class SavepointTagger {
public SavepointTagger(
SnapshotManager snapshotManager,
TagManager tagManager,
- TagDeletion tagDeletion,
List<TagCallback> callbacks,
Duration tagTimeRetained,
String commitUser) {
this.snapshotManager = checkNotNull(snapshotManager);
this.tagManager = checkNotNull(tagManager);
- this.tagDeletion = checkNotNull(tagDeletion);
this.callbacks = checkNotNull(callbacks);
this.tagTimeRetained = tagTimeRetained;
this.commitUser = checkNotNull(commitUser);
@@ -90,8 +86,7 @@ public class SavepointTagger {
/** Drops an aborted savepoint's pending intent and removes any tag
already created for it. */
public void dropAborted(long checkpointId) {
pendingIdentifiers.remove(checkpointId);
- SavepointTagUtils.deleteTagIfMatches(
- tagManager, commitUser, checkpointId, tagDeletion,
snapshotManager, callbacks);
+ SavepointTagUtils.deleteTagIfMatches(tagManager, commitUser,
checkpointId, callbacks);
}
private void createTags(Collection<Long> identifiers) {
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/AutoTagForSavepointCommitterOperatorTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/AutoTagForSavepointCommitterOperatorTest.java
index 1ebddec2bf..91e1bf3aa0 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/AutoTagForSavepointCommitterOperatorTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/AutoTagForSavepointCommitterOperatorTest.java
@@ -239,7 +239,6 @@ public class AutoTagForSavepointCommitterOperatorTest
extends CommitterOperatorT
table, commitUser, committableStateManager),
table::snapshotManager,
table::tagManager,
- () -> table.store().newTagDeletion(),
() -> table.store().createTagCallbacks(table),
table.store().options().tagDefaultTimeRetained());
}
@@ -257,7 +256,6 @@ public class AutoTagForSavepointCommitterOperatorTest
extends CommitterOperatorT
table, commitUser, committableStateManager,
initializeFunction),
table::snapshotManager,
table::tagManager,
- () -> table.store().newTagDeletion(),
() -> table.store().createTagCallbacks(table),
table.store().options().tagDefaultTimeRetained());
}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/SavepointTagUtilsTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/SavepointTagUtilsTest.java
index 2e858e41c8..b513dec752 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/SavepointTagUtilsTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/SavepointTagUtilsTest.java
@@ -125,8 +125,6 @@ public class SavepointTagUtilsTest extends
CommitterTestBase {
table.tagManager(),
commitUser,
commitIdentifier,
- table.store().newTagDeletion(),
- table.snapshotManager(),
table.store().createTagCallbacks(table));
}
}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/CommittingWriteOperatorCoordinatorTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/CommittingWriteOperatorCoordinatorTest.java
index 57ac6b66aa..6fec03d7b3 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/CommittingWriteOperatorCoordinatorTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/CommittingWriteOperatorCoordinatorTest.java
@@ -1259,7 +1259,6 @@ public class CommittingWriteOperatorCoordinatorTest
extends CommitterTestBase {
new SavepointTagger(
table.snapshotManager(),
table.tagManager(),
- table.store().newTagDeletion(),
table.store().createTagCallbacks(table),
table.coreOptions().tagDefaultTimeRetained(),
user));
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/SavepointTaggerTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/SavepointTaggerTest.java
index a84bc205c9..b99599a9a9 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/SavepointTaggerTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/SavepointTaggerTest.java
@@ -191,7 +191,6 @@ public class SavepointTaggerTest extends CommitterTestBase {
return new SavepointTagger(
table.snapshotManager(),
table.tagManager(),
- table.store().newTagDeletion(),
table.store().createTagCallbacks(table),
table.coreOptions().tagDefaultTimeRetained(),
commitUser);
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/RollbackToAsLatestProcedure.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/RollbackToAsLatestProcedure.java
index 69366598ee..43705c1e12 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/RollbackToAsLatestProcedure.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/RollbackToAsLatestProcedure.java
@@ -151,11 +151,7 @@ public class RollbackToAsLatestProcedure extends
BaseProcedure {
} catch (Exception e) {
try {
if (createdRollbackTag != null &&
canDeleteCreatedTag) {
- tagManager.deleteTag(
- createdRollbackTag,
- store.newTagDeletion(),
- store.snapshotManager(),
- Collections.emptyList());
+ tagManager.deleteTag(createdRollbackTag,
Collections.emptyList());
}
} catch (Exception cleanupException) {
e.addSuppressed(cleanupException);
