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

docs/docs/concepts/spec/manifest.md

@@ -63,6 +63,131 @@ 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
+
+With `manifest.sidecar.write` enabled, a manifest writer can create a binary
+`<manifest-file-name>.avro.sidecar` sidecar. Its name is stored in the manifest-list
+record's `_EXTRA_FILES`; the existing Avro schemas and `_VERSION` identifiers are unchanged.
+Readers identify the sidecar by the `.avro.sidecar` suffix among these explicit
+references, not by probing for a derived file name. Other extra-file references are preserved.
+
+With `manifest.sidecar.read` enabled and a partition, row-ID or bucket filter available, readers can use
+the sidecar to select complete Avro blocks before reading manifest entries. Both options
+default to `false`. Old manifests, null or empty extra-file lists, and lists containing only
+other extra-file types use the normal manifest read path. Missing, unsupported, corrupt,
+or over-budget containers also fall back to that path. Each block's partition, row-ID and bucket
+coverage is independently usable; an unavailable dimension cannot exclude a block.
+Cancellation and interruption errors propagate instead of triggering a full-manifest fallback.
+
+Version 1 uses the following layout. Container integers and payload integers
+are fixed-width big endian. Encoding IDs are unsigned bytes with separate namespaces.
+
+```text
+magic : 8 bytes                         // ASCII PAIMSCAR
+formatVersion : int                    // 1
+manifestNameHash : 32 bytes             // SHA-256 of the UTF-8 basename
+manifestLength : long
+manifestEntryCount : long               // ADD + DELETE
+avroHeaderLength : int
+avroHeader : bytes                      // original schema, codec and sync marker
+partitionCount : int
+partitionDictionary[]
+  partitionByteLength : int
+  partitionBytes : bytes                // existing manifest BinaryRow serialization
+blockCount : int
+blocks[]                               // original physical order
+  offset : long
+  length : long                         // complete encoded block, including sync marker
+  recordCount : long
+  partitionEncoding : byte
+  if partitionEncoding != 0:
+    partitionPayloadLength : int
+    partitionPayload : bytes
+  rowIdEncoding : byte
+  if rowIdEncoding != 0:
+    rowIdPayloadLength : int
+    rowIdPayload : bytes
+  bucketEncoding : byte
+  if bucketEncoding != 0:
+    bucketPayloadLength : int
+    bucketPayload : bytes
+checksum : 32 bytes                     // SHA-256 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` | Positive `partitionIdCount: int` followed by sorted unique dictionary IDs (`int`). |
+| Row ID | `1` | Positive `rangeCount: int` followed by sorted disjoint inclusive `(start: long, end: long)` pairs. Coverage may conservatively include gaps. |
+| Bucket | `1` | Positive `pairCount: int` followed by sorted unique `(bucket: int, totalBuckets: int)` pairs. |
+| Any | Other nonzero ID | Skip exactly the bounded 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. Invalid lengths, known-payload
+framing, dictionary references, interval order, checksums or physical coverage invalidate
+the container. Byte spans must cover the entire original manifest after its header;
+record counts must sum to the manifest entry count. Readers validate the checksum,
+payload framing (including known count/length consistency), and the complete block directory
+even when a block is rejected. Block payload contents are decoded and validated only for
+dimensions still needed by the filters.
+
+Bucket encoding 1 contains a positive `pairCount: int` followed by that many
+`(bucket: int, totalBuckets: int)` pairs. Pairs are sorted by bucket, then totalBuckets,
+and deduplicated. They preserve bucket-count changes between writes; the bucket number
+alone is not sufficient for point lookup after rescaling. A valid pair satisfies
+`0 <= bucket < totalBuckets`. Missing, invalid, negative/synthetic or over-budget bucket
+metadata makes that block's bucket coverage unavailable (encoding 0, no length or payload). Partition
+and row-ID coverage remain independently usable; no mutual-exclusion restriction is imposed.
+
+Readers test bucket-only queries using the existing bucket-selection logic, including
+the total-bucket count. Java uses conservative partition-independent bounds for
+`ManifestBucketFilter`; arbitrary partition-dependent callbacks remain at the entry
+filter stage. An unavailable bucket payload cannot exclude a block. Malformed payload lengths,
+pair counts, ordering or values invalidate the container rather than excluding a block.
+
+All entries contribute, including ADD, DELETE and every file format/column group.
+Row-ID ranges are never expanded into individual values. If an exact union exceeds its
+available byte budget, it becomes the inclusive `[min,max]` envelope with encoding 1. Processing
+continues through the end of the block to extend those bounds and detect unknown row IDs.
+An unknown or invalid row-ID range makes only that block's row-ID payload unavailable.
+Partition budget exhaustion independently makes that block's partition payload unavailable.
+The dictionary can consequently be incomplete for the manifest: a dictionary miss never
+excludes a block with unavailable partition coverage. Later blocks can still use existing IDs.
+
+`manifest.sidecar.max-bytes` bounds the whole serialized container, including the
+partition dictionary and all three payload types. It accepts memory sizes such as
+`16 mb` and, when unset, defaults to twice the configured `manifest.target-file-size`
+(16 MiB with the default 8 MiB manifest target). An explicit sidecar size overrides this default.
+The Avro header is also capped at 1 MiB and the directory at 131072 blocks.
+Writers discard optional row-ID payloads, bucket payloads, then partition payloads/dictionary if necessary,
+to fit the complete directory. If the directory itself cannot fit, no sidecar is published.
+No emitted sidecar omits block descriptors. These are encoded-size bounds; construction
+also incurs bounded object/buffer overhead. Query concurrency multiplies per-reader costs.
+
+For conjunctive filters a block is retained only if each dimension is either unavailable
+or matches. Matching skips absent filters and short-circuits after a dimension rejects a
+block, skipping the contents of later payloads. Matches in different dimensions can come
+from different entries in the block, so entry filtering and deletion merging remain
+necessary. Block min/max is derived from the first/last interval before testing the
+individual intervals.
+
+Readers still consume and validate the whole bounded sidecar. A partition-only query
+therefore reads row-ID payload bytes too; payload lengths save decoding work for unknown
+encodings, not storage I/O. Selected compressed blocks are read by byte range with adjacent
+spans coalesced. Existing immutable manifests are not backfilled by enabling the write option.
+
+Java selections covering every block can reuse the full-manifest cache; partial selections
+bypass it. PyPaimon explain scans disable sidecar pruning to preserve complete entry counters.
+
+Selected blocks still pass through entry filtering and ADD/DELETE reconciliation. Snapshot,
+tag, changelog, orphan-file and failed-commit cleanup retain or remove the sidecar through
+its extra-file reference together with the owning manifest.
+
 ## Manifest
 
 Data manifests record **ADD** (`0`) and **DELETE** (`1`) entries. Readers reconcile these entries

paimon-api/src/main/java/org/apache/paimon/CoreOptions.java

@@ -510,6 +510,27 @@ public InlineElement getDescription() {
                                     + "in the previous file. This must not exceed "
                                     + "'variant.shredding.minFieldCardinalityRatio'.");
 
+    public static final ConfigOption<Boolean> MANIFEST_SIDECAR_WRITE =
+            key("manifest.sidecar.write")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription(
+                            "Write sidecars with independent partition, row-id and bucket coverage for newly created manifests.");
+
+    public static final ConfigOption<Boolean> MANIFEST_SIDECAR_READ =
+            key("manifest.sidecar.read")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription(
+                            "Read optional manifest sidecars for partition, row-id or bucket filters after coarse pruning. Missing or invalid sidecars fall back to manifest reads.");
+
+    public static final ConfigOption<MemorySize> MANIFEST_SIDECAR_MAX_BYTES =
+            key("manifest.sidecar.max-bytes")
+                    .memoryType()
+                    .noDefaultValue()
+                    .withDescription(
+                            "Maximum serialized manifest sidecar size, including header and checksum. Defaults to twice manifest.target-file-size. Optional payloads are dropped before omitting a sidecar whose complete block directory cannot fit.");
+
     public static final ConfigOption<String> MANIFEST_COMPRESSION =
             key("manifest.compression")
                     .stringType()
@@ -3210,6 +3231,11 @@ public MemorySize manifestTargetSize() {
         return options.get(MANIFEST_TARGET_FILE_SIZE);
     }
 
+    public MemorySize manifestSidecarMaxSize() {
+        return options.getOptional(MANIFEST_SIDECAR_MAX_BYTES)
+                .orElseGet(() -> manifestTargetSize().multiply(2));
+    }
+
     public MemorySize manifestFullCompactionThresholdSize() {
         return options.get(MANIFEST_FULL_COMPACTION_FILE_SIZE);
     }

paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java

@@ -204,14 +204,15 @@ public ChangelogManager changelogManager() {
     @Override
     public ManifestFile.Factory manifestFileFactory() {
         return new ManifestFile.Factory(
-                fileIO,
-                schemaManager,
-                partitionType,
-                FileFormat.manifestFormat(options),
-                options.manifestCompression(),
-                pathFactory(),
-                options.manifestTargetSize().getBytes(),
-                readManifestCache);
+                        fileIO,
+                        schemaManager,
+                        partitionType,
+                        FileFormat.manifestFormat(options),
+                        options.manifestCompression(),
+                        pathFactory(),
+                        options.manifestTargetSize().getBytes(),
+                        readManifestCache)
+                .withSidecarOptions(options.toConfiguration());
     }
 
     @Override

paimon-core/src/main/java/org/apache/paimon/manifest/BucketFilter.java

@@ -78,6 +78,22 @@ public boolean test(BinaryRow partition, int bucket, int totalBucket) {
                 || totalAwareBucketFilter.test(partition, bucket, totalBucket);
     }
 
+    /** Conservatively checks an indexed pair without inventing a partition for custom filters. */
+    public boolean mayContain(int bucket, int totalBuckets) {
+        if (onlyReadRealBuckets && bucket < 0) {
+            return false;
+        }
+        if (specifiedBucket != null && bucket != specifiedBucket) {
+            return false;
+        }
+        if (bucketFilter != null && !bucketFilter.test(bucket)) {
+            return false;
+        }
+        return !(totalAwareBucketFilter instanceof ManifestBucketFilter)
+                || ((ManifestBucketFilter) totalAwareBucketFilter)
+                        .mayContain(bucket, bucket, totalBuckets);
+    }
+
     /** Conservatively tests whether a manifest's bucket metadata can contain a matching entry. */
     public boolean mayContain(ManifestFileMeta manifest) {
         Integer minBucket = manifest.minBucket();

paimon-core/src/main/java/org/apache/paimon/manifest/ManifestAvroReader.java

@@ -70,6 +70,19 @@ public final class ManifestAvroReader implements AutoCloseable {
         }
     }
 
+    @Nullable
+    public byte[] headerBytes() {
+        return blockReader.headerBytes();
+    }
+
+    public long blockOffset() {
+        return blockReader.blockOffset();
+    }
+
+    public long blockLength() {
+        return blockReader.blockLength();
+    }
+
     /** Returns whether another raw Avro block is available. */
     public boolean hasNext() throws IOException {
         return blockReader.hasNextBlock();

paimon-core/src/main/java/org/apache/paimon/manifest/ManifestAvroWriter.java

@@ -43,6 +43,7 @@
 import java.io.UncheckedIOException;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.IdentityHashMap;
 import java.util.List;
 import java.util.Map;
@@ -68,6 +69,7 @@ public final class ManifestAvroWriter implements AutoCloseable {
     private final String compression;
     private final PathFactory pathFactory;
     private final long targetFileSize;
+    private final ManifestSidecar.Settings sidecarSettings;
 
     private final List<ManifestFileMeta> results = new ArrayList<>();
     private final List<Path> completedPaths = new ArrayList<>();
@@ -83,7 +85,8 @@ public final class ManifestAvroWriter implements AutoCloseable {
             ObjectSerializer<ManifestEntry> serializer,
             String compression,
             PathFactory pathFactory,
-            long targetFileSize) {
+            long targetFileSize,
+            ManifestSidecar.Settings sidecarSettings) {
         this.fileIO = fileIO;
         this.schemaManager = schemaManager;
         this.partitionType = partitionType;
@@ -92,6 +95,7 @@ public final class ManifestAvroWriter implements AutoCloseable {
         this.compression = compression;
         this.pathFactory = pathFactory;
         this.targetFileSize = targetFileSize;
+        this.sidecarSettings = sidecarSettings;
     }
 
     public void write(ManifestEntry entry) throws IOException {
@@ -218,6 +222,9 @@ private void closeCurrentWriter() throws IOException {
         currentWriter.close();
         ManifestFileMeta result = currentWriter.result();
         completedPaths.add(currentWriter.path);
+        if (currentWriter.sidecarCreated) {
+            completedPaths.add(ManifestSidecar.path(currentWriter.path));
+        }
         results.add(result);
         currentWriter = null;
     }
@@ -413,6 +420,7 @@ private final class FileWriter {
         private @Nullable RowIdStats rowIdStats = new RowIdStats();
         private boolean closed;
         private boolean aborted;
+        private boolean sidecarCreated;
 
         private FileWriter(Path path) {
             this.path = path;
@@ -488,7 +496,7 @@ private void collectStats(ManifestEntry entry) {
             maxLevel = Math.max(maxLevel, entry.level());
             if (rowIdStats != null) {
                 Long firstRowId = entry.file().firstRowId();
-                if (firstRowId == null) {
+                if (!validRowIdRange(firstRowId, entry.file().rowCount())) {
                     rowIdStats = null;
                 } else {
                     rowIdStats.collect(firstRowId, entry.file().rowCount());
@@ -515,7 +523,7 @@ private void collectStats(EncodedEntry entry) {
             minLevel = Math.min(minLevel, entry.level);
             maxLevel = Math.max(maxLevel, entry.level);
             if (rowIdStats != null) {
-                if (!entry.hasRowId) {
+                if (!entry.hasRowId || !validRowIdRange(entry.firstRowId, entry.rowCount)) {
                     rowIdStats = null;
                 } else {
                     rowIdStats.collect(entry.firstRowId, entry.rowCount);
@@ -697,6 +705,14 @@ private Throwable abortCollecting(@Nullable Throwable primaryFailure, boolean de
                             ExceptionUtils.firstOrSuppressed(cleanupFailure, primaryFailure);
                 }
             }
+            if (sidecarCreated) {
+                try {
+                    fileIO.deleteQuietly(ManifestSidecar.path(path));
+                } catch (Throwable cleanupFailure) {
+                    primaryFailure =
+                            ExceptionUtils.firstOrSuppressed(cleanupFailure, primaryFailure);
+                }
+            }
             return primaryFailure;
         }
 
@@ -711,6 +727,7 @@ private void close() throws IOException {
                 outputBytes = out.getPos();
                 out.close();
                 out = null;
+                writeSidecar();
             } catch (IOException | RuntimeException | Error failure) {
                 abortCollecting(failure, true);
                 throw failure;
@@ -719,6 +736,27 @@ private void close() throws IOException {
             }
         }
 
+        private void writeSidecar() throws IOException {
+            if (!sidecarSettings.write) {
+                return;
+            }
+            byte[] bytes =
+                    ManifestSidecar.build(
+                            fileIO,
+                            path,
+                            outputBytes,
+                            Math.addExact(numAddedFiles, numDeletedFiles),
+                            sidecarSettings);
+            if (bytes != null) {
+                // Publish result() only after both immutable objects have closed. No rename.
+                try (PositionOutputStream sidecarOut =
+                        fileIO.newOutputStream(ManifestSidecar.path(path), false)) {
+                    sidecarCreated = true;
+                    sidecarOut.write(bytes);
+                }
+            }
+        }
+
         private ManifestFileMeta result() {
             if (!closed || outputBytes == null) {
                 throw new IllegalStateException(
@@ -740,10 +778,16 @@ private ManifestFileMeta result() {
                     rowIdStats == null ? null : rowIdStats.minRowId,
                     rowIdStats == null ? null : rowIdStats.maxRowId,
                     totalBucketsKnown ? totalBuckets : null,
-                    null);
+                    sidecarCreated
+                            ? Collections.singletonList(ManifestSidecar.path(path).getName())
+                            : null);
         }
     }
 
+    private static boolean validRowIdRange(@Nullable Long first, long count) {
+        return first != null && first >= 0 && count > 0 && count - 1 <= Long.MAX_VALUE - first;
+    }
+
     private static class RowIdStats {
 
         private long minRowId = Long.MAX_VALUE;