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 575ce805ed [core] Support push-down valueStats prune when read
manifests. (#7392)
575ce805ed is described below
commit 575ce805ed069ab380d762ee6d9556f8393c6e16
Author: baiyangtx <[email protected]>
AuthorDate: Mon Jun 15 12:57:23 2026 +0800
[core] Support push-down valueStats prune when read manifests. (#7392)
When scanning Paimon tables, each `ManifestEntry` carries per-file
column statistics (value stats) — min/max values, null counts, etc. For
wide tables with many columns, these statistics can consume significant
heap memory during large scans (compaction, queries). In many scan
scenarios, the downstream caller does not need these statistics at all.
This PR adds a `dropStats()` option on `FileStoreScan` that prunes value
stats from `ManifestEntry` during manifest reading, reducing GC pressure
and peak memory usage.
To support this efficiently, two general-purpose improvements are made
to the read pipeline:
1. **`loadFilter` for cache read** — `ObjectsFile` / `ObjectsCache` now
accept a `loadFilter` parameter. When loading manifest files into the
in-memory segment cache, entries not matching the filter are skipped
before deserialization, reducing cache memory footprint.
2. **`convertor` for in-place transformation** — A `convertor` function
can be passed through the read pipeline, applied to each entry
immediately after filtering. This eliminates the need for a secondary
allocation-and-copy pass when entries need to be transformed (e.g.
dropping stats, projecting fields).
---
.../java/org/apache/paimon/manifest/FileEntry.java | 7 ++-
.../org/apache/paimon/manifest/ManifestFile.java | 28 ++++++++--
.../paimon/operation/AbstractFileStoreScan.java | 44 ++++++++-------
.../java/org/apache/paimon/utils/ObjectsCache.java | 23 +++++++-
.../java/org/apache/paimon/utils/ObjectsFile.java | 65 ++++++++++++++++++----
.../org/apache/paimon/utils/ObjectsCacheTest.java | 58 ++++++++++++++++---
6 files changed, 179 insertions(+), 46 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/manifest/FileEntry.java
b/paimon-core/src/main/java/org/apache/paimon/manifest/FileEntry.java
index 3a80088255..3daf423529 100644
--- a/paimon-core/src/main/java/org/apache/paimon/manifest/FileEntry.java
+++ b/paimon-core/src/main/java/org/apache/paimon/manifest/FileEntry.java
@@ -240,7 +240,12 @@ public interface FileEntry {
return readDeletedEntries(
m ->
manifestFile.read(
- m.fileName(), m.fileSize(), deletedFilter(),
Filter.alwaysTrue()),
+ m.fileName(),
+ m.fileSize(),
+ Filter.alwaysTrue(),
+ deletedFilter(),
+ Filter.alwaysTrue(),
+ SimpleFileEntry::from),
manifestFiles,
manifestReadParallelism);
}
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 0ce78f37fc..527568f7f0 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
@@ -47,6 +47,7 @@ import java.io.IOException;
import java.io.UncheckedIOException;
import java.util.ArrayList;
import java.util.List;
+import java.util.function.Function;
/**
* This file includes several {@link ManifestEntry}s, representing the
additional changes since last
@@ -106,18 +107,35 @@ public class ManifestFile extends
ObjectsFile<ManifestEntry> {
@Nullable BucketFilter bucketFilter,
Filter<InternalRow> readFilter,
Filter<ManifestEntry> readTFilter) {
+ return read(
+ fileName,
+ fileSize,
+ partitionFilter,
+ bucketFilter,
+ readFilter,
+ readTFilter,
+ Function.identity());
+ }
+
+ public <T> List<T> read(
+ String fileName,
+ @Nullable Long fileSize,
+ @Nullable PartitionPredicate partitionFilter,
+ @Nullable BucketFilter bucketFilter,
+ Filter<InternalRow> readFilter,
+ Filter<ManifestEntry> readTFilter,
+ Function<ManifestEntry, T> convertor) {
try {
Path path = pathFactory.toPath(fileName);
if (cache != null) {
- return cache.read(
- path,
- fileSize,
+ ManifestEntryFilters filters =
new ManifestEntryFilters(
- partitionFilter, bucketFilter, readFilter,
readTFilter));
+ partitionFilter, bucketFilter, readFilter,
readTFilter);
+ return cache.read(path, fileSize, filters, convertor);
}
return readFromIterator(
- createIterator(path, fileSize), serializer, readFilter,
readTFilter);
+ createIterator(path, fileSize), serializer, readFilter,
readTFilter, convertor);
} catch (IOException e) {
throw new UncheckedIOException(e);
}
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 30908cce3e..a78beae34d 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
@@ -407,11 +407,16 @@ public abstract class AbstractFileStoreScan implements
FileStoreScan {
private <T extends FileEntry> Iterator<T> readAndMergeFileEntries(
List<ManifestFileMeta> manifests,
- Function<List<ManifestEntry>, List<T>> converter,
+ Function<ManifestEntry, T> converter,
boolean useSequential) {
Set<Identifier> deletedEntries =
FileEntry.readDeletedEntries(
- manifest -> readManifest(manifest,
FileEntry.deletedFilter(), null),
+ manifest ->
+ readManifest(
+ manifest,
+ SimpleFileEntry::from,
+ FileEntry.deletedFilter(),
+ null),
manifests,
parallelism);
@@ -422,11 +427,11 @@ public abstract class AbstractFileStoreScan implements
FileStoreScan {
Function<ManifestFileMeta, List<T>> processor =
manifest ->
- converter.apply(
- readManifest(
- manifest,
- FileEntry.addFilter(),
- entry ->
!deletedEntries.contains(entry.identifier())));
+ readManifest(
+ manifest,
+ converter,
+ FileEntry.addFilter(),
+ entry ->
!deletedEntries.contains(entry.identifier()));
if (useSequential) {
return sequentialBatchedExecute(processor, manifests,
parallelism).iterator();
} else {
@@ -476,14 +481,20 @@ public abstract class AbstractFileStoreScan implements
FileStoreScan {
/** Note: Keep this thread-safe. */
@Override
public List<ManifestEntry> readManifest(ManifestFileMeta manifest) {
- return readManifest(manifest, null, null);
+ return readManifest(manifest, Function.identity(), null, null);
}
- private List<ManifestEntry> readManifest(
+ private <T> List<T> readManifest(
ManifestFileMeta manifest,
+ Function<ManifestEntry, T> converter,
@Nullable Filter<InternalRow> additionalFilter,
@Nullable Filter<ManifestEntry> additionalTFilter) {
- List<ManifestEntry> entries =
+
+ Filter<InternalRow> entryRowFilter = createEntryRowFilter();
+ Function<ManifestEntry, T> finalConverter =
+ dropStats ? e -> converter.apply(dropStats(e)) : converter;
+
+ List<T> entries =
manifestFileFactory
.create()
.withCacheMetrics(
@@ -493,19 +504,14 @@ public abstract class AbstractFileStoreScan implements
FileStoreScan {
manifest.fileSize(),
manifestsReader.partitionFilter(),
createBucketFilter(),
- createEntryRowFilter().and(additionalFilter),
+ entryRowFilter.and(additionalFilter),
entry ->
(additionalTFilter == null ||
additionalTFilter.test(entry))
&& (manifestEntryFilter == null
||
manifestEntryFilter.test(entry))
- && filterByStats(entry));
- if (dropStats) {
- List<ManifestEntry> copied = new ArrayList<>(entries.size());
- for (ManifestEntry entry : entries) {
- copied.add(dropStats(entry));
- }
- entries = copied;
- }
+ && filterByStats(entry),
+ finalConverter);
+ LOG.info("Read {} manifest entries from {}", entries.size(),
manifest.fileName());
return entries;
}
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 21b17549b3..bfc06d4b63 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
@@ -28,7 +28,9 @@ import javax.annotation.Nullable;
import javax.annotation.concurrent.ThreadSafe;
import java.io.IOException;
+import java.util.ArrayList;
import java.util.List;
+import java.util.function.Function;
import static org.apache.paimon.utils.ObjectsFile.readFromIterator;
@@ -63,13 +65,19 @@ public abstract class ObjectsCache<K, V, S extends
Segments> {
}
public List<V> read(K key, @Nullable Long fileSize, Filters<V> filters)
throws IOException {
+ return read(key, fileSize, filters, Function.identity());
+ }
+
+ public <R> List<R> read(
+ K key, @Nullable Long fileSize, Filters<V> filters, Function<V, R>
convertor)
+ throws IOException {
@SuppressWarnings("unchecked")
S segments = (S) cache.getIfPresents(key);
if (segments != null) {
if (cacheMetrics != null) {
cacheMetrics.increaseHitObject();
}
- return readFromSegments(segments, filters);
+ return convert(readFromSegments(segments, filters), convertor);
} else {
if (cacheMetrics != null) {
cacheMetrics.increaseMissedObject();
@@ -80,17 +88,26 @@ public abstract class ObjectsCache<K, V, S extends
Segments> {
if (fileSize <= cache.maxElementSize()) {
segments = createSegments(key, fileSize);
cache.put(key, segments);
- return readFromSegments(segments, filters);
+ return convert(readFromSegments(segments, filters), convertor);
} else {
return readFromIterator(
reader.apply(key, fileSize),
projectedSerializer,
filters.readFilter(),
- filters.readVFilter());
+ filters.readVFilter(),
+ convertor);
}
}
}
+ private <R> List<R> convert(List<V> values, Function<V, R> convertor) {
+ List<R> result = new ArrayList<>(values.size());
+ for (V v : values) {
+ result.add(convertor.apply(v));
+ }
+ return result;
+ }
+
protected abstract List<V> readFromSegments(S segments, Filters<V>
filters) throws IOException;
protected abstract S createSegments(K k, @Nullable Long fileSize);
diff --git a/paimon-core/src/main/java/org/apache/paimon/utils/ObjectsFile.java
b/paimon-core/src/main/java/org/apache/paimon/utils/ObjectsFile.java
index 2293a59fff..2ec44afdd5 100644
--- a/paimon-core/src/main/java/org/apache/paimon/utils/ObjectsFile.java
+++ b/paimon-core/src/main/java/org/apache/paimon/utils/ObjectsFile.java
@@ -37,6 +37,7 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
+import java.util.function.Function;
import static org.apache.paimon.utils.FileUtils.checkExists;
@@ -100,7 +101,13 @@ public abstract class ObjectsFile<T> implements
SimpleFileReader<T> {
}
public List<T> read(String fileName, @Nullable Long fileSize) {
- return read(fileName, fileSize, Filter.alwaysTrue(),
Filter.alwaysTrue());
+ return read(
+ fileName,
+ fileSize,
+ Filter.alwaysTrue(),
+ Filter.alwaysTrue(),
+ Filter.alwaysTrue(),
+ Function.identity());
}
public List<T> readWithIOException(String fileName) throws IOException {
@@ -109,7 +116,13 @@ public abstract class ObjectsFile<T> implements
SimpleFileReader<T> {
public List<T> readWithIOException(String fileName, @Nullable Long
fileSize)
throws IOException {
- return readWithIOException(fileName, fileSize, Filter.alwaysTrue(),
Filter.alwaysTrue());
+ return readWithIOException(
+ fileName,
+ fileSize,
+ Filter.alwaysTrue(),
+ Filter.alwaysTrue(),
+ Filter.alwaysTrue(),
+ Function.identity());
}
public boolean exists(String fileName) {
@@ -120,31 +133,51 @@ public abstract class ObjectsFile<T> implements
SimpleFileReader<T> {
}
}
- public List<T> read(
+ public <R> List<R> read(
String fileName,
@Nullable Long fileSize,
+ Filter<InternalRow> loadFilter,
Filter<InternalRow> readFilter,
- Filter<T> readTFilter) {
+ Filter<T> readTFilter,
+ Function<T, R> convertor) {
try {
- return readWithIOException(fileName, fileSize, readFilter,
readTFilter);
+ return readWithIOException(
+ fileName, fileSize, loadFilter, readFilter, readTFilter,
convertor);
} catch (IOException e) {
throw new RuntimeException("Failed to read " + fileName, e);
}
}
- private List<T> readWithIOException(
+ public List<T> read(
String fileName,
@Nullable Long fileSize,
Filter<InternalRow> readFilter,
- Filter<T> readTFilter)
+ Filter<T> readTFilter) {
+ return read(
+ fileName,
+ fileSize,
+ Filter.alwaysTrue(),
+ readFilter,
+ readTFilter,
+ Function.identity());
+ }
+
+ private <R> List<R> readWithIOException(
+ String fileName,
+ @Nullable Long fileSize,
+ Filter<InternalRow> loadFilter,
+ Filter<InternalRow> readFilter,
+ Filter<T> readTFilter,
+ Function<T, R> convertor)
throws IOException {
Path path = pathFactory.toPath(fileName);
if (cache != null) {
- return cache.read(path, fileSize, new
ObjectsCache.Filters<>(readFilter, readTFilter));
+ return cache.read(
+ path, fileSize, new ObjectsCache.Filters<>(readFilter,
readTFilter), convertor);
}
return readFromIterator(
- createIterator(path, fileSize), serializer, readFilter,
readTFilter);
+ createIterator(path, fileSize), serializer, readFilter,
readTFilter, convertor);
}
public String writeWithoutRolling(Collection<T> records) {
@@ -208,14 +241,24 @@ public abstract class ObjectsFile<T> implements
SimpleFileReader<T> {
ObjectSerializer<V> serializer,
Filter<InternalRow> readFilter,
Filter<V> readVFilter) {
+ return readFromIterator(
+ inputIterator, serializer, readFilter, readVFilter,
Function.identity());
+ }
+
+ public static <V, R> List<R> readFromIterator(
+ CloseableIterator<InternalRow> inputIterator,
+ ObjectSerializer<V> serializer,
+ Filter<InternalRow> readFilter,
+ Filter<V> readVFilter,
+ Function<V, R> convertor) {
try (CloseableIterator<InternalRow> iterator = inputIterator) {
- List<V> result = new ArrayList<>();
+ List<R> result = new ArrayList<>();
while (iterator.hasNext()) {
InternalRow row = iterator.next();
if (readFilter.test(row)) {
V v = serializer.fromRow(row);
if (readVFilter.test(v)) {
- result.add(v);
+ result.add(convertor.apply(v));
}
}
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/utils/ObjectsCacheTest.java
b/paimon-core/src/test/java/org/apache/paimon/utils/ObjectsCacheTest.java
index 4061044fb7..cb0c5c7c7f 100644
--- a/paimon-core/src/test/java/org/apache/paimon/utils/ObjectsCacheTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/utils/ObjectsCacheTest.java
@@ -35,7 +35,9 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.function.Function;
+import static org.apache.paimon.utils.ObjectsCache.Filters;
import static org.assertj.core.api.Assertions.assertThat;
/** Test for {@link ObjectsCache}. */
@@ -62,19 +64,34 @@ public class ObjectsCacheTest {
cache.withCacheMetrics(scanMetrics.getCacheMetrics());
// test empty
map.put("k1", Collections.emptyList());
- List<String> values = cache.read("k1", null, Filter.alwaysTrue(),
Filter.alwaysTrue());
+ List<String> values =
+ cache.read(
+ "k1",
+ null,
+ new Filters<>(Filter.alwaysTrue(),
Filter.alwaysTrue()),
+ Function.identity());
assertThat(values).isEmpty();
assertThat(scanMetrics.getCacheMetrics().getMissedObject()).hasValue(1);
// test values
List<String> expect = Arrays.asList("v1", "v2", "v3");
map.put("k2", expect);
- values = cache.read("k2", null, Filter.alwaysTrue(),
Filter.alwaysTrue());
+ values =
+ cache.read(
+ "k2",
+ null,
+ new Filters<>(Filter.alwaysTrue(),
Filter.alwaysTrue()),
+ Function.identity());
assertThat(values).containsExactlyElementsOf(expect);
assertThat(scanMetrics.getCacheMetrics().getMissedObject()).hasValue(2);
// test cache
- values = cache.read("k2", null, Filter.alwaysTrue(),
Filter.alwaysTrue());
+ values =
+ cache.read(
+ "k2",
+ null,
+ new Filters<>(Filter.alwaysTrue(),
Filter.alwaysTrue()),
+ Function.identity());
assertThat(values).containsExactlyElementsOf(expect);
assertThat(scanMetrics.getCacheMetrics().getHitObject()).hasValue(1);
@@ -83,10 +100,35 @@ public class ObjectsCacheTest {
cache.read(
"k2",
null,
- r -> r.getString(0).toString().endsWith("2"),
- Filter.alwaysTrue());
+ new Filters<>(
+ r -> r.getString(0).toString().endsWith("2"),
Filter.alwaysTrue()),
+ Function.identity());
assertThat(values).containsExactly("v2");
+ // test filter with loadFilter semantics moved to readFilter
+ expect = Arrays.asList("v1", "v2", "v3");
+ map.put("k3", expect);
+ values =
+ cache.read(
+ "k3",
+ null,
+ new Filters<>(
+ r -> r.getString(0).toString().endsWith("2"),
Filter.alwaysTrue()),
+ Function.identity());
+ assertThat(values).containsExactly("v2");
+
+ // test filter empty
+ expect = Arrays.asList("v1", "v2", "v3");
+ map.put("k4", expect);
+ values =
+ cache.read(
+ "k4",
+ null,
+ new Filters<>(
+ r -> r.getString(0).toString().endsWith("5"),
Filter.alwaysTrue()),
+ Function.identity());
+ assertThat(values).isEmpty();
+
// test read concurrently
map.clear();
for (int i = 0; i < 10; i++) {
@@ -101,8 +143,10 @@ public class ObjectsCacheTest {
cache.read(
k,
null,
- Filter.alwaysTrue(),
- Filter.alwaysTrue()))
+ new Filters<>(
+
Filter.alwaysTrue(),
+
Filter.alwaysTrue()),
+ Function.identity()))
.containsExactly(k);
} catch (IOException e) {
throw new RuntimeException(e);
