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..3f2063e29cd1 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 @@ -365,16 +365,19 @@ public List readSimpleEntries() { @Override public List readPartitionEntries() { List manifests = readManifests().filteredManifests; - Map partitions = new ConcurrentHashMap<>(); - Consumer processor = - m -> - PartitionEntry.merge( - readManifest(m, PartitionEntry::fromManifestEntry, null, null), - partitions); - randomlyOnlyExecute(getExecutorService(parallelism), processor, manifests); - return partitions.values().stream() - .filter(p -> p.fileCount() > 0) - .collect(Collectors.toList()); + return new PartitionEntryScanner( + manifestFileFactory, + manifest -> + readManifest( + manifest, PartitionEntry::fromManifestEntry, null, null), + manifestsReader.partitionFilter(), + createBucketFilter(), + specifiedLevel, + levelFilter, + fileNameFilter, + manifestEntryFilter != null || requiresFullManifestEntryForPartitionScan(), + parallelism) + .scan(manifests); } @Override @@ -476,6 +479,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 the partition entry projection. + */ + protected boolean requiresFullManifestEntryForPartitionScan() { + return true; + } + protected boolean postFilterManifestEntriesEnabled() { return false; } 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/main/java/org/apache/paimon/operation/PartitionEntryScanner.java b/paimon-core/src/main/java/org/apache/paimon/operation/PartitionEntryScanner.java new file mode 100644 index 000000000000..8f8cfdc8c168 --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/operation/PartitionEntryScanner.java @@ -0,0 +1,176 @@ +/* + * 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.data.BinaryRow; +import org.apache.paimon.io.DataFileMeta; +import org.apache.paimon.manifest.BucketFilter; +import org.apache.paimon.manifest.ManifestEntry; +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.partition.PartitionPredicate; +import org.apache.paimon.types.RowType; +import org.apache.paimon.utils.CloseableIterator; +import org.apache.paimon.utils.Filter; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; + +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.function.Function; +import java.util.stream.Collectors; + +import static org.apache.paimon.utils.ManifestReadThreadPool.getExecutorService; +import static org.apache.paimon.utils.ThreadPoolUtils.randomlyOnlyExecute; + +/** + * Scans and aggregates partition statistics from manifest entries. + * + *

It uses a narrow, streaming projection when the manifest cannot benefit from the cache and + * falls back to the caller's complete entry reader when other filters need the full schema. + */ +final class PartitionEntryScanner { + + private static final Logger LOG = LoggerFactory.getLogger(PartitionEntryScanner.class); + private static final ProjectedManifestEntry.Projection PARTITION_ENTRY_PROJECTION = + createPartitionEntryProjection(); + + private final ManifestFile.Factory manifestFileFactory; + private final Function> fullEntryReader; + @Nullable private final PartitionPredicate partitionFilter; + @Nullable private final BucketFilter bucketFilter; + @Nullable private final Integer specifiedLevel; + @Nullable private final Filter levelFilter; + @Nullable private final Filter fileNameFilter; + private final boolean requiresFullManifestEntry; + @Nullable private final Integer parallelism; + + PartitionEntryScanner( + ManifestFile.Factory manifestFileFactory, + Function> fullEntryReader, + @Nullable PartitionPredicate partitionFilter, + @Nullable BucketFilter bucketFilter, + @Nullable Integer specifiedLevel, + @Nullable Filter levelFilter, + @Nullable Filter fileNameFilter, + boolean requiresFullManifestEntry, + @Nullable Integer parallelism) { + this.manifestFileFactory = manifestFileFactory; + this.fullEntryReader = fullEntryReader; + this.partitionFilter = partitionFilter; + this.bucketFilter = bucketFilter; + this.specifiedLevel = specifiedLevel; + this.levelFilter = levelFilter; + this.fileNameFilter = fileNameFilter; + this.requiresFullManifestEntry = requiresFullManifestEntry; + this.parallelism = parallelism; + } + + List scan(List manifests) { + Map partitions = new ConcurrentHashMap<>(); + randomlyOnlyExecute( + getExecutorService(parallelism), + manifest -> scanManifest(manifest, partitions), + manifests); + return partitions.values().stream() + .filter(partition -> partition.fileCount() > 0) + .collect(Collectors.toList()); + } + + private void scanManifest( + ManifestFileMeta manifest, Map partitions) { + // Projected scans read the file directly, so preserve the normal path for cached manifests + // and filters which require fields outside the partition projection. + if (requiresFullManifestEntry || manifestFileFactory.isCacheable(manifest.fileSize())) { + PartitionEntry.merge(fullEntryReader.apply(manifest), partitions); + return; + } + + long count = 0; + try (CloseableIterator entries = + manifestFileFactory + .create() + .scan( + manifest.fileName(), + PARTITION_ENTRY_PROJECTION, + partitionFilter, + bucketFilter)) { + while (entries.hasNext()) { + ProjectedManifestEntry entry = entries.next(); + if (!filter(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 filter(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()); + } + + /** + * 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))))); + } +} 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(); + } }