[core] Optimize tag reads during snapshot expiration

paimon-core/src/main/java/org/apache/paimon/manifest/ManifestAvroReader.java

@@ -439,6 +439,18 @@ public RowIterator toRows(RowType projectedType) throws IOException {
             return toRows(projectedType, null, null, true);
         }
 
+        /**
+         * Lazily decompresses this block, applies manifest filters before decoding file metadata,
+         * and returns an iterator over one reusable row.
+         */
+        public RowIterator toRows(
+                RowType projectedType,
+                @Nullable PartitionPredicate partitionFilter,
+                @Nullable BucketFilter bucketFilter)
+                throws IOException {
+            return toRows(projectedType, partitionFilter, bucketFilter, true);
+        }
+
         private RowIterator toRows(
                 RowType projectedType,
                 @Nullable PartitionPredicate partitionFilter,

paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFile.java

@@ -48,6 +48,7 @@
 import java.util.Arrays;
 import java.util.List;
 import java.util.function.Function;
+import java.util.function.Predicate;
 
 /**
  * This file includes several {@link ManifestEntry}s, representing the additional changes since last
@@ -303,17 +304,41 @@ public long suggestedFileSize() {
     }
 
     public List<ExpireFileEntry> readExpireFileEntries(String fileName) {
+        return readExpireFileEntries(fileName, null, entry -> true);
+    }
+
+    /**
+     * Reads only expiring entries accepted by the supplied filters.
+     *
+     * <p>The bucket filter is evaluated by the Avro reader before nested data-file metadata is
+     * decoded. The entry filter then runs on a reusable projected view, before an {@link
+     * ExpireFileEntry} is materialized. The entry filter must not retain its argument.
+     */
+    public List<ExpireFileEntry> readExpireFileEntries(
+            String fileName,
+            @Nullable BucketFilter bucketFilter,
+            Predicate<ProjectedManifestEntry> entryFilter) {
         List<ExpireFileEntry> result = new ArrayList<>();
-        try (CloseableIterator<ProjectedManifestEntry> entries =
-                scan(fileName, EXPIRE_FILE_PROJECTION)) {
-            while (entries.hasNext()) {
-                result.add(ExpireFileEntry.from(entries.next()));
+        ProjectedManifestEntry entry = EXPIRE_FILE_PROJECTION.createEntry();
+        try (ManifestAvroReader reader = scanAvroBlocks(fileName, null)) {
+            while (reader.hasNext()) {
+                ManifestAvroReader.RowIterator rows =
+                        reader.next()
+                                .toRows(EXPIRE_FILE_PROJECTION.projectedType(), null, bucketFilter);
+                while (rows.hasNext()) {
+                    entry.replace(rows.next());
+                    if (entryFilter.test(entry)) {
+                        result.add(ExpireFileEntry.from(entry));
+                    }
+                }
             }
         } catch (Exception e) {
             throw new RuntimeException(
                     String.format(
                             "Failed to scan expiring entries from manifest file '%s'.", fileName),
                     e);
+        } finally {
+            entry.clear();
         }
         return result;
     }

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

@@ -25,14 +25,17 @@
 import org.apache.paimon.index.IndexFileHandler;
 import org.apache.paimon.index.IndexFileMeta;
 import org.apache.paimon.io.DataFilePathFactory;
+import org.apache.paimon.manifest.BucketFilter;
 import org.apache.paimon.manifest.ExpireFileEntry;
 import org.apache.paimon.manifest.FileEntry;
 import org.apache.paimon.manifest.FileEntry.Identifier;
 import org.apache.paimon.manifest.FileKind;
 import org.apache.paimon.manifest.IndexManifestEntry;
+import org.apache.paimon.manifest.ManifestBucketFilter;
 import org.apache.paimon.manifest.ManifestFile;
 import org.apache.paimon.manifest.ManifestFileMeta;
 import org.apache.paimon.manifest.ManifestList;
+import org.apache.paimon.manifest.ProjectedManifestEntry;
 import org.apache.paimon.stats.StatsFileHandler;
 import org.apache.paimon.utils.DataFilePathFactories;
 import org.apache.paimon.utils.FileOperationThreadPool;
@@ -56,7 +59,9 @@
 import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.NavigableSet;
 import java.util.Set;
+import java.util.TreeSet;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutionException;
@@ -185,7 +190,7 @@ protected void recordDeletionBuckets(ExpireFileEntry entry) {
     }
 
     /** Plan data files referenced by DELETE entries in the snapshot's delta manifest list. */
-    public List<Path> planDeletedInDeltaManifest(T snapshot, Predicate<ExpireFileEntry> skipper) {
+    public DataFileDeletionPlan planDeletedInDeltaManifest(T snapshot) {
         String deltaManifestList = snapshot.deltaManifestList();
         // data file path -> (original manifest entry, extra file paths)
         Map<Path, Pair<ExpireFileEntry, List<Path>>> dataFileToDelete = new HashMap<>();
@@ -218,12 +223,22 @@ public List<Path> planDeletedInDeltaManifest(T snapshot, Predicate<ExpireFileEnt
         } catch (Exception e) {
             // cancel deletion if any exception occurs
             LOG.warn("Failed to read some manifest files. Cancel deletion.", e);
-            return Collections.emptyList();
+            return DataFileDeletionPlan.empty();
         }
 
+        return new DataFileDeletionPlan(dataFileToDelete);
+    }
+
+    /** Plan data files referenced by DELETE entries in the snapshot's delta manifest list. */
+    public List<Path> planDeletedInDeltaManifest(T snapshot, Predicate<ExpireFileEntry> skipper) {
+        return dataFilesToDelete(planDeletedInDeltaManifest(snapshot), skipper);
+    }
+
+    public List<Path> dataFilesToDelete(
+            DataFileDeletionPlan plan, Predicate<ExpireFileEntry> skipper) {
         // apply skipper
         List<Path> actualDataFileToDelete = new ArrayList<>();
-        dataFileToDelete.forEach(
+        plan.dataFileToDelete.forEach(
                 (path, pair) -> {
                     ExpireFileEntry entry = pair.getLeft();
                     // check whether we should skip the data file
@@ -406,6 +421,38 @@ public Predicate<ExpireFileEntry> createDataFileSkipperForTag(Snapshot tag) thro
         return entry -> containsDataFile(tagDataFiles, entry);
     }
 
+    /**
+     * Creates a tag skipper restricted to the files in the supplied deletion plans.
+     *
+     * <p>A tag may reference every data file in a large table. Building an index for all of them
+     * makes snapshot expiration use memory proportional to the table size (and, when tags are read
+     * concurrently, to the number of tags). Only files that are candidates for the current
+     * expiration batch can be deleted, so merge only matching tag entries instead.
+     */
+    public Predicate<ExpireFileEntry> createDataFileSkipperForTag(
+            Snapshot tag, Collection<DataFileDeletionPlan> plans) throws Exception {
+        Map<BinaryRow, Map<Integer, Set<String>>> candidates = new HashMap<>();
+        for (DataFileDeletionPlan plan : plans) {
+            for (Pair<ExpireFileEntry, List<Path>> pair : plan.dataFileToDelete.values()) {
+                addDataFile(candidates, pair.getLeft());
+            }
+        }
+        if (candidates.isEmpty()) {
+            return entry -> false;
+        }
+
+        Collection<ExpireFileEntry> matchingEntries =
+                readMergedDataFiles(
+                        manifestList.readDataManifests(tag),
+                        createCandidateBucketFilter(candidates),
+                        entry -> containsDataFile(candidates, entry));
+        Map<BinaryRow, Map<Integer, Set<String>>> taggedCandidates = new HashMap<>();
+        for (ExpireFileEntry entry : matchingEntries) {
+            addDataFile(taggedCandidates, entry);
+        }
+        return entry -> containsDataFile(taggedCandidates, entry);
+    }
+
     /**
      * It is possible that a job was killed during expiration and some manifest files have been
      * deleted, so if the clean methods need to get manifests of a snapshot to be cleaned, we should
@@ -430,10 +477,7 @@ protected void addMergedDataFiles(
             throws IOException {
         for (ExpireFileEntry entry :
                 readMergedDataFiles(manifestList.readDataManifests(snapshot))) {
-            dataFiles
-                    .computeIfAbsent(entry.partition(), p -> new HashMap<>())
-                    .computeIfAbsent(entry.bucket(), b -> new HashSet<>())
-                    .add(entry.fileName());
+            addDataFile(dataFiles, entry);
         }
     }
 
@@ -444,13 +488,79 @@ protected Collection<ExpireFileEntry> readMergedDataFiles(List<ManifestFileMeta>
         return map.values();
     }
 
+    protected Collection<ExpireFileEntry> readMergedDataFiles(
+            List<ManifestFileMeta> manifests,
+            BucketFilter bucketFilter,
+            Predicate<ProjectedManifestEntry> filter)
+            throws IOException {
+        Map<Identifier, ExpireFileEntry> map = new HashMap<>();
+        FileEntry.mergeEntries(
+                ManifestReadThreadPool.sequentialBatchedExecute(
+                        manifest -> {
+                            if (!bucketFilter.mayContain(manifest)) {
+                                return Collections.emptyList();
+                            }
+                            return manifestFile.readExpireFileEntries(
+                                    manifest.fileName(), bucketFilter, filter);
+                        },
+                        manifests,
+                        manifestReadParallelism),
+                map);
+        return map.values();
+    }
+
+    private BucketFilter createCandidateBucketFilter(
+            Map<BinaryRow, Map<Integer, Set<String>>> candidates) {
+        NavigableSet<Integer> candidateBuckets = new TreeSet<>();
+        for (Map<Integer, Set<String>> buckets : candidates.values()) {
+            candidateBuckets.addAll(buckets.keySet());
+        }
+
+        ManifestBucketFilter filter =
+                new ManifestBucketFilter() {
+                    @Override
+                    public boolean test(BinaryRow partition, Integer bucket, Integer totalBuckets) {
+                        Map<Integer, Set<String>> buckets = candidates.get(partition);
+                        return buckets != null && buckets.containsKey(bucket);
+                    }
+
+                    @Override
+                    public boolean mayContain(int minBucket, int maxBucket, int totalBuckets) {
+                        Integer firstCandidate = candidateBuckets.ceiling(minBucket);
+                        return firstCandidate != null && firstCandidate <= maxBucket;
+                    }
+                };
+        return new BucketFilter(false, null, null, filter);
+    }
+
+    private void addDataFile(
+            Map<BinaryRow, Map<Integer, Set<String>>> dataFiles, ExpireFileEntry entry) {
+        dataFiles
+                .computeIfAbsent(entry.partition(), p -> new HashMap<>())
+                .computeIfAbsent(entry.bucket(), b -> new HashSet<>())
+                .add(entry.fileName());
+    }
+
     protected boolean containsDataFile(
             Map<BinaryRow, Map<Integer, Set<String>>> dataFiles, ExpireFileEntry entry) {
-        Map<Integer, Set<String>> buckets = dataFiles.get(entry.partition());
+        return containsDataFile(dataFiles, entry.partition(), entry.bucket(), entry.fileName());
+    }
+
+    private boolean containsDataFile(
+            Map<BinaryRow, Map<Integer, Set<String>>> dataFiles, ProjectedManifestEntry entry) {
+        return containsDataFile(dataFiles, entry.partition(), entry.bucket(), entry.fileName());
+    }
+
+    private boolean containsDataFile(
+            Map<BinaryRow, Map<Integer, Set<String>>> dataFiles,
+            BinaryRow partition,
+            int bucket,
+            String fileName) {
+        Map<Integer, Set<String>> buckets = dataFiles.get(partition);
         if (buckets != null) {
-            Set<String> fileNames = buckets.get(entry.bucket());
+            Set<String> fileNames = buckets.get(bucket);
             if (fileNames != null) {
-                return fileNames.contains(entry.fileName());
+                return fileNames.contains(fileName);
             }
         }
         return false;
@@ -560,4 +670,19 @@ public void executeAll(Collection<Runnable> tasks) {
             throw new RuntimeException(e.getCause());
         }
     }
+
+    /** Candidate data files from one snapshot delta manifest. */
+    public static class DataFileDeletionPlan {
+
+        private final Map<Path, Pair<ExpireFileEntry, List<Path>>> dataFileToDelete;
+
+        private DataFileDeletionPlan(
+                Map<Path, Pair<ExpireFileEntry, List<Path>>> dataFileToDelete) {
+            this.dataFileToDelete = dataFileToDelete;
+        }
+
+        private static DataFileDeletionPlan empty() {
+            return new DataFileDeletionPlan(Collections.emptyMap());
+        }
+    }
 }

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

@@ -70,8 +70,8 @@ public void cleanDeletedDataFiles(Snapshot snapshot, Predicate<ExpireFileEntry>
     }
 
     @Override
-    public List<Path> planDeletedInDeltaManifest(
-            Snapshot snapshot, Predicate<ExpireFileEntry> skipper) {
+    public List<Path> dataFilesToDelete(
+            DataFileDeletionPlan plan, Predicate<ExpireFileEntry> skipper) {
         Predicate<ExpireFileEntry> enriched = skipper;
         if (changelogDecoupled && !produceChangelog) {
             // Skip clean the 'APPEND' data files.If we do not have the file source information
@@ -83,7 +83,7 @@ public List<Path> planDeletedInDeltaManifest(
                                     || (manifestEntry.fileSource().orElse(FileSource.APPEND)
                                             == FileSource.APPEND);
         }
-        return super.planDeletedInDeltaManifest(snapshot, enriched);
+        return super.dataFilesToDelete(plan, enriched);
     }
 
     @Override