Skip to content

Navigation Menu

Sign in
Sign up

[core] Prune manifest blocks with row-id sidecar indexes - #9743

Draft
leaves12138 wants to merge 5 commits into
apache:master from
leaves12138:codex/manifest-row-id-block-index
Draft

[core] Prune manifest blocks with row-id sidecar indexes #9743
leaves12138 wants to merge 5 commits into
apache:master from
leaves12138:codex/manifest-row-id-block-index

Conversation

@leaves12138

@leaves12138 leaves12138 commented Sep 11, 2026
edited
Loading

Copy link
Copy Markdown
Contributor

Purpose

Row-ID filters can retain manifests whose coarse min/max overlaps the query even when only a few Avro blocks contain relevant entries. Add optional row-id sidecars that select complete blocks before the existing manifest entry filtering and ADD/DELETE merge.

  • Store block byte ranges, entry ordinals, and bounded row-id interval unions, covering ADD/DELETE entries and all column groups. Use block min/max before exact interval matching.
  • Use the _EXTRA_FILES manifest metadata introduced by [core][python] Add extra files to manifest metadata #9746 in Java and PyPaimon, without adding a dedicated index field. Identify .row-id-index sidecars only among explicit extra-file references and preserve unrelated extra files. Null, empty, and other-extra-only metadata do not probe for an index.
  • Validate the sidecar version, completeness, manifest identity, lengths and checksum. Missing, invalid or over-budget indexes fall back to the original read path. Cancellations (including PyArrow), interruptions and fatal errors propagate even when wrapped or suppressed during stream close; ordinary I/O failures still fall back. Partial reads bypass the full-manifest cache.
  • Publish the reference after the sidecar closes, and follow explicit references through rolling/rewrite, failed-commit cleanup, snapshot/tag/changelog retention and orphan cleanup.

Both manifest.row-id-index.write and manifest.row-id-index.read default to false. Java builds each index by projecting the final manifest after it closes; existing immutable manifests are not backfilled.

Tests

Java 8, with normal Maven checks enabled after merging master: 153 core tests passed.

mvn -pl paimon-core -am -DwildcardSuites=none -DfailIfNoTests=false \
 '-Dtest=CoreOptionsTest,ManifestFileMetaSerializerTest,ManifestListTest,ManifestFileTest,ManifestRowIdIndexTest,ManifestFileMergerTest,ManifestRewriteCleanupTest,ExpireSnapshotsTest#testSidecarsFollowSnapshotAndTagRetention+testCleanUnusedManifestExtraFiles+testCleanManifestExtraFilesFollowsManifestRetention,LocalOrphanFilesCleanTest#testOrphanCleanupProtectsReferencedSidecars+testPreservesManifestExtraFiles' test

Python: 113 tests passed, 4 skipped.

PYTHONPATH=paimon-python python3 -m pytest \
 paimon-python/pypaimon/tests/reader_append_only_test.py \
 paimon-python/pypaimon/tests/manifest \
 paimon-python/pypaimon/tests/file_store_commit_test.py -q

Coverage includes shared Java/Python format fixtures, 64-bit interval boundaries, block offsets and ordinals, selective body reads, ADD/DELETE correctness, legacy metadata compatibility, explicit suffixed index names, mixed extra-file references, no-index/no-query behavior, conservative fallback, rolling/rewrite failures, commit cleanup, snapshot/tag protection and orphan collection. The manifest tests also retain the upstream cache-size and bucket-filter regression coverage.

Additional local checks passed: 18,000 randomized interval queries across 3,000 manifests; Java-to-Python-to-Java manifest-list interoperability for five extra-file variants; direct and wrapped Java cancellation at open/read/close. Cancellation tests also cover close errors masking earlier failures, suppressed interruption/fatal errors and cyclic exception graphs.

Add optional bounded block indexes and Java/PyPaimon pruning. Publish explicit index references in manifest metadata and preserve them through serialization, rewrites, commit cleanup, snapshot retention, and orphan collection.
leaves12138 marked this pull request as draft September 11, 2026 09:54
Use bounded 1 MiB read requests in Java and Python to avoid object-store request amplification. Merge adjacent selected blocks into spans and buffer Java reads independently of the Avro consumer read size.
Add regression tests for request counts, skipped gaps, short reads, size budgets, stream closure and truncated inputs.

Copy link
Copy Markdown
Contributor

I suggest using a single manifest index sidecar organized by Avro block. Partition information would support partition predicate pushdown during planning. Since both partition and row-id information describe the same blocks, they can live in the same block record and share its physical location.

A possible layout is:

Header
 formatVersion
 manifest identity (name hash, file length, entry count)
 original Avro header
Partition dictionary
 partitionId -> complete partition tuple
blockCount : int
BlockIndexRecord[] // original manifest order
 offset : long // byte offset in the manifest
 length : long // complete Avro block length
 recordCount : long // number of manifest entries
 flags : byte // independent availability bits
 [if ROW_ID_AVAILABLE]
 rangeCount : int
 ranges : (start: long, end: long)[] // inclusive interval unions
 [if PARTITION_AVAILABLE]
 partitionIdCount : int
 partitionIds : int[] // sorted and deduplicated
Checksum of all preceding bytes

The partition dictionary is shared across the file and can reuse the existing manifest partition encoding, preserving full tuples, types and nulls. Each block only stores dictionary IDs. The block ID is implicit in its position; firstRecord can be derived from preceding recordCount values.

The two indexes should remain independently usable within each block:

  • An availability bit means that the corresponding information completely covers the block's entries, including both ADD and DELETE entries and all column groups.
  • If row-id coverage is unknown or exceeds its budget, omit that block's row-id payload while retaining its partition information. Apply the same rule independently to partition information.
  • An unavailable index means "cannot prune using this index," rather than an empty result. Invalid file metadata or a checksum failure should fall back to the normal manifest read.

During planning, evaluate the partition predicate against the dictionary once, then check each block's partition IDs and row-id intervals. For conjunctive filters, intersect their candidate block sets. Read the selected blocks and retain the existing entry filtering and ADD/DELETE merge, since block-level matches do not guarantee that the same entry satisfies both predicates.

This layout assumes reading the whole sidecar, as the current implementation does. A partition-only query would also read the row-id index bytes. I would start with this simpler layout and consider separate physical sections if measurements show that selective index reads materially improve planning time.

Forward selected_blocks through the append-only reader test wrapper. Fix the manifest target size and assert explicit retained and expired manifest sets so snapshot and tag retention coverage does not depend on randomized file sizes.
Merge current master and adopt its manifest extra-files metadata instead of a dedicated index-file-name field. Discover row-id indexes through explicit suffixed references and preserve other extra files across reads and cleanup.
Verify Java/Python compatibility, mixed extra-file references, retention and failed-commit cleanup. Java core: 151 tests passed. Python: 106 passed, 4 skipped. Random interval and byte checks: 18000 queries passed.
Propagate PyArrow cancellations and inspect chained and suppressed failures before falling back to full manifests. Preserve Java interruption state and fatal failures, guard against exception cycles, and cover stream open/read/close behavior with regression tests.

Copy link
Copy Markdown
Contributor

Here is a refined version of the block-oriented layout, keeping the file-level partition dictionary and making each block's two payloads independently extensible.

Header
 magic
 formatVersion
 manifest identity (name hash, file length, entry count)
 avroHeaderLength : int
 original Avro header : bytes
Partition Dictionary
 partitionCount : int
 entries[] // position is the partition ID
 partitionByteLength : int
 partitionBytes : bytes
blockCount : int
BlockIndexRecord[] // original manifest order
 offset : long // byte offset in the manifest
 length : long // complete Avro block length
 recordCount : long // number of manifest entries
 partitionEncoding : byte
 partitionPayloadLength : int
 partitionPayload : bytes
 rowIdEncoding : byte
 rowIdPayloadLength : int
 rowIdPayload : bytes
Checksum of all preceding bytes

The dictionary stores each complete partition tuple once, using the existing manifest partition serialization. This preserves tuple values and nulls; the scan's existing partitionType supplies their interpretation. Blocks reference dictionary IDs. The block ID is implicit in its position, and firstRecord is derived from preceding entry counts.

The encoding bytes identify how to decode the corresponding payload, with separate ID namespaces for partition and row-id payloads. They replace the availability flags:

Field Encoding Meaning and payload
partitionEncoding 0 Partition coverage is unavailable. Payload length must be zero.
partitionEncoding 1 Complete partition ID set: partitionIdCount: int, followed by that many sorted, unique partitionId: int values. Every ID references the file-level dictionary.
rowIdEncoding 0 Row-id coverage is unavailable. Payload length must be zero.
rowIdEncoding 1 Conservative interval coverage: rangeCount: int, followed by that many inclusive (start: long, end: long) pairs, sorted and disjoint.

The container's integers and the encoding-1 payload integers use fixed-width big-endian representation; partition bytes retain their existing serialization. Encoding bytes are interpreted as unsigned IDs. Each payload length counts only its payload bytes, excluding the encoding and length fields.

Other nonzero encoding IDs are reserved for future representations. If a reader does not recognize one, it skips exactly that payload length and treats that dimension as unavailable, while still being able to use the other dimension. Lengths must be bounded and validated. The outer formatVersion governs the container and dictionary framing; unsupported container versions or malformed metadata/payloads fall back to the normal manifest read.

For example, rowIdEncoding=1 with rangeCount=2 and ranges [100,109], [300,309] has a 36-byte payload: 4 + 2 * 16.

There are several important correctness and budget rules:

  • Encoding 0 means "cannot prune using this information," never "no matches." An available payload must cover all relevant entries in the block, including ADD, DELETE and all column groups.
  • Row-id coverage may be a conservative superset. If exact interval unions exceed the budget, merge intervals; the coarsest representation is rangeCount=1, [min,max], still using encoding 1. Continue processing the entire block to extend the bounds and detect unknown row IDs. If complete coverage cannot be established, use encoding 0.
  • Partition information can independently become unavailable when its budget is exceeded. Consequently, the global dictionary is not necessarily a complete list of partitions touched by the manifest. A dictionary miss must not eliminate blocks with unavailable partition coverage.
  • The physical block directory must always cover the entire manifest. Budget exhaustion may omit optional index payloads, but must never omit block descriptors. Validate byte coverage and entry counts, and verify the whole-file checksum before making pruning decisions.

For conjunctive partition and row-id filters, select each block using:

keepBlock =
 (partition coverage unavailable || partition predicate matches)
 &&
 (row-id coverage unavailable || query intersects indexed ranges)

Only an empty candidate block set permits skipping the manifest. Selected blocks still pass through the existing entry filtering and ADD/DELETE merge.

This keeps one sidecar and one record per block. It still assumes a bounded whole-sidecar read: payload lengths allow skipping decoding and unknown encodings, but do not by themselves save storage I/O. Index size, block selectivity and planning latency should determine whether selective physical reads are worthwhile later.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

No reviews

Assignees

No one assigned

Labels

None yet

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

AltStyle によって変換されたページ (->オリジナル) /