[core] Add manifest sidecars for partition, row-id and bucket pruning

[leaves12138]

Status

This is the tracking draft for the remaining manifest-sidecar integration. Rebased onto master after #9889 merged (including master commit 1b802b9b76). The standalone Java format, codec and golden file now come from master unchanged.

The format (#9845), lifecycle (#9886), and Java writer (#9889) PRs are merged. The next extracted PR, #9897, covers Java scan pruning. PyPaimon integration remains in this tracking draft for subsequent review.

Purpose

Use optional manifest sidecars to prune physical Avro blocks by independent partition, row-ID and bucket coverage before normal entry filtering and ADD/DELETE reconciliation.

• Integrate sidecar generation with rolling and rewritten manifests. Publish explicit _EXTRA_FILES references only after the immutable sidecar closes successfully.
• Control both reading and writing through manifest.sidecar.enabled, inheriting manifest-sort.enabled when unset. Java and PyPaimon use the same enablement semantics.
• Always generate partition coverage, including the empty tuple. Row-ID generation follows data evolution; bucket coverage is enabled when the table bucket option is not -1.
• Use master's version-1 layout: four-byte PMSC magic, canonical unsigned varint framing, sorted delta-encoded coverage, fixed row-ID min/max, paired bucket/total-bucket sequences with ZigZag total deltas, and a four-byte big-endian CRC32. No manifest-name hash or sidecar size budget remains.
• Adapt both PyPaimon writing and reading to the same layout. Java and Python serialize the same approximately 2 KiB golden file byte-for-byte, with no separate legacy format path.
• Retain referenced files across snapshot/tag/changelog retention and orphan cleanup. Failed commits and merges remove only their newly created manifests and referenced extra files. Shared references remain protected.
• Java block/sidecar reads share the manifest byte cache using separate key types. PyPaimon retains its existing metadata-cache integration and complete explain counters.

Validation

• Java 8: 312 tests passed; one pre-existing disabled test was skipped. Normal Maven checks passed, and micro-benchmark test sources compiled against the rebased API.
• Python: 222 manifest, scan/explain, cache and bucket-pruning tests passed. The randomized block-coverage test also checks exact selected-block sets against an entry-level oracle.
• Python lint passed using the repository configuration.
• The standalone Java sidecar implementation and shared golden file are unchanged from master.

mvn -B -ntp -nsu -pl paimon-benchmark/paimon-micro-benchmarks -am \
  -DwildcardSuites=none -DfailIfNoTests=false -Dflink.forkCount=2 \
  '-Dtest=CoreOptionsTest,ManifestSidecarWriteTest,ManifestFileTest,ManifestSidecarTest,ManifestBlockIndexTest,ManifestFileMetaTest,NoPartitionManifestFileMetaTest,ManifestRewriteCleanupTest,FileStoreCommitTest#testCommitManifestMerge+testManifestSortCompactManifestRespectsCompactionThresholds' test
PYTHONPATH=paimon-python python3 -m pytest paimon-python/pypaimon/tests/manifest \
  paimon-python/pypaimon/tests/read_builder_explain_test.py \
  paimon-python/pypaimon/tests/file_type_test.py \
  paimon-python/pypaimon/tests/caching_file_io_test.py \
  paimon-python/pypaimon/tests/pushdown_bucket_test.py -q

[JingsongLi]

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.

[JingsongLi]

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.

[JingsongLi]

We also need to consider adding bucket and totalBucket to accelerate point lookup performance for buckets; bucket and rowId are mutually exclusive types of metadata.

[leaves12138]

We also need to consider adding bucket and totalBucket to accelerate point lookup performance for buckets; bucket and rowId are mutually exclusive types of metadata.

OK, I will add bucket and totalBucket to each block preview