[core] Unify manifest entry caching by physical block

docs/docs/concepts/spec/manifest.md

@@ -84,9 +84,16 @@ perform sidecar I/O. Sidecar caching is controlled by the catalog option
 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.
+Manifest contents use one decoded-entry cache keyed by physical block path, offset and length.
+Full reads and sidecar-selected reads reuse the same complete, unfiltered blocks; compressed
+block bytes and whole-file entry copies are not retained separately. Query filters and
+converters run after loading the cached entries. Decoded buffers and their lookup metadata
+are accounted against the manifest content cache budget, and oversized blocks are read
+without retaining a partial cache entry. A complete block directory may be cached separately
+from the entries after reaching EOF, or after selecting every physical block. Without that
+directory, the first full read still scans the manifest to discover block boundaries, while
+reusing any already decoded blocks. The low-level `build` method returns sidecar bytes without
+writing or publishing another file.
 
 PyPaimon can read these sidecars and prune manifest blocks using partition, row-ID and bucket
 filters. Its `manifest.sidecar.enabled` option inherits `manifest-sort.enabled` when unset.
@@ -305,17 +312,21 @@ not sidecar storage I/O. Selected compressed Avro blocks are read by byte range
 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
+`read` accepts an optional caller-supplied `SegmentsCache<Path>`. 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 reads cache the decoded entries of each complete Avro block, keyed by
+`(path, offset, length)` in a namespace distinct from metadata path keys. These entries have
+the complete manifest schema and are independent of query filters and converters. Full and
+selected reads reuse the same entries; there is no separate compressed-block cache. Only
+successful complete decoding populates a block entry. If decoded buffers exceed the entry
+or memory limit, the reader replays its current raw block without another file read and
+without retaining a filtered prefix. Adjacent missing blocks are fetched together in bounded
+reads. Fully cached selections do not open the manifest. A path-keyed complete block directory
+contains only the header and physical descriptors, not a second copy of manifest entries.
+The caches retain their configured memory budgets, expiration and eviction policies.
 
 ## Manifest
 

docs/generated/catalog_configuration.html

@@ -72,7 +72,7 @@
             <td><h5>cache.manifest.small-file-threshold</h5></td>
             <td style="word-wrap: break-word;">1 mb</td>
             <td>MemorySize</td>
-            <td>Controls the threshold of small manifest file.</td>
+            <td>Controls the per-element cache threshold for metadata files and decoded manifest blocks.</td>
         </tr>
         <tr>
             <td><h5>cache.manifest.soft-values</h5></td>

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

@@ -122,7 +122,8 @@ public class CatalogOptions {
             key("cache.manifest.small-file-threshold")
                     .memoryType()
                     .defaultValue(MemorySize.ofMebiBytes(1))
-                    .withDescription("Controls the threshold of small manifest file.");
+                    .withDescription(
+                            "Controls the per-element cache threshold for metadata files and decoded manifest blocks.");
 
     public static final ConfigOption<MemorySize> CACHE_MANIFEST_MAX_MEMORY =
             key("cache.manifest.max-memory")

paimon-common/src/main/java/org/apache/paimon/io/DataPagedOutputSerializer.java

@@ -76,6 +76,13 @@ SimpleCollectingOutputView pagedOut() {
         return pagedOut;
     }
 
+    /** Returns the bytes currently allocated for serialized data. */
+    public long memorySize() {
+        return pagedOut == null
+                ? initialOut.getSharedBuffer().length
+                : (long) pagedOut.fullSegments().size() * pageSize;
+    }
+
     /**
      * Serializes a binary row to the output.
      *

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

@@ -0,0 +1,63 @@
+/*
+ * 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.manifest;
+
+import org.apache.paimon.fs.Path;
+
+import java.util.Objects;
+
+/** Identifies one complete physical block of an immutable manifest file. */
+public final class BlockKey {
+
+    private final Path path;
+    private final long offset;
+    private final long length;
+
+    public BlockKey(Path path, long offset, long length) {
+        this.path = Objects.requireNonNull(path);
+        this.offset = offset;
+        this.length = length;
+    }
+
+    public Path path() {
+        return path;
+    }
+
+    public long offset() {
+        return offset;
+    }
+
+    public long length() {
+        return length;
+    }
+
+    @Override
+    public boolean equals(Object other) {
+        if (!(other instanceof BlockKey)) {
+            return false;
+        }
+        BlockKey that = (BlockKey) other;
+        return path.equals(that.path) && offset == that.offset && length == that.length;
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(path, offset, length);
+    }
+}

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

@@ -439,7 +439,7 @@ public RowIterator toRows(RowType projectedType) throws IOException {
             return toRows(projectedType, null, null, true);
         }
 
-        private RowIterator toRows(
+        RowIterator toRows(
                 RowType projectedType,
                 @Nullable PartitionPredicate partitionFilter,
                 @Nullable BucketFilter bucketFilter,

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

@@ -0,0 +1,141 @@
+/*
+ * 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.manifest;
+
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.partition.PartitionPredicate;
+import org.apache.paimon.utils.CloseableIterator;
+import org.apache.paimon.utils.FileUtils;
+
+import javax.annotation.Nullable;
+
+import java.io.Closeable;
+import java.io.EOFException;
+import java.io.IOException;
+
+/** Query-local reader of full manifests or a sequence of cache-missing physical blocks. */
+final class ManifestBlockReader implements Closeable {
+
+    private final FileIO fileIO;
+    private final Path path;
+    @Nullable private final ManifestSidecar.Selection selected;
+    @Nullable private ManifestAvroReader reader;
+    @Nullable private ManifestAvroReader.RawBlock current;
+    @Nullable private BlockKey currentKey;
+    private int position;
+    private long firstRecord;
+
+    ManifestBlockReader(FileIO fileIO, Path path, @Nullable ManifestSidecar.Selection selected) {
+        this.fileIO = fileIO;
+        this.path = path;
+        this.selected = selected;
+    }
+
+    private ManifestAvroReader reader() throws IOException {
+        if (reader == null) {
+            try {
+                reader =
+                        new ManifestAvroReader(
+                                ManifestSidecar.openManifest(fileIO, path, selected));
+            } catch (IOException e) {
+                FileUtils.checkExists(fileIO, path);
+                throw e;
+            }
+        }
+        return reader;
+    }
+
+    byte[] header() throws IOException {
+        return reader().headerBytes();
+    }
+
+    boolean hasNext() throws IOException {
+        return reader().hasNext();
+    }
+
+    ManifestSidecar.Block next() throws IOException {
+        ManifestAvroReader input = reader();
+        current = input.next();
+        ManifestSidecar.Block block;
+        if (selected == null) {
+            block =
+                    new ManifestSidecar.Block(
+                            input.blockOffset(),
+                            input.blockLength(),
+                            firstRecord,
+                            current.recordCount());
+        } else {
+            if (position >= selected.blocks().size()) {
+                throw new IOException("Unexpected manifest block");
+            }
+            block = selected.blocks().get(position++);
+            if (current.recordCount() != block.recordCount) {
+                throw new IOException("Manifest block record count does not match its directory");
+            }
+        }
+        firstRecord = Math.addExact(firstRecord, current.recordCount());
+        currentKey = new BlockKey(path, block.offset, block.length);
+        return block;
+    }
+
+    CloseableIterator<InternalRow> rows(
+            BlockKey key,
+            @Nullable PartitionPredicate partitionFilter,
+            @Nullable BucketFilter bucketFilter)
+            throws IOException {
+        while (currentKey == null || currentKey.offset() < key.offset()) {
+            if (!hasNext()) {
+                throw new EOFException("Missing manifest block at " + key.offset());
+            }
+            next();
+        }
+        if (!key.equals(currentKey)) {
+            throw new IOException("Unexpected manifest block position");
+        }
+        ManifestAvroReader.RowIterator rows =
+                current.toRows(
+                        ManifestEntry.MANIFEST_ROW_TYPE, partitionFilter, bucketFilter, true);
+        return new CloseableIterator<InternalRow>() {
+
+            @Override
+            public boolean hasNext() {
+                return rows.hasNext();
+            }
+
+            @Override
+            public InternalRow next() {
+                return rows.next();
+            }
+
+            @Override
+            public void close() {
+                // The enclosing query owns the file reader, not an individual block iterator.
+            }
+        };
+    }
+
+    @Override
+    public void close() throws IOException {
+        if (reader != null) {
+            reader.close();
+        }
+    }
+}