This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 6bfd8418dc8fae6ae75c53dd0e04ece8be303a46 Author: Jingsong Lee <[email protected]> AuthorDate: Sat Sep 19 08:10:02 2026 +0800 [core] Optimize tag reads during snapshot expiration (#9967) --- .../apache/paimon/manifest/ManifestAvroReader.java | 12 ++ .../org/apache/paimon/manifest/ManifestFile.java | 33 ++++- .../apache/paimon/operation/FileDeletionBase.java | 145 +++++++++++++++++-- .../apache/paimon/operation/SnapshotDeletion.java | 6 +- .../apache/paimon/table/ExpireSnapshotsImpl.java | 81 +++++++---- .../apache/paimon/manifest/ManifestFileTest.java | 44 ++++++ .../paimon/operation/ExpireSnapshotsTest.java | 153 ++++++++++++++++++++- 7 files changed, 427 insertions(+), 47 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestAvroReader.java b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestAvroReader.java index 95134f36a0..1c6c121eb1 100644 --- a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestAvroReader.java +++ b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestAvroReader.java @@ -439,6 +439,18 @@ public final class ManifestAvroReader implements AutoCloseable { 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, diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFile.java b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFile.java index afc122c723..608c7dc576 100644 --- a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFile.java +++ b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFile.java @@ -48,6 +48,7 @@ import java.util.ArrayList; 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 class ManifestFile extends ObjectsFile<ManifestEntry> { } 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; } diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/FileDeletionBase.java b/paimon-core/src/main/java/org/apache/paimon/operation/FileDeletionBase.java index ad544af0ea..1b8b245e9b 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/FileDeletionBase.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/FileDeletionBase.java @@ -25,14 +25,17 @@ import org.apache.paimon.fs.Path; 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.HashSet; 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 @@ public abstract class FileDeletionBase<T extends Snapshot> { } /** 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 abstract class FileDeletionBase<T extends Snapshot> { } 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 abstract class FileDeletionBase<T extends Snapshot> { 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 @@ public abstract class FileDeletionBase<T extends Snapshot> { 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 @@ public abstract class FileDeletionBase<T extends Snapshot> { 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 abstract class FileDeletionBase<T extends Snapshot> { 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()); + } + } } diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/SnapshotDeletion.java b/paimon-core/src/main/java/org/apache/paimon/operation/SnapshotDeletion.java index 00bb731d03..bbeb124627 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/SnapshotDeletion.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/SnapshotDeletion.java @@ -70,8 +70,8 @@ public class SnapshotDeletion extends FileDeletionBase<Snapshot> { } @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 class SnapshotDeletion extends FileDeletionBase<Snapshot> { || (manifestEntry.fileSource().orElse(FileSource.APPEND) == FileSource.APPEND); } - return super.planDeletedInDeltaManifest(snapshot, enriched); + return super.dataFilesToDelete(plan, enriched); } @Override diff --git a/paimon-core/src/main/java/org/apache/paimon/table/ExpireSnapshotsImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/ExpireSnapshotsImpl.java index b7cabb4d32..4b80085489 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/ExpireSnapshotsImpl.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/ExpireSnapshotsImpl.java @@ -24,6 +24,7 @@ import org.apache.paimon.annotation.VisibleForTesting; import org.apache.paimon.consumer.ConsumerManager; import org.apache.paimon.fs.Path; import org.apache.paimon.manifest.ExpireFileEntry; +import org.apache.paimon.operation.FileDeletionBase.DataFileDeletionPlan; import org.apache.paimon.operation.SnapshotDeletion; import org.apache.paimon.options.ExpireConfig; import org.apache.paimon.tag.Tag; @@ -274,8 +275,7 @@ public class ExpireSnapshotsImpl implements ExpireSnapshots { List<Snapshot> taggedSnapshots, long beginInclusiveId) throws ExecutionException, InterruptedException { - Map<Long, Long> tagIdBySnapshotId = new HashMap<>(); - Map<Long, Snapshot> tags = new HashMap<>(); + Map<Long, Snapshot> tagBySnapshotId = new HashMap<>(); int tagIndex = -1; for (Snapshot snapshot : snapshotsIncludingEnd) { long id = snapshot.id(); @@ -286,16 +286,11 @@ public class ExpireSnapshotsImpl implements ExpireSnapshots { tagIndex = advancePreviousSnapshot(taggedSnapshots, tagIndex, id); if (tagIndex >= 0) { Snapshot tag = taggedSnapshots.get(tagIndex); - tagIdBySnapshotId.put(id, tag.id()); - tags.put(tag.id(), tag); + tagBySnapshotId.put(id, tag); } } - Map<Long, Optional<Predicate<ExpireFileEntry>>> skippers = - collectTagSkippers(tags.values()); - Predicate<ExpireFileEntry> deleteAll = entry -> false; - List<CompletableFuture<List<Path>>> futures = new ArrayList<>(); - int plannedSnapshots = 0; + List<Snapshot> batch = new ArrayList<>(snapshotExpireBatchSize); for (Snapshot snapshot : snapshotsIncludingEnd) { long id = snapshot.id(); if (id == beginInclusiveId) { @@ -304,44 +299,76 @@ public class ExpireSnapshotsImpl implements ExpireSnapshots { if (LOG.isDebugEnabled()) { LOG.debug("Ready to delete merge tree files not used by snapshot #{}", id); } + batch.add(snapshot); + if (batch.size() >= snapshotExpireBatchSize) { + cleanDataFileBatch(batch, tagBySnapshotId); + } + } + cleanDataFileBatch(batch, tagBySnapshotId); + } + + private void cleanDataFileBatch(List<Snapshot> snapshots, Map<Long, Snapshot> tagBySnapshotId) + throws ExecutionException, InterruptedException { + if (snapshots.isEmpty()) { + return; + } - Long tagId = tagIdBySnapshotId.get(id); + List<CompletableFuture<DataFileDeletionPlan>> planFutures = new ArrayList<>(); + for (Snapshot snapshot : snapshots) { + planFutures.add( + CompletableFuture.supplyAsync( + () -> snapshotDeletion.planDeletedInDeltaManifest(snapshot), + fileExecutor)); + } + List<DataFileDeletionPlan> plans = getAll(planFutures); + + Map<Long, Snapshot> tags = new HashMap<>(); + Map<Long, List<DataFileDeletionPlan>> plansByTag = new HashMap<>(); + for (int i = 0; i < snapshots.size(); i++) { + Snapshot tag = tagBySnapshotId.get(snapshots.get(i).id()); + if (tag != null) { + tags.put(tag.id(), tag); + plansByTag.computeIfAbsent(tag.id(), id -> new ArrayList<>()).add(plans.get(i)); + } + } + + Map<Long, Optional<Predicate<ExpireFileEntry>>> skippers = + collectTagSkippers(tags, plansByTag); + Predicate<ExpireFileEntry> deleteAll = entry -> false; + List<Path> paths = new ArrayList<>(); + for (int i = 0; i < snapshots.size(); i++) { + Snapshot snapshot = snapshots.get(i); + Snapshot tag = tagBySnapshotId.get(snapshot.id()); Optional<Predicate<ExpireFileEntry>> skipper = - tagId == null + tag == null ? Optional.of(deleteAll) - : skippers.getOrDefault(tagId, Optional.empty()); + : skippers.getOrDefault(tag.id(), Optional.empty()); if (!skipper.isPresent()) { LOG.info( "Skip cleaning data files of snapshot '{}' due to failed to build skipping set.", - id); + snapshot.id()); continue; } - futures.add( - CompletableFuture.supplyAsync( - () -> - snapshotDeletion.planDeletedInDeltaManifest( - snapshot, skipper.get()), - fileExecutor)); - if (++plannedSnapshots >= snapshotExpireBatchSize) { - cleanBatch(futures); - plannedSnapshots = 0; - } + paths.addAll(snapshotDeletion.dataFilesToDelete(plans.get(i), skipper.get())); } - cleanBatch(futures); + snapshotDeletion.cleanDataFiles(paths); + snapshots.clear(); } private Map<Long, Optional<Predicate<ExpireFileEntry>>> collectTagSkippers( - Collection<Snapshot> tags) throws ExecutionException, InterruptedException { + Map<Long, Snapshot> tags, Map<Long, List<DataFileDeletionPlan>> plansByTag) + throws ExecutionException, InterruptedException { Map<Long, CompletableFuture<Optional<Predicate<ExpireFileEntry>>>> futures = new HashMap<>(); - for (Snapshot tag : tags) { + for (Snapshot tag : tags.values()) { futures.put( tag.id(), CompletableFuture.supplyAsync( () -> { try { return Optional.of( - snapshotDeletion.createDataFileSkipperForTag(tag)); + snapshotDeletion.createDataFileSkipperForTag( + tag, plansByTag.get(tag.id()))); } catch (Exception e) { LOG.info( "Failed to build data file skipping set for tag snapshot '{}'.", diff --git a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileTest.java b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileTest.java index 5172c30521..24f84659d1 100644 --- a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileTest.java @@ -1084,6 +1084,50 @@ public class ManifestFileTest { } } + @Test + void testReadExpireFileEntriesPushesFiltersIntoStreamingScan() { + List<ManifestEntry> entries = Arrays.asList(gen.next(), gen.next(), gen.next()); + ManifestFile manifestFile = createManifestFile(tempDir.toString(), Long.MAX_VALUE); + ManifestFileMeta manifest = writeSingleManifest(manifestFile, entries); + + int[] exactFilterCalls = {0}; + List<ExpireFileEntry> bucketRejected = + manifestFile.readExpireFileEntries( + manifest.fileName(), + new BucketFilter(false, null, bucket -> false, null), + entry -> { + exactFilterCalls[0]++; + return true; + }); + assertThat(bucketRejected).isEmpty(); + assertThat(exactFilterCalls[0]).isZero(); + + String selectedFile = entries.get(1).fileName(); + ProjectedManifestEntry[] reusableView = {null}; + List<ExpireFileEntry> selected = + manifestFile.readExpireFileEntries( + manifest.fileName(), + null, + entry -> { + if (reusableView[0] == null) { + reusableView[0] = entry; + } else { + assertThat(entry).isSameAs(reusableView[0]); + } + exactFilterCalls[0]++; + return entry.fileName().equals(selectedFile); + }); + + assertThat(exactFilterCalls[0]).isEqualTo(entries.size()); + assertThat(selected) + .containsExactly( + ExpireFileEntry.from( + entries.stream() + .filter(entry -> entry.fileName().equals(selectedFile)) + .findFirst() + .orElseThrow(AssertionError::new))); + } + @Test void testScanProjectedManifestCreatesDistinctEntryWrappers() throws Exception { List<ManifestEntry> entries = Arrays.asList(gen.next(), gen.next(), gen.next()); 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 957e4b7317..7fac929cdd 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 @@ -33,6 +33,7 @@ import org.apache.paimon.fs.Path; import org.apache.paimon.fs.local.LocalFileIO; import org.apache.paimon.io.DataFileMeta; import org.apache.paimon.io.DataFilePathFactory; +import org.apache.paimon.manifest.BucketFilter; import org.apache.paimon.manifest.ExpireFileEntry; import org.apache.paimon.manifest.FileKind; import org.apache.paimon.manifest.FileSource; @@ -40,7 +41,9 @@ import org.apache.paimon.manifest.ManifestEntry; import org.apache.paimon.manifest.ManifestFileMeta; import org.apache.paimon.manifest.ManifestIndexTestUtils; import org.apache.paimon.manifest.ManifestSidecar; +import org.apache.paimon.manifest.ProjectedManifestEntry; import org.apache.paimon.mergetree.compact.DeduplicateMergeFunction; +import org.apache.paimon.operation.FileDeletionBase.DataFileDeletionPlan; import org.apache.paimon.options.ExpireConfig; import org.apache.paimon.options.MemorySize; import org.apache.paimon.schema.FileSystemSchemaManager; @@ -74,6 +77,7 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; import java.util.Collections; +import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; @@ -966,6 +970,85 @@ public class ExpireSnapshotsTest { assertSnapshot(tagManager.getOrThrow("tag6").trimToSnapshot(), allData, snapshotPositions); } + @Test + public void testExpireFiltersTagFilesByDeletionCandidates() throws Exception { + List<KeyValue> candidateData = FileStoreTestUtils.partitionedData(5, gen, "0401", 8); + List<KeyValue> unrelatedData = FileStoreTestUtils.partitionedData(5, gen, "0402", 8); + BinaryRow candidatePartition = gen.getPartition(candidateData.get(0)); + BinaryRow unrelatedPartition = gen.getPartition(unrelatedData.get(0)); + Map<BinaryRow, Map<Integer, RecordWriter<KeyValue>>> writers = new HashMap<>(); + writers.put( + candidatePartition, + Collections.singletonMap( + 0, + FileStoreTestUtils.writeData(store, candidateData, candidatePartition, 0))); + writers.put( + unrelatedPartition, + Collections.singletonMap( + 0, + FileStoreTestUtils.writeData(store, unrelatedData, unrelatedPartition, 0))); + FileStoreTestUtils.commitData(store, 0, writers); + + Snapshot taggedSnapshot = snapshotManager.latestSnapshot(); + TagManager tagManager = store.newTagManager(); + tagManager.createTag( + taggedSnapshot, + "tag1", + store.options().tagDefaultTimeRetained(), + Collections.emptyList(), + false); + + List<ManifestEntry> delete = + store.newScan().plan().files().stream() + .filter(entry -> entry.partition().equals(candidatePartition)) + .map( + entry -> + ManifestEntry.create( + FileKind.DELETE, + entry.partition(), + entry.bucket(), + entry.totalBuckets(), + entry.file())) + .collect(Collectors.toList()); + try (FileStoreCommitImpl commit = store.newCommit()) { + commit.tryCommitOnce( + null, + delete, + Collections.emptyList(), + Collections.emptyList(), + 1, + null, + Collections.emptyMap(), + Snapshot.CommitKind.APPEND, + false, + taggedSnapshot, + true, + null); + } + + CandidateTrackingSnapshotDeletion snapshotDeletion = + new CandidateTrackingSnapshotDeletion(store); + ExpireSnapshotsImpl expire = + newExpireWithSnapshotDeletion(store, snapshotManager, snapshotDeletion); + expire.config( + ExpireConfig.builder() + .snapshotRetainMin(1) + .snapshotRetainMax(1) + .snapshotTimeRetain(Duration.ofMillis(Long.MAX_VALUE)) + .build()); + expire.expire(); + + assertThat(snapshotDeletion.candidateTagSkipperCalls()).isGreaterThan(0); + assertThat(snapshotDeletion.filteredTagReads()).isGreaterThan(0); + assertThat(snapshotDeletion.materializedTagEntries()).isOne(); + List<KeyValue> taggedData = new ArrayList<>(candidateData); + taggedData.addAll(unrelatedData); + assertSnapshot( + tagManager.getOrThrow("tag1").trimToSnapshot(), + taggedData, + Collections.singletonList(taggedData.size())); + } + @Test public void testExpireReadsTagsConcurrentlyWithObjectStoreFileIO() throws Exception { TestFileStore slowStore = createSlowStore(); @@ -1567,12 +1650,11 @@ public class ExpireSnapshotsTest { } @Override - public List<Path> planDeletedInDeltaManifest( - Snapshot snapshot, Predicate<ExpireFileEntry> skipper) { + public DataFileDeletionPlan planDeletedInDeltaManifest(Snapshot snapshot) { if (blockDataFilePlans && shouldBlock(snapshot.id())) { dataFilePlans.awaitConcurrentCall(); } - return super.planDeletedInDeltaManifest(snapshot, skipper); + return super.planDeletedInDeltaManifest(snapshot); } @Override @@ -1608,6 +1690,71 @@ public class ExpireSnapshotsTest { } } + private static class CandidateTrackingSnapshotDeletion extends SnapshotDeletion { + + private final AtomicInteger candidateTagSkipperCalls = new AtomicInteger(); + private final AtomicInteger filteredTagReads = new AtomicInteger(); + private final AtomicInteger materializedTagEntries = new AtomicInteger(); + + private CandidateTrackingSnapshotDeletion(TestFileStore store) { + super( + store.fileIO(), + store.pathFactory(), + store.manifestFileFactory().create(), + store.manifestListFactory().create(), + store.newIndexFileHandler(), + store.newStatsFileHandler(), + store.options().changelogProducer() != CoreOptions.ChangelogProducer.NONE, + store.options().cleanEmptyDirectories(), + store.options().fileOperationThreadNum(), + store.options().scanManifestParallelism()); + } + + @Override + public Predicate<ExpireFileEntry> createDataFileSkipperForTag(Snapshot tag) + throws Exception { + throw new AssertionError("Snapshot expiration must not index all files in a tag."); + } + + @Override + public Predicate<ExpireFileEntry> createDataFileSkipperForTag( + Snapshot tag, Collection<DataFileDeletionPlan> plans) throws Exception { + candidateTagSkipperCalls.incrementAndGet(); + return super.createDataFileSkipperForTag(tag, plans); + } + + @Override + protected Collection<ExpireFileEntry> readMergedDataFiles(List<ManifestFileMeta> manifests) + throws IOException { + throw new AssertionError("Snapshot expiration must not merge all files in a tag."); + } + + @Override + protected Collection<ExpireFileEntry> readMergedDataFiles( + List<ManifestFileMeta> manifests, + BucketFilter bucketFilter, + Predicate<ProjectedManifestEntry> filter) + throws IOException { + filteredTagReads.incrementAndGet(); + Collection<ExpireFileEntry> entries = + super.readMergedDataFiles(manifests, bucketFilter, filter); + materializedTagEntries.addAndGet(entries.size()); + return entries; + } + + private int candidateTagSkipperCalls() { + return candidateTagSkipperCalls.get(); + } + + private int filteredTagReads() { + return filteredTagReads.get(); + } + + private int materializedTagEntries() { + return materializedTagEntries.get(); + } + } + private static class CapturingSnapshotDeletion extends SnapshotDeletion { private final List<List<Object>> deleteBatches = new ArrayList<>();
