perf(manifest): share indexed entries across file and snapshot caches

docs/source/user_guide/manifest_cache.rst

@@ -21,24 +21,26 @@ Manifest Cache
 Overview
 --------
 
-Paimon C++ caches decoded, schema-aligned manifest batches as Arrow IPC streams
-in ``ObjectsFile<T>``. The cache uses the public ``Cache`` abstraction and is
-enabled through ``ScanContextBuilder::WithCache()``. It covers data manifests,
-manifest lists, and index manifests because they all read through ``ObjectsFile<T>``.
-
-For repeated ``get``, ``scan``, or batch ``get/scan -f`` requests in the same
-process, the same snapshot often reads the same manifest files repeatedly. On a
-cache hit, the read path skips remote filesystem ``open/read`` and Avro/ORC
-source-format decoding. It still reads the IPC stream, deserializes objects,
-and applies scan filters. Bucket and row-range reads can select entries before
-constructing their file metadata. Ordinary, bucket, and row-range reads share
-the same complete, query-independent cache entry.
-
-A cold cache load decodes the complete manifest and serializes its aligned
-batches into IPC. The loading reader consumes the original batches without an
-IPC round trip. Cold bucket reads may therefore decode more entries than
-uncached selective reads. Without a cache, reads retain the existing
-source-format reader path.
+Paimon C++ caches complete, schema-aligned data manifests as indexed binary rows
+in ``ManifestFile``. Rows are grouped by partition, bucket, and total bucket count,
+with ordinals preserving the original ADD/DELETE order. Manifest lists and index
+manifests continue to use Arrow IPC streams in ``ObjectsFile<T>``. All three use
+the public ``Cache`` abstraction, enabled through ``ScanContextBuilder::WithCache()``.
+
+The immutable manifest file path identifies each cache value. Different snapshots
+and scan filters can therefore share the same file. A hit skips filesystem reads
+and source-format decoding. Partition and bucket selection skips entire groups
+before constructing ``ManifestEntry`` or ``DataFileMeta`` objects; row-range
+selection can further prune the selected rows. Ordinary, bucket, and row-range
+reads share one complete, query-independent cache entry.
+
+A cold load reads the complete manifest and builds its index. The loading reader
+consumes the original Arrow batches without a cache serialization round trip.
+Cold bucket reads may therefore read more metadata than uncached selective reads.
+Without a cache, reads retain the existing source-format reader path. Cache
+serialization or admission failures do not prevent reading the source. Obsolete
+or malformed data-manifest cache envelopes are rebuilt; predicate errors propagate
+without retry. This private cache format does not change persisted manifest files.
 
 Configuration
 -------------
@@ -97,12 +99,13 @@ Cache Implementation Responsibilities
 
 Embedding applications can implement hit/miss and eviction statistics in their
 ``Cache`` implementation. Coordination of concurrent loads for the same key
-also belongs to that implementation; ``ObjectsFile<T>`` does not deduplicate
+also belongs to that implementation; the manifest readers do not deduplicate
 concurrent cache misses.
 
 Cache implementations should use ``CacheValue::GetMemoryUsage()`` for admission
-and eviction accounting. For manifest IPC, this reports the retained Arrow
-buffer capacity, including unused space from growth. ``GetSegment().Size()``
-continues to describe the valid IPC byte length. The built-in ``LruCache`` uses
-the memory usage value; existing cache values without an explicit allocation
-size continue to charge their segment length.
+and eviction accounting. Indexed data manifests charge the serialized bytes and
+the retained ``Bytes`` object. Manifest-list and index-manifest IPC values charge
+the retained Arrow buffer capacity, including unused space from growth.
+``GetSegment().Size()`` describes the serialized payload length. The built-in
+``LruCache`` uses the memory usage value; existing cache values without an explicit
+allocation size continue to charge their segment length.

docs/source/user_guide/manifest_entry_cache.rst

@@ -41,30 +41,31 @@ Selective manifest decoding
 
 With ``Options::SCAN_MANIFEST_ENTRY_LAZY_DECODE_ENABLED`` enabled (the default),
 scans evaluate partition filters before constructing ``ManifestEntry`` and
-``DataFileMeta`` objects or decoding file statistics. The same row filter runs
-on ordinary manifest reads, cached manifest Arrow batches, and exact snapshot
-cache hits. This avoids full metadata decoding for excluded partitions even
-when no cache is configured. Row-range reads apply both range and partition
-selection before decoding; ADD and DELETE entries receive the same selection.
+``DataFileMeta`` objects or decoding file statistics. Uncached reads inspect each
+row's partition, while data-manifest and exact snapshot cache hits test partitions
+in the shared indexed container before visiting their rows. This avoids full
+metadata decoding for excluded partitions even when no cache is configured.
+Row-range reads apply both range and partition selection before decoding; ADD and
+DELETE entries receive the same selection.
 Level and data predicates still run through the normal scan filters.
 
 The snapshot cache keeps the complete merged entries for each bucket and
 snapshot in serialized form. A hit selects the exact retained snapshot.
-Rows belonging to other snapshots are skipped by their serialized lengths,
-without allocating or copying their payloads. When there is no partition filter,
-all entries of the selected snapshot are decoded. Disabling lazy decoding also
+Other snapshots are skipped as length-delimited blocks, without allocating or
+copying their payloads. When there is no partition filter, all entries of the
+selected snapshot are decoded. Disabling lazy decoding also
 restores full entry decoding before partition filtering.
 
 For example, if a bucket contains 604 entries across 84 partitions and a query
-needs six entries from one partition, the scan inspects 604 partition fields
-but fully decodes only six entries. This reduces metadata allocation and
-statistics decoding without introducing another cache or changing the
-serialized cache format.
-
-The filtered result is request-local. Manifest Arrow caches retain complete
-batches, and snapshot cache misses rebuild the complete bucket before storing
-it. A later scan with another partition filter or snapshot cannot reuse an
-incomplete result.
+needs six entries from one partition, an indexed cache hit tests 84 partitions
+and fully decodes only the six matching entries. File and snapshot-live caches
+reuse this indexed representation: file values retain raw ADD/DELETE entries,
+while snapshot values retain merged live entries and also avoid repeating the
+snapshot merge. Their cache kinds and keys remain separate.
+
+The filtered result is request-local. Data-manifest caches retain complete files,
+and snapshot cache misses rebuild the complete bucket before storing it. A later
+scan with another partition filter or snapshot cannot reuse an incomplete result.
 The existing snapshot retention limit also applies to selective reads.
 
 Configuration

src/paimon/CMakeLists.txt

@@ -320,6 +320,7 @@ set(PAIMON_CORE_SRCS
     core/manifest/file_source.cpp
     core/manifest/index_manifest_entry_serializer.cpp
     core/manifest/index_manifest_file.cpp
+    core/manifest/indexed_manifest_entries.cpp
     core/manifest/manifest_entry.cpp
     core/manifest/manifest_entry_serializer.cpp
     core/manifest/manifest_entry_writer.cpp
@@ -842,6 +843,7 @@ if(PAIMON_BUILD_TESTS)
                     core/manifest/manifest_committable_test.cpp
                     core/manifest/manifest_entry_serializer_test.cpp
                     core/manifest/manifest_file_meta_serializer_test.cpp
+                    core/manifest/indexed_manifest_entries_test.cpp
                     core/manifest/manifest_file_test.cpp
                     core/manifest/manifest_row_range_test.cpp
                     core/manifest/manifest_list_test.cpp

src/paimon/core/manifest/indexed_manifest_entries.cpp

@@ -0,0 +1,244 @@
+/*
+ * 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.
+ */
+
+#include "paimon/core/manifest/indexed_manifest_entries.h"
+
+#include <algorithm>
+#include <cstring>
+#include <limits>
+
+#include "paimon/common/io/memory_segment_output_stream.h"
+#include "paimon/common/utils/serialization_utils.h"
+#include "paimon/core/manifest/manifest_entry_serializer.h"
+#include "paimon/io/byte_array_input_stream.h"
+#include "paimon/io/data_input_stream.h"
+#include "paimon/memory/bytes.h"
+
+namespace paimon {
+namespace {
+constexpr int32_t kMagic = 0x494d4501;  // IME, cache format version 1
+constexpr int32_t kNumFields = 6;
+
+Result<MemorySegment> ReadSlice(DataInputStream* in, const MemorySegment& segment) {
+    PAIMON_ASSIGN_OR_RAISE(int32_t size, in->ReadValue<int32_t>());
+    PAIMON_ASSIGN_OR_RAISE(int64_t pos, in->GetPos());
+    if (size < 0 || size > segment.Size() - pos) {
+        return Status::Invalid("invalid indexed manifest block length");
+    }
+    PAIMON_RETURN_NOT_OK(in->Seek(pos + size));
+    return MemorySegment::WrapView(segment.Data() + pos, size);
+}
+}  // namespace
+
+Status IndexedManifestEntries::Add(const BinaryRow& row) {
+    if (row.GetFieldCount() != kNumFields || entry_count_ == std::numeric_limits<int32_t>::max()) {
+        return Status::Invalid("invalid indexed manifest row or entry count");
+    }
+    if (ManifestEntrySerializer::ValidateVersion(entry_version_).ok()) {
+        entry_version_ = row.GetInt(0);
+    }
+    // Validate the partition before publishing an index that can skip an entire group.
+    PAIMON_ASSIGN_OR_RAISE(BinaryRow partition, ManifestEntrySerializer::GetPartitionView(row));
+    (void)partition;
+    GroupKey key(std::string(row.GetStringView(2)),
+                 row.IsNullAt(3) ? std::nullopt : std::optional<int32_t>(row.GetInt(3)),
+                 row.IsNullAt(4) ? std::nullopt : std::optional<int32_t>(row.GetInt(4)));
+    auto [iter, inserted] = groups_.try_emplace(std::move(key), pool_);
+    auto& group = iter->second;
+    group.rows.WriteValue<int32_t>(entry_count_++);
+    group.rows.WriteValue<int32_t>(row.GetSizeInBytes());
+    group.rows.Write(row.GetSegment(), row.GetOffset(), row.GetSizeInBytes());
+    ++group.count;
+    return Status::OK();
+}
+
+Result<std::shared_ptr<Bytes>> IndexedManifestEntries::CopyBytes(
+    const MemorySegmentOutputStream& out, const std::shared_ptr<MemoryPool>& pool) {
+    if (out.CurrentSize() > std::numeric_limits<int32_t>::max()) {
+        return Status::Invalid("indexed manifest cache exceeds maximum segment size");
+    }
+    struct OwnedBytes {
+        std::shared_ptr<MemoryPool> pool;
+        std::shared_ptr<Bytes> bytes;
+    };
+    auto owner = std::make_shared<OwnedBytes>();
+    owner->pool = pool;
+    owner->bytes = Bytes::AllocateBytes(out.CurrentSize(), pool.get());
+    int64_t offset = 0;
+    for (const auto& segment : out.Segments()) {
+        const int64_t length = std::min<int64_t>(segment.Size(), out.CurrentSize() - offset);
+        if (length == 0) {
+            break;
+        }
+        std::memcpy(owner->bytes->data() + offset, segment.Data(), length);
+        offset += length;
+    }
+    auto* bytes = owner->bytes.get();
+    return std::shared_ptr<Bytes>(std::move(owner), bytes);
+}
+
+Result<std::shared_ptr<Bytes>> IndexedManifestEntries::Serialize(
+    const std::shared_ptr<MemoryPool>& pool) const {
+    MemorySegmentOutputStream out(MemorySegmentOutputStream::DEFAULT_SEGMENT_SIZE, pool);
+    out.WriteValue<int32_t>(kMagic);
+    out.WriteValue<int32_t>(entry_version_);
+    out.WriteValue<int32_t>(entry_count_);
+    out.WriteValue<int32_t>(groups_.size());
+    for (const auto& [key, group] : groups_) {
+        const auto& [partition, bucket, total_buckets] = key;
+        out.WriteValue<int32_t>(partition.size());
+        out.Write(partition.data(), partition.size());
+        out.WriteValue<int8_t>((bucket ? 1 : 0) | (total_buckets ? 2 : 0));
+        out.WriteValue<int32_t>(bucket.value_or(0));
+        out.WriteValue<int32_t>(total_buckets.value_or(0));
+        out.WriteValue<int32_t>(group.count);
+        const int64_t block_size = group.rows.CurrentSize();
+        if (block_size > std::numeric_limits<int32_t>::max() ||
+            out.CurrentSize() + sizeof(int32_t) + block_size >
+                std::numeric_limits<int32_t>::max()) {
+            return Status::Invalid("indexed manifest cache exceeds maximum segment size");
+        }
+        out.WriteValue<int32_t>(block_size);
+        int64_t remaining = block_size;
+        for (const auto& segment : group.rows.Segments()) {
+            const int32_t length = std::min<int64_t>(remaining, segment.Size());
+            out.Write(segment, 0, length);
+            remaining -= length;
+        }
+    }
+    return CopyBytes(out, pool);
+}
+
+Result<std::shared_ptr<Bytes>> IndexedManifestEntries::Serialize(
+    const std::vector<ManifestEntry>& entries, const std::shared_ptr<MemoryPool>& pool) {
+    IndexedManifestEntries index(pool);
+    ManifestEntrySerializer serializer(pool);
+    for (const auto& entry : entries) {
+        if (entry.File()->creation_time.GetNanoOfMillisecond() != 0) {
+            return Status::Invalid("manifest cache requires millisecond creation_time");
+        }
+        PAIMON_ASSIGN_OR_RAISE(BinaryRow row, serializer.ToRow(entry));
+        PAIMON_RETURN_NOT_OK(index.Add(row));
+    }
+    return index.Serialize(pool);
+}
+
+Result<uint64_t> IndexedManifestEntries::Read(
+    const MemorySegment& segment, const Filter& filter,
+    const std::function<Status(const InternalRow&)>& consumer, bool* consumer_or_version_error) {
+    if (consumer_or_version_error) {
+        *consumer_or_version_error = false;
+    }
+    if (!segment.Data() || segment.Size() < 4 * static_cast<int32_t>(sizeof(int32_t))) {
+        return Status::Invalid("truncated indexed manifest entries header");
+    }
+    auto semantic_status = [consumer_or_version_error](const Status& status) -> Status {
+        if (!status.ok() && consumer_or_version_error) {
+            *consumer_or_version_error = true;
+        }
+        return status;
+    };
+    auto stream = std::make_shared<ByteArrayInputStream>(segment.Data(), segment.Size());
+    DataInputStream in(stream);
+    PAIMON_ASSIGN_OR_RAISE(int32_t magic, in.ReadValue<int32_t>());
+    if (magic != kMagic) {
+        return Status::Invalid("invalid indexed manifest entries magic");
+    }
+    // Preserve version validation even for groups excluded by bucket or partition pruning.
+    PAIMON_ASSIGN_OR_RAISE(int32_t entry_version, in.ReadValue<int32_t>());
+    PAIMON_RETURN_NOT_OK(semantic_status(ManifestEntrySerializer::ValidateVersion(entry_version)));
+    PAIMON_ASSIGN_OR_RAISE(int32_t entry_count, in.ReadValue<int32_t>());
+    PAIMON_ASSIGN_OR_RAISE(int32_t group_count, in.ReadValue<int32_t>());
+    if (entry_count < 0 || group_count < 0 || group_count > entry_count) {
+        return Status::Invalid("invalid indexed manifest entry or group count");
+    }
+    std::vector<std::pair<int32_t, BinaryRow>> selected;
+    std::optional<std::string_view> previous_partition;
+    bool partition_matches = true;
+    int64_t counted = 0;
+    for (int32_t i = 0; i < group_count; ++i) {
+        PAIMON_ASSIGN_OR_RAISE(MemorySegment partition_bytes, ReadSlice(&in, segment));
+        std::string_view partition(partition_bytes.Data(), partition_bytes.Size());
+        PAIMON_ASSIGN_OR_RAISE(BinaryRow partition_row,
+                               SerializationUtils::DeserializeBinaryRowView(partition));
+        PAIMON_ASSIGN_OR_RAISE(int8_t flags, in.ReadValue<int8_t>());
+        PAIMON_ASSIGN_OR_RAISE(int32_t bucket, in.ReadValue<int32_t>());
+        PAIMON_ASSIGN_OR_RAISE(int32_t total_buckets, in.ReadValue<int32_t>());
+        PAIMON_ASSIGN_OR_RAISE(int32_t count, in.ReadValue<int32_t>());
+        PAIMON_ASSIGN_OR_RAISE(MemorySegment block, ReadSlice(&in, segment));
+        if (flags < 0 || flags > 3 || count <= 0 || count > block.Size() / 8 ||
+            counted + count > entry_count) {
+            return Status::Invalid("invalid indexed manifest group");
+        }
+        counted += count;
+        const bool historical =
+            filter.total_buckets && (flags != 3 || total_buckets != *filter.total_buckets);
+        if (filter.bucket && !historical && bucket != *filter.bucket) {
+            continue;
+        }
+        // Groups are ordered by partition first; test each selected partition only once.
+        if (filter.partition && (!previous_partition || *previous_partition != partition)) {
+            auto matches = filter.partition(partition_row);
+            PAIMON_RETURN_NOT_OK(semantic_status(matches.status()));
+            partition_matches = matches.value();
+            previous_partition = partition;
+        }
+        if (!partition_matches) {
+            continue;
+        }
+        auto row_stream = std::make_shared<ByteArrayInputStream>(block.Data(), block.Size());
+        DataInputStream rows(row_stream);
+        for (int32_t j = 0; j < count; ++j) {
+            PAIMON_ASSIGN_OR_RAISE(int32_t ordinal, rows.ReadValue<int32_t>());
+            PAIMON_ASSIGN_OR_RAISE(MemorySegment bytes, ReadSlice(&rows, block));
+            if (ordinal < 0 || ordinal >= entry_count ||
+                bytes.Size() <
+                    BinaryRow::CalculateBitSetWidthInBytes(kNumFields) + kNumFields * 8) {
+                return Status::Invalid("invalid indexed manifest row");
+            }
+            BinaryRow row(kNumFields);
+            row.PointTo(bytes, 0, bytes.Size());
+            PAIMON_RETURN_NOT_OK(
+                semantic_status(ManifestEntrySerializer::ValidateVersion(row.GetInt(0))));
+            selected.emplace_back(ordinal, std::move(row));
+        }
+        PAIMON_ASSIGN_OR_RAISE(int64_t end, rows.GetPos());
+        if (end != block.Size()) {
+            return Status::Invalid("trailing indexed manifest group bytes");
+        }
+    }
+    PAIMON_ASSIGN_OR_RAISE(int64_t end, in.GetPos());
+    if (counted != entry_count || end != segment.Size()) {
+        return Status::Invalid("invalid indexed manifest entries length or count");
+    }
+    std::sort(selected.begin(), selected.end(),
+              [](const auto& lhs, const auto& rhs) { return lhs.first < rhs.first; });
+    int32_t previous_ordinal = -1;
+    for (const auto& [ordinal, row] : selected) {
+        if (ordinal == previous_ordinal) {
+            return Status::Invalid("duplicate indexed manifest ordinal");
+        }
+        previous_ordinal = ordinal;
+    }
+    for (const auto& [ordinal, row] : selected) {
+        PAIMON_RETURN_NOT_OK(semantic_status(consumer(row)));
+    }
+    return static_cast<uint64_t>(entry_count);
+}
+}  // namespace paimon