Skip to content

[core] Optimize partition discovery with projected manifest scans #9321

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
zhoulii wants to merge 2 commits into
base: master
Choose a base branch
from
Open

[core] Optimize partition discovery with projected manifest scans #9321

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
1ade821
[core] Optimize partition discovery with projected manifest scans
zhoulii Aug 20, 2026
587b8d3
[core] Centralize projected manifest scan selection
zhoulii Aug 21, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 19 additions & 2 deletions paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFile.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -157,14 +157,26 @@ public <T> List<T> read(
* materialized with the complete manifest schema.
*/
public CloseableIterator<ProjectedManifestEntry> 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<ProjectedManifestEntry> scan(
String fileName,
Projection projection,
@Nullable PartitionPredicate partitionFilter,
@Nullable BucketFilter bucketFilter) {
try {
CloseableIterator<InternalRow> rows =
createManifestIterator(
fileIO,
pathFactory.toPath(fileName),
projection.projectedType(),
null,
null);
partitionFilter,
bucketFilter);
return new CloseableIterator<ProjectedManifestEntry>() {

@Override
Expand Down Expand Up @@ -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) {

@JingsongLi JingsongLi Aug 21, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why not just use isCacheEnabled?

@zhoulii zhoulii Aug 21, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

isCacheable(fileSize) is intentional. When a manifest exceeds maxElementSize, ObjectsCache bypasses it even if caching is enabled. Using isCacheEnabled() would unnecessarily disable projected scans for
these large, non-cacheable manifests.

@JingsongLi JingsongLi Aug 23, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why we need to bypass it? Could you describe this scenario in more detail?

@zhoulii zhoulii Aug 24, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do you mean why cached reads cannot use projection as well?

Currently, projection is only supported by the file reader, while the cache stores and materializes full manifest entries. This change preserves the existing full-cache path for cacheable manifests and applies file-level projection only when the manifest cannot be cached.

Cache-side projected materialization could be added as a separate optimization. Is that what you are suggesting?

@JingsongLi JingsongLi Aug 24, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

When does the manifest size exceed the maxElementSize?

@zhoulii zhoulii Aug 24, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

With the default caching catalog settings, maxElementSize is 1 MB (cache.manifest.small-file-threshold), while manifest.target-file-size defaults to 8 MB. Therefore, normal manifest files can exceed maxElementSize.

return cache != null && fileSize <= cache.maxElementSize();
}

public ManifestFile create() {
return new ManifestFile(
fileIO,
Expand Down
107 changes: 103 additions & 4 deletions paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand All @@ -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;
Expand All @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -367,10 +373,19 @@ public List<PartitionEntry> readPartitionEntries() {
List<ManifestFileMeta> manifests = readManifests().filteredManifests;
Map<BinaryRow, PartitionEntry> partitions = new ConcurrentHashMap<>();
Consumer<ManifestFileMeta> processor =
m ->
PartitionEntry.merge(
readManifest(m, 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)
Expand Down Expand Up @@ -476,6 +491,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;
}
Expand All @@ -490,6 +514,55 @@ public List<ManifestEntry> readManifest(ManifestFileMeta manifest) {
return readManifest(manifest, Function.identity(), null, null);
}

private <T> void readManifest(
ManifestFileMeta manifest,
Function<ManifestEntry, T> converter,
ProjectedManifestEntry.Projection projection,
Filter<ProjectedManifestEntry> projectedFilter,
Consumer<T> 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<ProjectedManifestEntry> 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 <T> List<T> readManifest(
ManifestFileMeta manifest,
Function<ManifestEntry, T> converter,
Expand Down Expand Up @@ -578,6 +651,32 @@ private Filter<InternalRow> 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
// ------------------------------------------------------------------------
Expand Down
5 changes: 5 additions & 0 deletions paimon-core/src/main/java/org/apache/paimon/operation/AppendOnlyFileStoreScan.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down
5 changes: 5 additions & 0 deletions paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionFileStoreScan.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down
5 changes: 5 additions & 0 deletions paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreScan.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
2 changes: 2 additions & 0 deletions paimon-core/src/test/java/org/apache/paimon/catalog/CachingCatalogTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 -> {});
Expand Down
109 changes: 109 additions & 0 deletions ...core/src/test/java/org/apache/paimon/operation/FileStoreScanPartitionBucketEntryTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -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<ManifestEntry> files = table.store().newScan().plan().files();
assertThat(files).hasSize(2);
ManifestEntry selected =
files.stream().filter(file -> file.file().rowCount() == 5).findFirst().get();

List<PartitionEntry> 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<PartitionEntry> 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<PartitionEntry> 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<PartitionEntry> 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
Expand Down Expand Up @@ -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<CommitMessage> messages = write.prepareCommit();
TableCommitImpl commit = table.newCommit(commitUser);
commit.commit(messages);
write.close();
commit.close();
}
}
Loading