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 db45ed237ada50eb0206352cc191a8c499a27cdc Author: Dapeng Sun(孙大鹏) <[email protected]> AuthorDate: Fri Sep 11 15:36:30 2026 +0800 [core] Push partition and bucket filters into the uncached manifest read path (#9739) --- .../apache/paimon/manifest/ManifestEntryCache.java | 36 ++++++++++++++ .../org/apache/paimon/manifest/ManifestFile.java | 13 ++++- .../java/org/apache/paimon/utils/ObjectsCache.java | 8 ++- .../apache/paimon/manifest/ManifestFileTest.java | 58 +++++++++++++++++++++- 4 files changed, 112 insertions(+), 3 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestEntryCache.java b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestEntryCache.java index 75af92c75a..ab5b87551e 100644 --- a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestEntryCache.java +++ b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestEntryCache.java @@ -65,13 +65,49 @@ import static org.apache.paimon.manifest.ManifestEntrySerializer.totalBucketGett @ThreadSafe public class ManifestEntryCache extends ObjectsCache<Path, ManifestEntry, ManifestEntrySegments> { + @Nullable private final FilteredReader filteredReader; + public ManifestEntryCache( SegmentsCache<Path> cache, ObjectSerializer<ManifestEntry> projectedSerializer, RowType formatSchema, FunctionWithIOException<Path, Long> fileSizeFunction, BiFunctionWithIOE<Path, Long, CloseableIterator<InternalRow>> reader) { + this(cache, projectedSerializer, formatSchema, fileSizeFunction, reader, null); + } + + public ManifestEntryCache( + SegmentsCache<Path> cache, + ObjectSerializer<ManifestEntry> projectedSerializer, + RowType formatSchema, + FunctionWithIOException<Path, Long> fileSizeFunction, + BiFunctionWithIOE<Path, Long, CloseableIterator<InternalRow>> reader, + @Nullable FilteredReader filteredReader) { super(cache, projectedSerializer, formatSchema, fileSizeFunction, reader); + this.filteredReader = filteredReader; + } + + /** Uncached reads skip non-matching partitions and buckets before decoding their stats. */ + @Override + protected CloseableIterator<InternalRow> createFilteredIterator( + Path path, @Nullable Long fileSize, Filters<ManifestEntry> filters) throws IOException { + if (filteredReader != null && filters instanceof ManifestEntryFilters) { + ManifestEntryFilters manifestFilters = (ManifestEntryFilters) filters; + return filteredReader.read( + path, fileSize, manifestFilters.partitionFilter, manifestFilters.bucketFilter); + } + return super.createFilteredIterator(path, fileSize, filters); + } + + /** Reader of manifest rows which can skip entries by partition and bucket while decoding. */ + @FunctionalInterface + public interface FilteredReader { + CloseableIterator<InternalRow> read( + Path path, + @Nullable Long fileSize, + @Nullable PartitionPredicate partitionFilter, + @Nullable BucketFilter bucketFilter) + throws IOException; } @Override 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 0088744dc5..0dc99a0470 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 @@ -91,7 +91,18 @@ public class ManifestFile extends ObjectsFile<ManifestEntry> { protected ManifestEntryCache createCache( @Nullable SegmentsCache<Path> cache, RowType formatType) { return new ManifestEntryCache( - cache, serializer, formatType, super::fileSize, this::createIterator); + cache, + serializer, + formatType, + super::fileSize, + this::createIterator, + (path, fileSize, partitionFilter, bucketFilter) -> + createManifestIterator( + fileIO, + path, + ManifestEntry.MANIFEST_ROW_TYPE, + partitionFilter, + bucketFilter)); } @Override diff --git a/paimon-core/src/main/java/org/apache/paimon/utils/ObjectsCache.java b/paimon-core/src/main/java/org/apache/paimon/utils/ObjectsCache.java index d978fdbbd1..4530f698e2 100644 --- a/paimon-core/src/main/java/org/apache/paimon/utils/ObjectsCache.java +++ b/paimon-core/src/main/java/org/apache/paimon/utils/ObjectsCache.java @@ -90,7 +90,7 @@ public abstract class ObjectsCache<K, V, S extends Segments> { return readFromSegments(segments, filters, convertor); } else { return readFromIterator( - reader.apply(key, fileSize), + createFilteredIterator(key, fileSize, filters), projectedSerializer, filters.readFilter(), filters.readVFilter(), @@ -99,6 +99,12 @@ public abstract class ObjectsCache<K, V, S extends Segments> { } } + /** Iterator for a file too large to cache; subclasses may push {@code filters} into it. */ + protected CloseableIterator<InternalRow> createFilteredIterator( + K key, @Nullable Long fileSize, Filters<V> filters) throws IOException { + return reader.apply(key, fileSize); + } + protected abstract <R> List<R> readFromSegments( S segments, Filters<V> filters, Function<V, R> convertor) throws IOException; 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 38194fd2a8..c3a50f4ef1 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 @@ -31,6 +31,7 @@ import org.apache.paimon.fs.PositionOutputStream; import org.apache.paimon.fs.local.LocalFileIO; import org.apache.paimon.io.DataFileMeta; import org.apache.paimon.io.DataFileMetaWriteColsLegacySerializer; +import org.apache.paimon.options.MemorySize; import org.apache.paimon.options.Options; import org.apache.paimon.partition.PartitionPredicate; import org.apache.paimon.schema.FileSystemSchemaManager; @@ -40,6 +41,8 @@ import org.apache.paimon.types.RowType; import org.apache.paimon.utils.CloseableIterator; import org.apache.paimon.utils.FailingFileIO; import org.apache.paimon.utils.FileStorePathFactory; +import org.apache.paimon.utils.Filter; +import org.apache.paimon.utils.SegmentsCache; import org.junit.jupiter.api.RepeatedTest; import org.junit.jupiter.api.Test; @@ -48,6 +51,8 @@ import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.Arguments; import org.junit.jupiter.params.provider.MethodSource; +import javax.annotation.Nullable; + import java.io.IOException; import java.io.UncheckedIOException; import java.nio.ByteBuffer; @@ -1224,6 +1229,11 @@ public class ManifestFileTest { } private ManifestFile createManifestFile(String pathStr, long suggestedFileSize) { + return createManifestFile(pathStr, suggestedFileSize, null); + } + + private ManifestFile createManifestFile( + String pathStr, long suggestedFileSize, @Nullable SegmentsCache<Path> cache) { Path path = new Path(pathStr); FileStorePathFactory pathFactory = new FileStorePathFactory( @@ -1251,10 +1261,56 @@ public class ManifestFileTest { "zstd", pathFactory, suggestedFileSize, - null) + cache) .create(); } + @Test + void testBucketFilterPushedDownWhenManifestExceedsCacheElementSize() throws Exception { + List<ManifestEntry> entries = generateData(); + Set<Integer> buckets = + entries.stream().map(ManifestEntry::bucket).collect(Collectors.toSet()); + assertThat(buckets.size()).isGreaterThan(1); + + // A manifest above the cache element size limit is read uncached; the bucket filter must + // still reach the Avro reader instead of being applied after decoding every entry. + SegmentsCache<Path> tinyElementCache = + new SegmentsCache<>(1024, MemorySize.ofMebiBytes(8), 1L); + ManifestFile manifestFile = + createManifestFile(tempDir.toString(), Long.MAX_VALUE, tinyElementCache); + List<ManifestFileMeta> metas = manifestFile.write(entries); + assertThat(metas).hasSize(1); + ManifestFileMeta meta = metas.get(0); + + for (int bucket : buckets) { + BucketFilter bucketFilter = BucketFilter.create(false, bucket, null, null); + List<ManifestEntry> actual = + manifestFile.read( + meta.fileName(), + meta.fileSize(), + null, + bucketFilter, + Filter.alwaysTrue(), + Filter.alwaysTrue()); + List<ManifestEntry> expected = + entries.stream() + .filter(entry -> entry.bucket() == bucket) + .collect(Collectors.toList()); + assertThat(actual).isEqualTo(expected); + } + + // Without any pushdown filter every entry must still be returned. + assertThat( + manifestFile.read( + meta.fileName(), + meta.fileSize(), + null, + null, + Filter.alwaysTrue(), + Filter.alwaysTrue())) + .isEqualTo(entries); + } + private ManifestFileMeta writeSingleManifest( ManifestFile manifestFile, List<ManifestEntry> entries) { List<ManifestFileMeta> manifests = manifestFile.write(entries);
