From 1ade82112a2618fbf5d5538d54b78d75f739bc36 Mon Sep 17 00:00:00 2001 From: zhoulii Date: Thu, 20 Aug 2026 17:21:07 +0800 Subject: [PATCH 1/2] [core] Optimize partition discovery with projected manifest scans --- .../apache/paimon/manifest/ManifestFile.java | 21 +++- .../operation/AbstractFileStoreScan.java | 100 +++++++++++++++- .../operation/AppendOnlyFileStoreScan.java | 5 + .../operation/DataEvolutionFileStoreScan.java | 5 + .../operation/KeyValueFileStoreScan.java | 5 + .../paimon/catalog/CachingCatalogTest.java | 2 + ...FileStoreScanPartitionBucketEntryTest.java | 109 ++++++++++++++++++ 7 files changed, 243 insertions(+), 4 deletions(-) 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 0088744dc5fa..a20ca04b8261 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 @@ -157,14 +157,26 @@ public List read( * materialized with the complete manifest schema. */ public CloseableIterator scan(String fileName, Projection projection) { + return scan(fileName, projection, null, null); + } + + /** + * Scans projected manifest entries and prunes partitions and buckets before materializing the + * nested data file row. + */ + public CloseableIterator scan( + String fileName, + Projection projection, + @Nullable PartitionPredicate partitionFilter, + @Nullable BucketFilter bucketFilter) { try { CloseableIterator rows = createManifestIterator( fileIO, pathFactory.toPath(fileName), projection.projectedType(), - null, - null); + partitionFilter, + bucketFilter); return new CloseableIterator() { @Override @@ -363,6 +375,11 @@ public boolean isCacheEnabled() { return cache != null; } + /** Returns whether a manifest of this size is eligible for the configured cache. */ + public boolean isCacheable(long fileSize) { + return cache != null && fileSize <= cache.maxElementSize(); + } + public ManifestFile create() { return new ManifestFile( fileIO, diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java b/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java index a9ef5902ec9e..2cfc0ec1d56f 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java @@ -21,6 +21,7 @@ import org.apache.paimon.Snapshot; import org.apache.paimon.data.BinaryRow; import org.apache.paimon.data.InternalRow; +import org.apache.paimon.io.DataFileMeta; import org.apache.paimon.manifest.BucketEntry; import org.apache.paimon.manifest.BucketFilter; import org.apache.paimon.manifest.FileEntry; @@ -30,6 +31,7 @@ import org.apache.paimon.manifest.ManifestFile; import org.apache.paimon.manifest.ManifestFileMeta; import org.apache.paimon.manifest.PartitionEntry; +import org.apache.paimon.manifest.ProjectedManifestEntry; import org.apache.paimon.manifest.SimpleFileEntry; import org.apache.paimon.operation.metrics.ScanMetrics; import org.apache.paimon.operation.metrics.ScanStats; @@ -40,6 +42,7 @@ import org.apache.paimon.table.source.ScanMode; import org.apache.paimon.types.RowType; import org.apache.paimon.utils.BiFilter; +import org.apache.paimon.utils.CloseableIterator; import org.apache.paimon.utils.Filter; import org.apache.paimon.utils.ListUtils; import org.apache.paimon.utils.Pair; @@ -54,6 +57,7 @@ import javax.annotation.Nullable; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collections; import java.util.Iterator; import java.util.List; @@ -74,6 +78,8 @@ public abstract class AbstractFileStoreScan implements FileStoreScan { private static final Logger LOG = LoggerFactory.getLogger(AbstractFileStoreScan.class); + private static final ProjectedManifestEntry.Projection PARTITION_ENTRY_PROJECTION = + createPartitionEntryProjection(); private final ManifestsReader manifestsReader; private final SnapshotManager snapshotManager; @@ -367,16 +373,71 @@ public List readPartitionEntries() { List manifests = readManifests().filteredManifests; Map partitions = new ConcurrentHashMap<>(); Consumer processor = - m -> + manifest -> { + if (canUseProjectedPartitionScan(manifest)) { + readProjectedPartitionEntries(manifest, partitions); + } else { PartitionEntry.merge( - readManifest(m, PartitionEntry::fromManifestEntry, null, null), + readManifest( + manifest, PartitionEntry::fromManifestEntry, null, null), partitions); + } + }; randomlyOnlyExecute(getExecutorService(parallelism), processor, manifests); return partitions.values().stream() .filter(p -> p.fileCount() > 0) .collect(Collectors.toList()); } + private boolean canUseProjectedPartitionScan(ManifestFileMeta manifest) { + // Projected scans bypass the manifest cache, so preserve the cached read path for files + // which are eligible for caching. + return !manifestFileFactory.isCacheable(manifest.fileSize()) + && manifestEntryFilter == null + && !requiresFullManifestEntryForPartitionScan(); + } + + private void readProjectedPartitionEntries( + ManifestFileMeta manifest, Map partitions) { + long count = 0; + try (CloseableIterator entries = + manifestFileFactory + .create() + .scan( + manifest.fileName(), + PARTITION_ENTRY_PROJECTION, + manifestsReader.partitionFilter(), + createBucketFilter())) { + while (entries.hasNext()) { + ProjectedManifestEntry entry = entries.next(); + if (!filterProjectedPartitionEntry(entry)) { + continue; + } + + PartitionEntry partitionEntry = PartitionEntry.fromManifestEntry(entry); + partitions.compute( + partitionEntry.partition(), + (partition, previous) -> + previous == null ? partitionEntry : previous.merge(partitionEntry)); + count++; + } + } catch (Exception e) { + throw new RuntimeException("Failed to scan manifest " + manifest.fileName(), e); + } + LOG.info("Read {} projected manifest entries from {}", count, manifest.fileName()); + } + + private boolean filterProjectedPartitionEntry(ProjectedManifestEntry entry) { + int level = entry.level(); + if (specifiedLevel != null && level != specifiedLevel) { + return false; + } + if (levelFilter != null && !levelFilter.test(level)) { + return false; + } + return fileNameFilter == null || fileNameFilter.test(entry.fileName()); + } + @Override public List readBucketEntries() { List manifests = readManifests().filteredManifests; @@ -476,6 +537,15 @@ protected TableSchema scanTableSchema(long id) { /** Note: Keep this thread-safe. */ protected abstract boolean filterByStats(ManifestEntry entry); + /** + * Returns whether partition scanning needs a complete manifest entry for subclass-specific + * filtering. Subclasses should opt in to projected scanning only when all active filters can be + * evaluated from {@link #PARTITION_ENTRY_PROJECTION}. + */ + protected boolean requiresFullManifestEntryForPartitionScan() { + return true; + } + protected boolean postFilterManifestEntriesEnabled() { return false; } @@ -578,6 +648,32 @@ private Filter createEntryRowFilter() { }; } + /** + * Keeps the fields required to aggregate {@link PartitionEntry}: kind controls the sign of + * added/deleted files, partition is the grouping key, total buckets is part of the result, and + * file size, row count and creation time form its statistics. File name and level are also kept + * to preserve the corresponding structural filters. + */ + private static ProjectedManifestEntry.Projection createPartitionEntryProjection() { + RowType manifestType = ManifestEntry.MANIFEST_ROW_TYPE; + return ProjectedManifestEntry.Projection.create( + new RowType( + false, + Arrays.asList( + manifestType.getField(ManifestEntry.KIND), + manifestType.getField(ManifestEntry.PARTITION), + manifestType.getField(ManifestEntry.TOTAL_BUCKETS), + manifestType + .getField(ManifestEntry.FILE) + .newType( + DataFileMeta.SCHEMA.project( + DataFileMeta.FILE_NAME, + DataFileMeta.FILE_SIZE, + DataFileMeta.ROW_COUNT, + DataFileMeta.LEVEL, + DataFileMeta.CREATION_TIME))))); + } + // ------------------------------------------------------------------------ // End Thread Safe Methods // ------------------------------------------------------------------------ diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/AppendOnlyFileStoreScan.java b/paimon-core/src/main/java/org/apache/paimon/operation/AppendOnlyFileStoreScan.java index 7a0cdec31293..46ecce576c4e 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/AppendOnlyFileStoreScan.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/AppendOnlyFileStoreScan.java @@ -89,6 +89,11 @@ public AppendOnlyFileStoreScan withFilter(Predicate predicate) { return this; } + @Override + protected boolean requiresFullManifestEntryForPartitionScan() { + return inputFilter != null; + } + @Override public FileStoreScan withCompleteFilter(Predicate predicate) { this.bucketSelectConverter.convert(predicate).ifPresent(this::withTotalAwareBucketFilter); diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionFileStoreScan.java b/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionFileStoreScan.java index 88053b03e300..bfae7a92e127 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionFileStoreScan.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionFileStoreScan.java @@ -125,6 +125,11 @@ public DataEvolutionFileStoreScan withFilter(Predicate predicate) { return this; } + @Override + protected boolean requiresFullManifestEntryForPartitionScan() { + return super.requiresFullManifestEntryForPartitionScan() || rowRangeIndex != null; + } + @Override public FileStoreScan withReadType(RowType readType) { if (readType != null) { diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreScan.java b/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreScan.java index a3623c1b903d..1b9f1a71a13b 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreScan.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreScan.java @@ -130,6 +130,11 @@ public KeyValueFileStoreScan withValueFilter(Predicate predicate) { return this; } + @Override + protected boolean requiresFullManifestEntryForPartitionScan() { + return keyFilter != null || isValueFilterEnabled(); + } + @Override public FileStoreScan enableValueFilter() { this.valueFilterForceEnabled = true; diff --git a/paimon-core/src/test/java/org/apache/paimon/catalog/CachingCatalogTest.java b/paimon-core/src/test/java/org/apache/paimon/catalog/CachingCatalogTest.java index 37b991fc6846..fb3dc50ae2c4 100644 --- a/paimon-core/src/test/java/org/apache/paimon/catalog/CachingCatalogTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/catalog/CachingCatalogTest.java @@ -563,6 +563,8 @@ private void readTableForTestManifestCache(Catalog catalog, Identifier tableIden // test copy too table = catalog.getTable(tableIdent).copy(Collections.singletonMap("a", "b")); ReadBuilder readBuilder = table.newReadBuilder(); + // Partition discovery should keep working from cache after the manifest is deleted. + assertThat(readBuilder.newScan().listPartitionEntries()).isNotEmpty(); TableScan scan = readBuilder.newScan(); TableRead read = readBuilder.newRead(); read.createReader(scan.plan()).forEachRemaining(r -> {}); diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreScanPartitionBucketEntryTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreScanPartitionBucketEntryTest.java index 8ff5023b160a..2648e584f941 100644 --- a/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreScanPartitionBucketEntryTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreScanPartitionBucketEntryTest.java @@ -20,7 +20,9 @@ import org.apache.paimon.data.GenericRow; import org.apache.paimon.manifest.BucketEntry; +import org.apache.paimon.manifest.ManifestEntry; import org.apache.paimon.manifest.PartitionEntry; +import org.apache.paimon.predicate.PredicateBuilder; import org.apache.paimon.table.sink.BatchTableWrite; import org.apache.paimon.table.sink.CommitMessage; import org.apache.paimon.table.sink.TableCommitImpl; @@ -32,6 +34,7 @@ import java.util.Collections; import java.util.List; import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Collectors; import static org.assertj.core.api.Assertions.assertThat; @@ -187,6 +190,100 @@ public void testReadPartitionEntriesAppendOnlyTable() throws Exception { } } + @Test + public void testProjectedPartitionEntriesWithStructuralFilters() throws Exception { + createAppendOnlyTable(); + writeRows(1, 0, 5); + writeRows(1, 5, 15); + + List files = table.store().newScan().plan().files(); + assertThat(files).hasSize(2); + ManifestEntry selected = + files.stream().filter(file -> file.file().rowCount() == 5).findFirst().get(); + + List fileNameFiltered = + table.newSnapshotReader() + .withDataFileNameFilter( + fileName -> fileName.equals(selected.file().fileName())) + .partitionEntries(); + assertThat(fileNameFiltered) + .singleElement() + .satisfies( + entry -> { + assertThat(entry.recordCount()).isEqualTo(5); + assertThat(entry.fileCount()).isEqualTo(1); + }); + + assertThat(table.newSnapshotReader().withLevelFilter(level -> false).partitionEntries()) + .isEmpty(); + assertThat(table.newSnapshotReader().withBucket(1).partitionEntries()).isEmpty(); + } + + @Test + public void testPartitionEntriesFallbackForManifestEntryFilter() throws Exception { + createAppendOnlyTable(); + writeRows(1, 0, 5); + writeRows(1, 5, 15); + AtomicInteger filterCalls = new AtomicInteger(); + + List entries = + table.newSnapshotReader() + .withManifestEntryFilter( + entry -> { + filterCalls.incrementAndGet(); + return entry.file().embeddedIndex() == null; + }) + .partitionEntries(); + + assertThat(filterCalls).hasValueGreaterThan(0); + assertThat(entries) + .singleElement() + .satisfies( + entry -> { + assertThat(entry.recordCount()).isEqualTo(15); + assertThat(entry.fileCount()).isEqualTo(2); + }); + } + + @Test + public void testPartitionEntriesFallbackForAppendStatsFilter() throws Exception { + createAppendOnlyTable(); + writeRows(1, 0, 5); + writeRows(1, 5, 15); + + List entries = + table.newSnapshotReader() + .withFilter(new PredicateBuilder(table.rowType()).lessThan(1, 5)) + .partitionEntries(); + + assertThat(entries) + .singleElement() + .satisfies( + entry -> { + assertThat(entry.recordCount()).isEqualTo(5); + assertThat(entry.fileCount()).isEqualTo(1); + }); + } + + @Test + public void testPartitionEntriesFallbackForKeyStatsFilter() throws Exception { + writeRows(1, 0, 5); + writeRows(1, 5, 15); + + List entries = + table.newSnapshotReader() + .withFilter(new PredicateBuilder(table.rowType()).lessThan(1, 5)) + .partitionEntries(); + + assertThat(entries) + .singleElement() + .satisfies( + entry -> { + assertThat(entry.recordCount()).isEqualTo(5); + assertThat(entry.fileCount()).isEqualTo(1); + }); + } + @Test public void testReadBucketEntriesSinglePartition() throws Exception { // Write data to a single partition with 1 bucket @@ -297,4 +394,16 @@ public void testReadBucketEntriesAppendOnlyTable() throws Exception { assertThat(entry.fileCount()).isEqualTo(1); } } + + private void writeRows(int partition, int from, int to) throws Exception { + BatchTableWrite write = table.newWrite(commitUser); + for (int i = from; i < to; i++) { + write.write(GenericRow.of(partition, i, (long) i)); + } + List messages = write.prepareCommit(); + TableCommitImpl commit = table.newCommit(commitUser); + commit.commit(messages); + write.close(); + commit.close(); + } } From 587b8d378058deeff048a6c1590938a50a3096e3 Mon Sep 17 00:00:00 2001 From: zhoulii Date: Fri, 21 Aug 2026 22:31:16 +0800 Subject: [PATCH 2/2] [core] Centralize projected manifest scan selection --- .../operation/AbstractFileStoreScan.java | 121 +++++++++--------- 1 file changed, 62 insertions(+), 59 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java b/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java index 2cfc0ec1d56f..22ada9bd49ee 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java @@ -373,71 +373,25 @@ public List readPartitionEntries() { List manifests = readManifests().filteredManifests; Map partitions = new ConcurrentHashMap<>(); Consumer processor = - manifest -> { - if (canUseProjectedPartitionScan(manifest)) { - readProjectedPartitionEntries(manifest, partitions); - } else { - PartitionEntry.merge( - readManifest( - manifest, PartitionEntry::fromManifestEntry, null, null), - partitions); - } - }; + manifest -> + readManifest( + manifest, + PartitionEntry::fromManifestEntry, + PARTITION_ENTRY_PROJECTION, + this::filterProjectedPartitionEntry, + entry -> + partitions.compute( + entry.partition(), + (partition, previous) -> + previous == null + ? entry + : previous.merge(entry))); randomlyOnlyExecute(getExecutorService(parallelism), processor, manifests); return partitions.values().stream() .filter(p -> p.fileCount() > 0) .collect(Collectors.toList()); } - private boolean canUseProjectedPartitionScan(ManifestFileMeta manifest) { - // Projected scans bypass the manifest cache, so preserve the cached read path for files - // which are eligible for caching. - return !manifestFileFactory.isCacheable(manifest.fileSize()) - && manifestEntryFilter == null - && !requiresFullManifestEntryForPartitionScan(); - } - - private void readProjectedPartitionEntries( - ManifestFileMeta manifest, Map partitions) { - long count = 0; - try (CloseableIterator entries = - manifestFileFactory - .create() - .scan( - manifest.fileName(), - PARTITION_ENTRY_PROJECTION, - manifestsReader.partitionFilter(), - createBucketFilter())) { - while (entries.hasNext()) { - ProjectedManifestEntry entry = entries.next(); - if (!filterProjectedPartitionEntry(entry)) { - continue; - } - - PartitionEntry partitionEntry = PartitionEntry.fromManifestEntry(entry); - partitions.compute( - partitionEntry.partition(), - (partition, previous) -> - previous == null ? partitionEntry : previous.merge(partitionEntry)); - count++; - } - } catch (Exception e) { - throw new RuntimeException("Failed to scan manifest " + manifest.fileName(), e); - } - LOG.info("Read {} projected manifest entries from {}", count, manifest.fileName()); - } - - private boolean filterProjectedPartitionEntry(ProjectedManifestEntry entry) { - int level = entry.level(); - if (specifiedLevel != null && level != specifiedLevel) { - return false; - } - if (levelFilter != null && !levelFilter.test(level)) { - return false; - } - return fileNameFilter == null || fileNameFilter.test(entry.fileName()); - } - @Override public List readBucketEntries() { List manifests = readManifests().filteredManifests; @@ -560,6 +514,55 @@ public List readManifest(ManifestFileMeta manifest) { return readManifest(manifest, Function.identity(), null, null); } + private void readManifest( + ManifestFileMeta manifest, + Function converter, + ProjectedManifestEntry.Projection projection, + Filter projectedFilter, + Consumer consumer) { + // A projected scan reads the file directly, so use the normal path when this manifest can + // benefit from the cache or an active filter needs fields outside the projection. + if (manifestFileFactory.isCacheable(manifest.fileSize()) + || manifestEntryFilter != null + || requiresFullManifestEntryForPartitionScan()) { + readManifest(manifest, converter, null, null).forEach(consumer); + return; + } + + long count = 0; + try (CloseableIterator entries = + manifestFileFactory + .create() + .scan( + manifest.fileName(), + projection, + manifestsReader.partitionFilter(), + createBucketFilter())) { + while (entries.hasNext()) { + ProjectedManifestEntry entry = entries.next(); + if (!projectedFilter.test(entry)) { + continue; + } + consumer.accept(converter.apply(entry)); + count++; + } + } catch (Exception e) { + throw new RuntimeException("Failed to scan manifest " + manifest.fileName(), e); + } + LOG.info("Read {} projected manifest entries from {}", count, manifest.fileName()); + } + + private boolean filterProjectedPartitionEntry(ProjectedManifestEntry entry) { + int level = entry.level(); + if (specifiedLevel != null && level != specifiedLevel) { + return false; + } + if (levelFilter != null && !levelFilter.test(level)) { + return false; + } + return fileNameFilter == null || fileNameFilter.test(entry.fileName()); + } + private List readManifest( ManifestFileMeta manifest, Function converter,