[core] Prune manifest blocks with sidecars during scans

docs/docs/concepts/spec/manifest.md

@@ -72,20 +72,28 @@ using independent partition, row-ID and bucket coverage. A sidecar uses the
 derived file name. The Avro schemas and `_VERSION` identifiers remain unchanged.
 
 The utility includes construction, validation, block selection and optional caching. Java table
-writers generate sidecars when `manifest.sidecar.enabled` is true; when unset, it inherits
+writers and scans use sidecars when `manifest.sidecar.enabled` is true; when unset, it inherits
 `manifest-sort.enabled`. Both ordinary writes and raw manifest rewrites build the sidecar from
 the completed output manifest and publish its `_EXTRA_FILES` reference only after both files
 close successfully. Failed writes and aborted writers clean up their own manifest/sidecar pairs.
-Scans do not yet invoke sidecar pruning automatically. Callers remain responsible for applying
-entry filters and reconciling ADD/DELETE entries after block selection. The low-level `build`
-method returns sidecar bytes without writing or publishing another file.
+Scans with partition, row-ID or bucket filters select blocks before reading manifest entries.
+Normal entry filtering and ADD/DELETE reconciliation still apply. Missing or unusable sidecars
+fall back to normal manifest reads; disabled sidecars and scans without these filters do not
+perform sidecar I/O. Sidecar caching is controlled by the catalog option
+`cache.manifest-sidecar.max-memory` (64 MiB by default). A positive value supplies an
+additional budget independent of the manifest content cache. When set to 0, sidecars reuse
+the manifest content cache, or remain uncached if that cache is disabled. Sidecar caching
+uses the catalog's `cache.expire-after-access` and `cache.manifest.soft-values` policies.
+Selected block bytes still share the manifest content cache without populating the
+whole-manifest entry cache with partial results. The low-level `build` method returns
+sidecar bytes without writing or publishing 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.
+metadata makes only the affected block's dimension unavailable. Cache limits do not truncate
+sidecars: 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

docs/generated/catalog_configuration.html

@@ -50,6 +50,12 @@
             <td>Duration</td>
             <td>Cache expiration policy: marks cache entries to expire after a specified duration has passed since their last refresh.</td>
         </tr>
+        <tr>
+            <td><h5>cache.manifest-sidecar.max-memory</h5></td>
+            <td style="word-wrap: break-word;">64 mb</td>
+            <td>MemorySize</td>
+            <td>Controls the maximum memory for the separate manifest sidecar cache. This budget is additional to the manifest content cache. Set to 0 to reuse the manifest content cache. It uses 'cache.expire-after-access' and 'cache.manifest.soft-values'.</td>
+        </tr>
         <tr>
             <td><h5>cache.manifest.max-memory</h5></td>
             <td style="word-wrap: break-word;">(none)</td>

paimon-api/src/main/java/org/apache/paimon/options/CatalogOptions.java

@@ -130,6 +130,16 @@ public class CatalogOptions {
                     .noDefaultValue()
                     .withDescription("Controls the maximum memory to cache manifest content.");
 
+    public static final ConfigOption<MemorySize> CACHE_MANIFEST_SIDECAR_MAX_MEMORY =
+            key("cache.manifest-sidecar.max-memory")
+                    .memoryType()
+                    .defaultValue(MemorySize.ofMebiBytes(64))
+                    .withDescription(
+                            "Controls the maximum memory for the separate manifest sidecar cache. "
+                                    + "This budget is additional to the manifest content cache. "
+                                    + "Set to 0 to reuse the manifest content cache. It uses "
+                                    + "'cache.expire-after-access' and 'cache.manifest.soft-values'.");
+
     public static final ConfigOption<Boolean> CACHE_MANIFEST_SOFT_VALUES =
             key("cache.manifest.soft-values")
                     .booleanType()

paimon-benchmark/paimon-micro-benchmarks/src/test/java/org/apache/paimon/benchmark/ManifestFileSorterBenchmark.java

@@ -301,6 +301,7 @@ private ManifestFile createManifestFile() {
                         pathFactory,
                         TARGET_MANIFEST_SIZE,
                         null,
+                        null,
                         new CoreOptions(new Options()))
                 .create();
     }

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

@@ -109,6 +109,7 @@ abstract class AbstractFileStore<T> implements FileStore<T> {
     protected final CatalogEnvironment catalogEnvironment;
 
     @Nullable private SegmentsCache<Path> readManifestCache;
+    @Nullable private SegmentsCache<Path> manifestSidecarCache;
     @Nullable private Cache<Path, Snapshot> snapshotCache;
 
     protected AbstractFileStore(
@@ -212,6 +213,7 @@ public ManifestFile.Factory manifestFileFactory() {
                 pathFactory(),
                 options.manifestTargetSize().getBytes(),
                 readManifestCache,
+                manifestSidecarCache,
                 options);
     }
 
@@ -616,6 +618,11 @@ public void setManifestCache(SegmentsCache<Path> manifestCache) {
         this.readManifestCache = manifestCache;
     }
 
+    @Override
+    public void setManifestSidecarCache(SegmentsCache<Path> manifestSidecarCache) {
+        this.manifestSidecarCache = manifestSidecarCache;
+    }
+
     @Override
     public void setSnapshotCache(Cache<Path, Snapshot> cache) {
         this.snapshotCache = cache;

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

@@ -127,5 +127,7 @@ boolean mergeSchema(
 
     void setManifestCache(SegmentsCache<Path> manifestCache);
 
+    void setManifestSidecarCache(SegmentsCache<Path> manifestSidecarCache);
+
     void setSnapshotCache(Cache<Path, Snapshot> cache);
 }

paimon-core/src/main/java/org/apache/paimon/catalog/CachingCatalog.java

@@ -51,6 +51,7 @@
 import static org.apache.paimon.options.CatalogOptions.CACHE_EXPIRE_AFTER_ACCESS;
 import static org.apache.paimon.options.CatalogOptions.CACHE_EXPIRE_AFTER_WRITE;
 import static org.apache.paimon.options.CatalogOptions.CACHE_MANIFEST_MAX_MEMORY;
+import static org.apache.paimon.options.CatalogOptions.CACHE_MANIFEST_SIDECAR_MAX_MEMORY;
 import static org.apache.paimon.options.CatalogOptions.CACHE_MANIFEST_SMALL_FILE_MEMORY;
 import static org.apache.paimon.options.CatalogOptions.CACHE_MANIFEST_SMALL_FILE_THRESHOLD;
 import static org.apache.paimon.options.CatalogOptions.CACHE_MANIFEST_SOFT_VALUES;
@@ -71,6 +72,7 @@ public class CachingCatalog extends DelegateCatalog {
     protected Cache<String, Database> databaseCache;
     protected Cache<Identifier, Table> tableCache;
     @Nullable protected final SegmentsCache<Path> manifestCache;
+    @Nullable protected final SegmentsCache<Path> manifestSidecarCache;
     // partition cache will affect data latency
     @Nullable protected Cache<Identifier, List<Partition>> partitionCache;
     @Nullable protected DVMetaCache dvMetaCache;
@@ -108,6 +110,17 @@ public CachingCatalog(Catalog wrapped, Options options) {
                         expireAfterAccess,
                         manifestCacheSoftValues);
 
+        MemorySize sidecarMaxMemory = options.get(CACHE_MANIFEST_SIDECAR_MAX_MEMORY);
+        this.manifestSidecarCache =
+                sidecarMaxMemory.getBytes() == 0
+                        ? manifestCache
+                        : SegmentsCache.create(
+                                (int) CoreOptions.PAGE_SIZE.defaultValue().getBytes(),
+                                sidecarMaxMemory,
+                                sidecarMaxMemory.getBytes(),
+                                expireAfterAccess,
+                                manifestCacheSoftValues);
+
         this.cachedPartitionMaxNum = options.get(CACHE_PARTITION_MAX_NUM);
 
         int cacheDvMaxNum = options.get(CACHE_DV_MAX_NUM);
@@ -299,6 +312,9 @@ private Table loadTable(Identifier identifier) {
             if (manifestCache != null) {
                 storeTable.setManifestCache(manifestCache);
             }
+            if (manifestSidecarCache != null) {
+                storeTable.setManifestSidecarCache(manifestSidecarCache);
+            }
             if (dvMetaCache != null) {
                 storeTable.setDVMetaCache(dvMetaCache);
             }
@@ -425,6 +441,11 @@ public CacheSizes estimatedCacheSizes() {
             manifestCacheSize = manifestCache.estimatedSize();
             manifestCacheBytes = manifestCache.totalCacheBytes();
         }
+        // Keep reporting total manifest metadata usage even though sidecars have their own budget.
+        if (manifestSidecarCache != null && manifestSidecarCache != manifestCache) {
+            manifestCacheSize += manifestSidecarCache.estimatedSize();
+            manifestCacheBytes += manifestSidecarCache.totalCacheBytes();
+        }
         long partitionCacheSize = 0L;
         if (partitionCache != null) {
             for (Map.Entry<Identifier, List<Partition>> entry : partitionCache.asMap().entrySet()) {

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/ManifestFile.java

@@ -37,6 +37,7 @@
 import org.apache.paimon.utils.Filter;
 import org.apache.paimon.utils.ObjectsFile;
 import org.apache.paimon.utils.PathFactory;
+import org.apache.paimon.utils.RowRangeIndex;
 import org.apache.paimon.utils.SegmentsCache;
 
 import javax.annotation.Nullable;
@@ -61,6 +62,7 @@ public class ManifestFile extends ObjectsFile<ManifestEntry> {
     private final AvroFileFormat avroFileFormat;
     private final long suggestedFileSize;
     private final CoreOptions options;
+    @Nullable private final SegmentsCache<Path> sidecarCache;
 
     private ManifestFile(
             FileIO fileIO,
@@ -72,6 +74,7 @@ private ManifestFile(
             PathFactory pathFactory,
             long suggestedFileSize,
             @Nullable SegmentsCache<Path> cache,
+            @Nullable SegmentsCache<Path> sidecarCache,
             CoreOptions options) {
         super(
                 fileIO,
@@ -89,6 +92,7 @@ private ManifestFile(
         this.avroFileFormat = avroFileFormat;
         this.suggestedFileSize = suggestedFileSize;
         this.options = options;
+        this.sidecarCache = sidecarCache == null ? cache : sidecarCache;
     }
 
     @Override
@@ -140,9 +144,33 @@ public <T> List<T> read(
             Filter<InternalRow> readFilter,
             Filter<ManifestEntry> readTFilter,
             Function<ManifestEntry, T> convertor) {
+        return read(
+                fileName,
+                fileSize,
+                partitionFilter,
+                bucketFilter,
+                readFilter,
+                readTFilter,
+                convertor,
+                null);
+    }
+
+    public <T> List<T> read(
+            String fileName,
+            @Nullable Long fileSize,
+            @Nullable PartitionPredicate partitionFilter,
+            @Nullable BucketFilter bucketFilter,
+            Filter<InternalRow> readFilter,
+            Filter<ManifestEntry> readTFilter,
+            Function<ManifestEntry, T> convertor,
+            @Nullable ManifestSidecar.Selection selected) {
+        if (selected != null && selected.blocks().isEmpty()) {
+            return java.util.Collections.emptyList();
+        }
         try {
             Path path = pathFactory.toPath(fileName);
-            if (cache != null) {
+            // Sidecar selections use the block cache, even when every block is selected.
+            if (cache != null && selected == null) {
                 ManifestEntryFilters filters =
                         new ManifestEntryFilters(
                                 partitionFilter, bucketFilter, readFilter, readTFilter);
@@ -155,7 +183,9 @@ public <T> List<T> read(
                             path,
                             ManifestEntry.MANIFEST_ROW_TYPE,
                             partitionFilter,
-                            bucketFilter);
+                            bucketFilter,
+                            selected,
+                            cache == null ? null : cache.segmentsCache());
             return readFromIterator(iterator, serializer, readFilter, readTFilter, convertor);
         } catch (IOException e) {
             throw new UncheckedIOException(e);
@@ -209,8 +239,23 @@ private static CloseableIterator<InternalRow> createManifestIterator(
             @Nullable PartitionPredicate partitionFilter,
             @Nullable BucketFilter bucketFilter)
             throws IOException {
+        return createManifestIterator(
+                fileIO, path, projectedType, partitionFilter, bucketFilter, null, null);
+    }
+
+    private static CloseableIterator<InternalRow> createManifestIterator(
+            FileIO fileIO,
+            Path path,
+            RowType projectedType,
+            @Nullable PartitionPredicate partitionFilter,
+            @Nullable BucketFilter bucketFilter,
+            @Nullable ManifestSidecar.Selection selected,
+            @Nullable SegmentsCache<Object> cache)
+            throws IOException {
         try {
-            ManifestAvroReader reader = new ManifestAvroReader(fileIO.newInputStream(path));
+            ManifestAvroReader reader =
+                    new ManifestAvroReader(
+                            ManifestSidecar.openManifest(fileIO, path, selected, cache));
             return reader.read(projectedType, partitionFilter, bucketFilter);
         } catch (IOException e) {
             FileUtils.checkExists(fileIO, path);
@@ -345,6 +390,32 @@ public Path toPath(String fileName) {
         };
     }
 
+    @Nullable
+    public ManifestSidecar.Selection selectBlocks(
+            ManifestFileMeta manifest, @Nullable RowRangeIndex query) {
+        return selectBlocks(manifest, query, null, null);
+    }
+
+    @Nullable
+    public ManifestSidecar.Selection selectBlocks(
+            ManifestFileMeta manifest,
+            @Nullable RowRangeIndex query,
+            @Nullable PartitionPredicate partitionFilter,
+            @Nullable BucketFilter bucketFilter) {
+        return !options.manifestSidecarEnabled()
+                        || (query == null && partitionFilter == null && bucketFilter == null)
+                ? null
+                : ManifestSidecar.read(
+                        fileIO,
+                        pathFactory.toPath(manifest.fileName()),
+                        manifest,
+                        query,
+                        partitionFilter,
+                        partitionType,
+                        bucketFilter == null ? null : bucketFilter::mayContain,
+                        sidecarCache);
+    }
+
     /** Deletes an unreferenced manifest and its explicitly referenced extra files. */
     public void delete(ManifestFileMeta manifest) {
         delete(manifest.fileName());
@@ -365,6 +436,7 @@ public static class Factory {
         private final long suggestedFileSize;
         private final CoreOptions options;
         @Nullable private final SegmentsCache<Path> cache;
+        @Nullable private final SegmentsCache<Path> sidecarCache;
 
         public Factory(
                 FileIO fileIO,
@@ -375,6 +447,7 @@ public Factory(
                 FileStorePathFactory pathFactory,
                 long suggestedFileSize,
                 @Nullable SegmentsCache<Path> cache,
+                @Nullable SegmentsCache<Path> sidecarCache,
                 CoreOptions options) {
             this.fileIO = fileIO;
             this.schemaManager = schemaManager;
@@ -384,13 +457,10 @@ public Factory(
             this.pathFactory = pathFactory;
             this.suggestedFileSize = suggestedFileSize;
             this.cache = cache;
+            this.sidecarCache = sidecarCache;
             this.options = options;
         }
 
-        public boolean isCacheEnabled() {
-            return cache != null;
-        }
-
         public ManifestFile create() {
             return new ManifestFile(
                     fileIO,
@@ -402,6 +472,7 @@ public ManifestFile create() {
                     pathFactory.manifestFileFactory(),
                     suggestedFileSize,
                     cache,
+                    sidecarCache,
                     options);
         }
     }

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

@@ -632,7 +632,7 @@ public static Selection read(
             @Nullable PartitionPredicate partitionFilter,
             @Nullable RowType partitionType,
             @Nullable BiPredicate<Integer, Integer> bucketFilter,
-            @Nullable SegmentsCache<Object> cache) {
+            @Nullable SegmentsCache<Path> cache) {
         String sidecarFileName = fileName(manifest);
         if (sidecarFileName == null) {
             return null;
@@ -660,7 +660,7 @@ public static Selection read(
         }
     }
 
-    /** Complete sidecar bytes stored in the shared manifest cache. */
+    /** Complete sidecar bytes stored in the configured sidecar cache. */
     static final class ManifestSidecarSegment implements Segments {
         private final byte[] bytes;