This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 2f671d6e02 [core] Push partition and bucket filters into the uncached
manifest read path (#9739)
2f671d6e02 is described below
commit 2f671d6e02772ac47e1737619c91efcacdac3e9b
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);
