[core] Add standalone manifest sidecar format and utilities

docs/docs/concepts/spec/manifest.md

@@ -63,6 +63,242 @@ skip manifests before opening them.
 Each extra file belongs exclusively to one manifest. It is retained and cleaned up together with
 that manifest during snapshot, tag, or changelog deletion.
 
+### Manifest Sidecar
+
+`ManifestSidecar` provides a binary sidecar for selecting complete Avro manifest blocks
+using independent partition, row-ID and bucket coverage. A sidecar uses the
+`<manifest-file-name>.avro.sidecar` naming convention. Readers find it through an explicit
+`.avro.sidecar` reference in the manifest metadata's `_EXTRA_FILES`, without probing a
+derived file name. The Avro schemas and `_VERSION` identifiers remain unchanged.
+
+The utility includes construction, validation, block selection and optional caching. Table
+writers and scans do not yet invoke it automatically. Callers are responsible for publishing
+sidecar references, managing file ownership, applying entry filters and reconciling ADD/DELETE
+entries after block selection. `build` reads the completed physical manifest and returns
+sidecar bytes; it does not write or publish another file.
+
+Callers decide whether to invoke `build` and `read`; these utilities have no read/write switches.
+`build` and `Builder` accept `rowIdEnabled` and `bucketEnabled` arguments for independent
+payload generation. Partition generation is always enabled,
+including the empty partition tuple for unpartitioned tables. Missing or invalid
+metadata makes only the affected block's dimension unavailable. There is no sidecar byte budget:
+construction keeps complete coverage and `read` consumes the entire file once it is opened.
+
+`read` returns null for an absent sidecar reference or an `IOException`, allowing the caller
+to fall back to the manifest. If the thread is interrupted, the I/O failure is propagated as
+`UncheckedIOException`. Other exceptions and errors propagate unchanged. `select` validates
+supplied bytes directly and reports invalid containers with `IOException`.
+
+Version 1 uses the following layout. Counts, lengths, offsets and the version use canonical
+nonnegative unsigned LEB128 varints. Counts and payload lengths are bounded by `Integer.MAX_VALUE`;
+block offsets, lengths and record counts are bounded by `Long.MAX_VALUE`. Row-ID envelope
+endpoints remain fixed-width, eight-byte big-endian longs. Encoding IDs are unsigned bytes
+with separate namespaces. The existing serialized partition tuple bytes are unchanged.
+
+```text
+magic : 4 bytes                         // ASCII PMSC
+formatVersion : varint                 // 1
+avroHeaderLength : varint
+avroHeader : bytes                      // original schema, codec and sync marker
+partitionCount : varint
+partitionDictionary[]
+  partitionByteLength : varint
+  partitionBytes : bytes                // existing manifest BinaryRow serialization
+blockCount : varint
+blocks[]                               // original physical order
+  offset : varint
+  length : varint                       // complete encoded block, including sync marker
+  recordCount : varint
+  partitionEncoding : byte
+  if partitionEncoding != 0:
+    partitionPayloadLength : varint
+    partitionPayload : bytes
+  rowIdEncoding : byte
+  if rowIdEncoding != 0:
+    rowIdPayloadLength : varint
+    rowIdPayload : bytes
+  bucketEncoding : byte
+  if bucketEncoding != 0:
+    bucketPayloadLength : varint
+    bucketPayload : bytes
+checksum : 4 bytes                      // big-endian CRC32 of all preceding bytes
+```
+
+The block ID is its position. Its first entry ordinal is the sum of preceding record counts
+and is not stored. Each complete partition tuple appears once in the dictionary, including
+all its fields and nulls. The scan's partition type interprets the existing serialized tuple.
+Partition predicates are evaluated once per dictionary entry.
+
+| Dimension | Encoding | Payload |
+| --- | --- | --- |
+| Any | `0` | Unavailable; only the encoding byte is present. |
+| Partition | `1` | `intsDeltaPayload` of sorted unique dictionary IDs. |
+| Row ID | `1` | Minimum, maximum, and `intsDeltaPayload` of sorted interior interval endpoints. |
+| Bucket | `1` | Two paired `intsDeltaPayload` sequences: sorted bucket IDs and their recorded total bucket counts. |
+| Any | Other nonzero ID | Skip the declared payload length; treat only this dimension as unavailable. |
+
+Only nonzero encodings are followed by a length and payload. Payload lengths exclude the
+encoding and length fields, but include the count and other fields within the payload.
+Partition IDs and bucket pairs have positive counts no greater than the block's record count.
+Row-ID coverage contains one or more intervals; its interior endpoint count can be zero for
+a single interval. Encoding 0 represents unavailable coverage, not an empty known set.
+
+#### Integer Delta Payload
+
+Partition, row-ID and both bucket sequences share this structure:
+
+```text
+intsDeltaPayload
+  count : varint
+  deltas[count] : varint
+```
+
+The count is in `[0, Integer.MAX_VALUE]`. Nonnegative deltas use unsigned LEB128 varints,
+occupying one to nine bytes for values from 0 through
+`Long.MAX_VALUE`. Seven value bits are stored per byte, least significant group first; the
+high bit indicates another byte follows.
+Encodings use the shortest representation without padding.
+
+A nondecreasing sequence is delta-encoded from a specified base. Each value contributes one
+unsigned varint containing its difference from the preceding value. The first difference
+is relative to the base:
+
+```text
+deltas[] : varint
+value[0] = base + deltas[0]
+value[i] = value[i - 1] + deltas[i]
+```
+
+The shared `DeltaVarintCodec` utility writes the count and then each delta immediately,
+and reads values on demand using `VarLengthIntUtils`. A reader consumes exactly the declared
+number of values, leaving any following sequence available in the buffer. Callers check
+their enclosing payload boundaries. Counts, overflow and value bounds are checked without
+materializing arrays. Reads may stop early.
+
+Only the `totalBuckets` sequence uses signed differences, because totals need not increase
+when pairs are sorted by bucket. Its differences use ZigZag before unsigned varint encoding:
+`encoded = (delta << 1) ^ (delta >> 63)` and
+`delta = (encoded >>> 1) ^ -(encoded & 1)`. Values are nonnegative ints, so encoded deltas
+are at most `2 * Integer.MAX_VALUE` and require at most five bytes. The field defines this
+signed mode; no additional mode byte is stored. Other sequences use nonnegative differences.
+
+#### Partition Payload
+
+When `partitionEncoding == 1`, the block stores IDs of all distinct partition tuples
+represented by its entries:
+
+```text
+partitionPayload
+  intsDeltaPayload                    // dictionary IDs, base = 0
+```
+
+An ID is the zero-based position of a complete tuple in the sidecar's shared dictionary.
+IDs satisfy `0 <= id < partitionCount` and are strictly increasing. Tuple bytes appear only
+in the dictionary and are not repeated in each block. For IDs `[0, 1, 2, 3, 4]`, the deltas
+are `[0, 1, 1, 1, 1]`. The payload contains a one-byte count of 5 followed by these five
+varint bytes: 6 bytes, or 8 bytes including the encoding and length fields.
+
+With a partition filter, a block matches if any referenced tuple matches. A tuple containing
+a null partition value still has a dictionary ID. Unpartitioned tables record the empty
+tuple. If any entry's entire partition tuple is unavailable, the block uses encoding 0,
+so a dictionary miss cannot exclude that block. Later blocks can still use existing IDs.
+
+#### Row-ID Payload
+
+The writer merges overlapping and adjacent inclusive intervals contributed by entries.
+An entry contributes `[firstRowId, firstRowId + rowCount - 1]`. The resulting intervals are
+sorted and disjoint; they are never expanded into individual row IDs or coarsened to include gaps.
+
+```text
+rowIdPayload
+  minRowId : long                      // first interval's start
+  maxRowId : long                      // last interval's inclusive end
+  intsDeltaPayload                    // stores longs: 2 * (N - 1) sorted interior endpoints, base = minRowId
+```
+
+The envelope satisfies `0 <= minRowId <= maxRowId <= Long.MAX_VALUE`. Flatten the intervals
+as `[start0, end0, start1, end1, ...]`. The first start is supplied by `minRowId`, and the last
+end by `maxRowId`; only the remaining `2 * (N - 1)` interior endpoints are delta/varint encoded.
+Their count must be even, and the derived interval count `N = count / 2 + 1` must not exceed
+the block's record count. There is no separate stored interval count.
+Pairing the reconstructed endpoints recovers the intervals. Each pair satisfies
+`0 <= start <= end <= Long.MAX_VALUE`; each following start must exceed the preceding end.
+
+For `[(10, 19), (30, 39)]`, the minimum is 10 and maximum is 39. The interior
+endpoints `[19, 30]` have deltas `[9, 11]` from base 10, each encoded as one varint byte.
+The payload starts with an eight-byte minimum of 10 and an eight-byte maximum of 39,
+followed by `intsDeltaPayload` bytes `[2, 9, 11]`: 19 bytes, or 21 bytes with framing.
+For a single interval, the two eight-byte endpoints and a one-byte zero count define the
+interval: 17 payload bytes and no deltas.
+
+The reader first tests the envelope without decoding any deltas. A query for row ID 25
+passes the example's envelope check but matches neither interval. Unknown or invalid row-ID
+metadata makes that block's row-ID payload unavailable; partition and bucket coverage remain usable.
+
+#### Bucket Payload
+
+When `bucketEncoding == 1`, the block stores distinct bucket/count pairs:
+
+```text
+bucketPayload
+  buckets : intsDeltaPayload           // N > 0, base = 0, nonnegative deltas
+  totalBuckets : intsDeltaPayload      // N values, base = 0, ZigZag signed deltas
+```
+
+Pairs are sorted first by bucket, then by total bucket count, with duplicates removed.
+Each pair satisfies `0 <= bucket < totalBuckets <= Integer.MAX_VALUE`. Both sequences have
+the same count, and values at the same position form one pair. The totals must not be sorted
+independently. The same bucket may occur with different totals after rescaling.
+
+For `[(1, 4), (1, 8), (3, 4)]`, bucket values `[1, 1, 3]` have deltas `[1, 0, 2]`.
+Paired totals `[4, 8, 4]` have signed deltas `[4, 4, -4]`, ZigZag-encoded as `[8, 8, 7]`.
+The two payloads are `[3, 1, 0, 2]` and `[3, 8, 8, 7]`: 8 bytes total, or 10 bytes with framing.
+
+Missing, invalid or negative/synthetic bucket metadata makes the block's bucket coverage
+unavailable. A caller can supply a predicate on `(bucket, totalBuckets)` which conservatively
+retains every potentially matching pair. Filters requiring an entry's partition belong at
+the entry-filtering stage; omit the bucket predicate if no safe check is available.
+
+#### Validation and Reading
+
+Readers validate the checksum, container fields, payload lengths and leading count headers,
+and the complete physical block directory regardless of the query. Byte spans must cover
+the whole original manifest after its header; record counts must sum to the manifest entry
+count. Manifest length and entry count come from the supplied `ManifestFileMeta` rather
+than being duplicated in the sidecar. Unknown nonzero encodings skip their declared bytes
+without interpreting a count.
+
+Compressed contents are decoded only for dimensions needed by the filters. A row-ID envelope
+rejection skips all its deltas; a matching interval or partition ID skips remaining values.
+Bucket filtering first walks the bucket sequence to locate the paired totals without
+allocating arrays, then decodes pairs until a match. Unused totals may be skipped.
+Invalid varints, value counts, overflows,
+out-of-range values or ordering encountered while decoding invalidate the container. Delta
+contents skipped by short-circuiting are not individually validated.
+
+For conjunctive filters a block is retained only if every dimension is unavailable or matches.
+Matching tests row ID, partition, then bucket coverage. Absent filters are skipped, and a
+rejection skips the remaining dimensions. Matches in different dimensions can come from
+different entries, so entry filtering and ADD/DELETE reconciliation remain necessary.
+
+The entire sidecar is read in chunks of at most 1 MiB, including payloads unused by a query.
+There is no size-based fallback or payload dropping. Payload lengths save decoding work,
+not sidecar storage I/O. Selected compressed Avro blocks are read by byte range with adjacent
+spans coalesced and individual read requests bounded to 4 MiB. Building a sidecar does not
+modify the original manifest.
+
+`read` and `openManifest` accept an optional caller-supplied `SegmentsCache<Object>`. Complete
+sidecar bytes are keyed by their explicit `Path`. Only successful reads and selections
+populate the cache; query-specific selections are not cached. Cache entry-size limits affect
+admission only: larger sidecars are still fully read, validated and used.
+
+Selected Avro blocks also share this cache. Each entry contains one complete compressed block,
+keyed by the manifest's full path, original offset and encoded length, separately from whole-file
+keys. Different selections reuse the same blocks. Only complete reads populate the cache;
+oversized blocks stream through the read buffer. Adjacent uncached blocks fitting the buffer
+are read together and cached individually. Fully cached selections do not open the manifest.
+The cache retains its configured memory budget, entry-size limit, expiration and eviction policy.
+
 ## Manifest
 
 Data manifests record **ADD** (`0`) and **DELETE** (`1`) entries. Readers reconcile these entries

paimon-common/src/main/java/org/apache/paimon/utils/DeltaVarintCodec.java

@@ -0,0 +1,126 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.utils;
+
+import java.io.DataOutput;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Objects;
+
+/** Streaming count-prefixed delta/varint encoding of nonnegative integer sequences. */
+public final class DeltaVarintCodec {
+
+    private DeltaVarintCodec() {}
+
+    /** Writes a varint count followed by deltas, without buffering the sequence. */
+    public static final class Writer {
+        private final DataOutput out;
+        private final boolean signedDeltas;
+        private long remaining;
+        private long previous;
+
+        public Writer(DataOutput out, int count, long base) throws IOException {
+            this(out, count, base, false);
+        }
+
+        /** Signed deltas support non-monotonic nonnegative int values using ZigZag. */
+        public Writer(DataOutput out, int count, long base, boolean signedDeltas)
+                throws IOException {
+            if (base < 0 || count < 0 || (signedDeltas && base > Integer.MAX_VALUE)) {
+                throw new IllegalArgumentException("Invalid delta/varint count or base");
+            }
+            this.out = Objects.requireNonNull(out);
+            this.signedDeltas = signedDeltas;
+            this.remaining = count;
+            previous = base;
+            VarLengthIntUtils.encodeInt(out, count);
+        }
+
+        public void write(long value) throws IOException {
+            require(remaining > 0 && value >= 0);
+            long delta = value - previous;
+            if (signedDeltas) {
+                require(value <= Integer.MAX_VALUE);
+                delta = (delta << 1) ^ (delta >> 63);
+            } else {
+                require(delta >= 0);
+            }
+            VarLengthIntUtils.encodeLong(out, delta);
+            previous = value;
+            remaining--;
+        }
+    }
+
+    /**
+     * Reads the count and then values on demand. A complete read leaves the buffer at the next
+     * payload, allowing count-prefixed sequences to be concatenated. Reading may stop early.
+     */
+    public static final class Reader {
+        private final ByteBuffer data;
+        private final long max;
+        private final int count;
+        private final boolean signedDeltas;
+        private long remaining;
+        private long value;
+
+        public Reader(ByteBuffer data, long base, long max) throws IOException {
+            this(data, base, max, false);
+        }
+
+        public Reader(ByteBuffer data, long base, long max, boolean signedDeltas)
+                throws IOException {
+            this.data = Objects.requireNonNull(data);
+            long encodedCount = VarLengthIntUtils.decodeLong(data);
+            require(encodedCount <= Integer.MAX_VALUE && encodedCount <= data.remaining());
+            require(base >= 0 && max >= base && (!signedDeltas || max <= Integer.MAX_VALUE));
+            count = (int) encodedCount;
+            remaining = count;
+            value = base;
+            this.max = max;
+            this.signedDeltas = signedDeltas;
+        }
+
+        public int count() {
+            return count;
+        }
+
+        public boolean hasNext() {
+            return remaining > 0;
+        }
+
+        public long next() throws IOException {
+            require(remaining > 0);
+            long delta = VarLengthIntUtils.decodeLong(data);
+            if (signedDeltas) {
+                require(delta <= 2L * Integer.MAX_VALUE);
+                delta = (delta >>> 1) ^ -(delta & 1);
+            }
+            require(delta >= -value && delta <= max - value);
+            value += delta;
+            remaining--;
+            return value;
+        }
+    }
+
+    private static void require(boolean valid) throws IOException {
+        if (!valid) {
+            throw new IOException("Invalid delta/varint sequence");
+        }
+    }
+}