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

docs/docs/concepts/spec/manifest.md

@@ -63,6 +63,26 @@ 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.
 
+### Row-ID Block Index
+
+With `manifest.row-id-index.write` enabled, a manifest writer can create a binary
+`<manifest-file-name>.row-id-index` 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 row-ID index by the `.row-id-index` suffix among these explicit
+references, not by probing for a derived file name. Other extra-file references are preserved.
+
+With `manifest.row-id-index.read` enabled and a row-ID 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 indexes also fall back to that path. Writers omit the sidecar if complete
+row-ID coverage cannot be established within the configured range and byte budgets.
+Cancellation and interruption errors propagate instead of triggering a full-manifest fallback.
+
+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,34 @@ public InlineElement getDescription() {
                                     + "in the previous file. This must not exceed "
                                     + "'variant.shredding.minFieldCardinalityRatio'.");
 
+    public static final ConfigOption<Boolean> MANIFEST_ROW_ID_INDEX_WRITE =
+            key("manifest.row-id-index.write")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription(
+                            "Write complete row-id block indexes for newly created manifests.");
+
+    public static final ConfigOption<Boolean> MANIFEST_ROW_ID_INDEX_READ =
+            key("manifest.row-id-index.read")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription(
+                            "Read optional row-id sidecars after coarse manifest pruning. Missing or invalid indexes fall back to manifest reads.");
+
+    public static final ConfigOption<Integer> MANIFEST_ROW_ID_INDEX_MAX_RANGES =
+            key("manifest.row-id-index.max-ranges")
+                    .intType()
+                    .defaultValue(131072)
+                    .withDescription(
+                            "Maximum disjoint row-id intervals across all Avro blocks in a manifest. Exceeding the limit disables the entire index. Range: 1 to 1048576.");
+
+    public static final ConfigOption<Integer> MANIFEST_ROW_ID_INDEX_MAX_BYTES =
+            key("manifest.row-id-index.max-bytes")
+                    .intType()
+                    .defaultValue(8388608)
+                    .withDescription(
+                            "Maximum serialized row-id sidecar bytes, including header and checksum. Exceeding the limit disables the entire index. Range: 128 to 67108864.");
+
     public static final ConfigOption<String> MANIFEST_COMPRESSION =
             key("manifest.compression")
                     .stringType()

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)
+                .withRowIdIndexOptions(options.toConfiguration());
     }
 
     @Override

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 ManifestRowIdIndex.Settings rowIdIndexSettings;
 
     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,
+            ManifestRowIdIndex.Settings rowIdIndexSettings) {
         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.rowIdIndexSettings = rowIdIndexSettings;
     }
 
     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(ManifestRowIdIndex.path(currentWriter.path));
+        }
         results.add(result);
         currentWriter = null;
     }
@@ -403,6 +410,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;
@@ -477,7 +485,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());
@@ -503,7 +511,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);
@@ -668,6 +676,14 @@ private Throwable abortCollecting(@Nullable Throwable primaryFailure, boolean de
                             ExceptionUtils.firstOrSuppressed(cleanupFailure, primaryFailure);
                 }
             }
+            if (sidecarCreated) {
+                try {
+                    fileIO.deleteQuietly(ManifestRowIdIndex.path(path));
+                } catch (Throwable cleanupFailure) {
+                    primaryFailure =
+                            ExceptionUtils.firstOrSuppressed(cleanupFailure, primaryFailure);
+                }
+            }
             return primaryFailure;
         }
 
@@ -682,6 +698,7 @@ private void close() throws IOException {
                 outputBytes = out.getPos();
                 out.close();
                 out = null;
+                writeRowIdIndex();
             } catch (IOException | RuntimeException | Error failure) {
                 abortCollecting(failure, true);
                 throw failure;
@@ -690,6 +707,27 @@ private void close() throws IOException {
             }
         }
 
+        private void writeRowIdIndex() throws IOException {
+            if (!rowIdIndexSettings.write) {
+                return;
+            }
+            byte[] bytes =
+                    ManifestRowIdIndex.build(
+                            fileIO,
+                            path,
+                            outputBytes,
+                            Math.addExact(numAddedFiles, numDeletedFiles),
+                            rowIdIndexSettings);
+            if (bytes != null) {
+                // Publish result() only after both immutable objects have closed. No rename.
+                try (PositionOutputStream indexOut =
+                        fileIO.newOutputStream(ManifestRowIdIndex.path(path), false)) {
+                    sidecarCreated = true;
+                    indexOut.write(bytes);
+                }
+            }
+        }
+
         private ManifestFileMeta result() {
             if (!closed || outputBytes == null) {
                 throw new IllegalStateException(
@@ -709,10 +747,17 @@ private ManifestFileMeta result() {
                     levelStatsKnown ? minLevel : null,
                     levelStatsKnown ? maxLevel : null,
                     rowIdStats == null ? null : rowIdStats.minRowId,
-                    rowIdStats == null ? null : rowIdStats.maxRowId);
+                    rowIdStats == null ? null : rowIdStats.maxRowId,
+                    sidecarCreated
+                            ? Collections.singletonList(ManifestRowIdIndex.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;