perf(scan): prune manifest partitions before metadata decoding

[gripleaf]

Purpose

Linked issue: Closes #433.

Apply partition selection before full manifest-entry decoding across ordinary manifest scans, manifest Arrow-cache reads, and exact snapshot-cache hits. Scans selecting a small subset of partitions can avoid constructing irrelevant ManifestEntry / DataFileMeta objects and decoding their statistics, whether or not caching is enabled.

Scenario. A partitioned, bucketed Paimon table serves repeated point reads. An application batches 50 requested cells, grouping cells for the same key. Each underlying lookup uses a key equality predicate and a fixed date-partition filter, allowing the planner to select one bucket. The bucket-scoped manifest cache, however, contains live files from many date partitions. Reusing that cache currently decodes the full file metadata before discarding entries outside the requested partition.

For example, one sampled bucket contains 604 live entries across 84 partitions, but only six entries belong to the queried partition:

Before: deserialize 604 full entries -> apply partition filter -> keep 6
After:  inspect 604 partition fields -> apply partition filter -> fully decode 6

ManifestFile::ReadSelectedEntries extends the existing row-range reader to combine row-range and partition selection before deserialization in one scan loop. It uses the existing ReadArrowBatches flow for both disk and Arrow-cache reads. When neither row-range nor partition selection is needed, it delegates to the existing ObjectsFile::Read path. Callers supply every argument explicitly, using nullptr for a selection they do not need. The generic ObjectsFile::Read interface and its other consumers are unchanged.

FileStoreScan supplies the same partition filter to this reader and to snapshot selection. Both readers validate the entry version before early field access or invoking the row filter. ManifestEntrySerializer::GetPartitionView borrows partition bytes for synchronous filtering; fully decoded entries retain owned partition data. Entries already selected by partition run only the remaining post-decode filters. ADD and DELETE entries receive the same selection before merging. Existing bucket reader pushdown is preserved.

On an exact snapshot-cache hit, ReadSnapshot selects that retained snapshot and skips other snapshots by serialized row length without allocating their row payloads. It returns a request-local selection; full Deserialize remains available for cache rebuilding and updates. The shared Arrow cache retains complete batches, and snapshot cache misses rebuild and store the complete bucket before query filtering, so later scans can safely query another partition. Existing snapshot retention and normal miss/rebuild behavior are preserved. Each scan looks up the snapshot cache once. Snapshot selection distinguishes predicate failures from cache parsing failures. Predicate failures propagate to CreatePlan without an internal retry. Malformed cache values are treated as misses and rebuilt from the manifests through the existing miss path. A successful read with no retained snapshot uses full deserialization only to retain historical snapshots before rebuilding the target snapshot; a present selection is a cache hit. Error provenance is recorded only at partition_filter_->Test(), so a failure to decode partition bytes remains a cache parsing failure, even though both sources can return Status::Invalid. Unfiltered snapshot reads reserve the known entry count; filtered reads grow with the selected results.

The existing scan.manifest-entry.lazy-decode.enabled option controls partition selection before metadata decoding on all three paths. Disabling it preserves full entry decoding before partition filtering. The implementation still traverses serialized snapshot payloads and loads aligned batches on disk/Arrow-cache paths; it reduces entry-object and statistics materialization. Scans selecting most partitions are not expected to gain the same benefit.

Component measurements (not end-to-end query latency). A standalone prototype harness compared full deserialization followed by partition filtering against selective decoding, using the same serialized cache payloads from four real buckets. Each bucket had 84 partitions; each comparison selected one partition. Each path ran 200 iterations per pair, with five pairs and alternating execution order, on the same host. The table reports mean elapsed time per iteration across those pairs. Both paths returned exactly equal selected entries.

Sample	Cached entries	Selected entries	Before (ms)	After (ms)	Time reduction
A	604	6	4.566	0.839	81.6%
B	616	6	4.663	0.849	81.8%
C	566	5	4.282	0.774	81.9%
D	604	5	4.593	0.827	82.0%

These measurements motivated the upstream port; they are prototype measurements, not measurements of this refactored PR or its disk/Arrow-cache paths. Full cold/warm end-to-end performance and long-running concurrent behavior have not yet been benchmarked for this upstream build.

Tests

All 25 tests in the focused scan suites passed, followed by the complete core suite (2356 passed). The complete common suite also passed (1725 passed, 2 skipped); the skips are the existing prefetch warmup cases without a read-ahead cache to rearm.

cmake --build build --target paimon-core-test paimon-common-test -j 32
./build/debug/paimon-core-test --gtest_filter='AppendOnlyFileStoreScanTest.*:FileStoreScanTest.*'
./build/debug/paimon-core-test
./build/debug/paimon-common-test
pre-commit run --files $(git diff --name-only apache/main)
git diff --check

All applicable pre-commit checks passed for the PR's 19 changed files. Release, sanitizer, integration suites, and end-to-end performance benchmarks were not run.

Regression coverage:

• RowRangeManifestFileTest.PartitionFilterComposesWithRowRanges exercises disk and Arrow-cache reads, partition/range intersection, ADD/DELETE merging, filter errors, and skipped metadata decoding. SchemaEvolutionAndVersionValidation also checks that an unsupported version is rejected before the row filter is invoked.
• AppendOnlyFileStoreScanTest.PartitionPruningAcrossCacheModes covers uncached, Arrow-cache, and snapshot-cache reads with lazy decoding enabled and disabled. It checks partition results, one partition predicate evaluation per candidate, cache reuse, single snapshot-cache lookups, empty results, and later unfiltered reads.
• PartitionPruningSkipsExcludedMetadataInCreatePlan proves the scan wires early partition selection into each read path: an invalid file kind in an excluded partition is skipped with lazy decoding enabled and rejected when it is disabled.
• SnapshotCachePropagatesPartitionPredicateError injects one-time IOError and Invalid predicate failures and checks that CreatePlan propagates their codes and messages without internally retrying, with lazy decoding enabled and disabled. A subsequent caller retry succeeds using the complete cached snapshot.
• SnapshotCacheRebuildsCorruptValues injects bad magic, unsupported entry versions, truncated rows, and invalid partition bytes into the snapshot cache. With lazy decoding enabled and disabled, the first scan rebuilds and matches an uncached scan, the second scan hits the repaired cache, and an unfiltered scan still returns the complete bucket.
• TestSnapshotLiveManifestCacheRebuildOnMiss checks historical snapshot retention and reduced retention limits configured through ScanContextBuilder.
• The unused LatestBeforeOrEqual / Entry interface has been removed. Snapshot tests retain serialization round-trip and eviction coverage through the supported APIs. Snapshot selection tests cover exact hits/misses, retention, empty snapshots, allocation-free skipping of other snapshots, malformed headers, filter errors, and version validation before filtering.
• Serializer tests cover borrowed partition views, owned result lifetime, unaligned buffers, fixed-field bounds, and skipping length-prefixed rows. Existing row-range, schema-evolution, cache reuse, and concurrency tests exercise the unified manifest reader.

API and Format

No public API symbols, metric names, serialized cache format, table storage format, or protocols are added or changed. The existing lazy-decode option documentation describes its broader scope. ReadSelectedEntries and exact snapshot selection are internal APIs. Snapshot selection returns its pre-filter candidate count to preserve the meaning of existing lazy-scan metrics.

The private shared SerializationUtils decoder now also rejects binary rows whose declared arity requires more fixed-field bytes than the payload contains, using overflow-safe bounds checks. This tightens validation for both borrowed and owned inputs, including existing consumers such as file keys, statistics, splits, and commit messages. Well-formed serialized rows and their encoding are unchanged. The check remains shared rather than being limited to partition views.

Predicate evaluation failures during selective snapshot reading propagate to the caller. Errors parsing cached values, including malformed headers, unsupported entry versions, truncated payloads, and invalid partition bytes, retain the existing recovery behavior: rebuild the complete bucket from manifests and write it back to the cache. Normal snapshot misses retain historical snapshots through full deserialization. Cache lookup failure handling is unchanged.

Documentation

Updated manifest_entry_cache.rst with selective decoding across read sources, row-range composition, option behavior, the original motivating example, and cache completeness guarantees. Updated the existing option comment in include/paimon/defs.h.

Generative AI tooling

Generated-by: OpenAI Codex (codex-cli 0.162.1)