perf(scan): prune manifest partitions before metadata decoding

docs/source/user_guide/manifest_entry_cache.rst

@@ -36,6 +36,37 @@ Request-specific filters are not stored in the cache. Partition, level, and
 predicate filters are still evaluated for each scan, so cached entries can be
 reused safely across different scan predicates for the same bucket.
 
+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.
+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
+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.
+The existing snapshot retention limit also applies to selective reads.
+
 Configuration
 -------------
 
@@ -67,11 +98,11 @@ the cache path. Set it to a positive value to enable the cache when
 is still controlled by the configured ``Cache`` implementation, for example the
 capacity of ``LruCache``.
 
-If no cache is provided through ``ScanContextBuilder::WithCache()``, this
-optimization is skipped. The snapshot manifest entry cache shares the same
-``Cache`` interface with raw manifest and data-file footer caches, but it uses a
-dedicated ``CacheKind`` and a table/branch/bucket key instead of file byte
-ranges.
+If no cache is provided through ``ScanContextBuilder::WithCache()``, snapshot
+caching is skipped; selective manifest decoding remains available. The snapshot
+manifest entry cache shares the same ``Cache`` interface with raw manifest and
+data-file footer caches, but it uses a dedicated ``CacheKind`` and a
+table/branch/bucket key instead of file byte ranges.
 
 Limitations
 -----------

include/paimon/defs.h

@@ -224,7 +224,8 @@ struct PAIMON_EXPORT Options {
     static const char SCAN_MANIFEST_ENTRY_CACHE_MAX_SNAPSHOTS[];
 
     /// "scan.manifest-entry.lazy-decode.enabled" - Whether to deserialize only manifest entries
-    /// for the target bucket when rebuilding the cache. Default value is true.
+    /// selected by bucket, partition, and row-range pruning before constructing file metadata.
+    /// Applies to supported manifest read paths with or without caching. Default value is true.
     static const char SCAN_MANIFEST_ENTRY_LAZY_DECODE_ENABLED[];
 
     /// "prefetch.io-metrics.enabled" - Whether to collect per-I/O metrics for prefetch reads.

src/paimon/common/data/serializer/binary_row_serializer.cpp

@@ -53,4 +53,16 @@ Result<BinaryRow> BinaryRowSerializer::Deserialize(DataInputStream* source) cons
     return row;
 }
 
+Status BinaryRowSerializer::Skip(DataInputStream* source) {
+    PAIMON_ASSIGN_OR_RAISE(int32_t read_length, source->ReadValue<int32_t>());
+    PAIMON_ASSIGN_OR_RAISE(int64_t pos, source->GetPos());
+    PAIMON_ASSIGN_OR_RAISE(int64_t length, source->Length());
+    if (read_length < 0 || pos < 0 || pos > length || read_length > length - pos) {
+        return Status::Invalid(
+            fmt::format("invalid binary row length {} at position {}, stream length {}",
+                        read_length, pos, length));
+    }
+    return source->Seek(pos + read_length);
+}
+
 }  // namespace paimon

src/paimon/common/data/serializer/binary_row_serializer.h

@@ -57,6 +57,9 @@ class BinaryRowSerializer {
 
     Result<BinaryRow> Deserialize(DataInputStream* source) const;
 
+    /// Skip a length-prefixed row without allocating or copying its payload.
+    static Status Skip(DataInputStream* source);
+
     Status Serialize(const BinaryRow& row, MemorySegmentOutputStream* target) const;
 
  private:

src/paimon/common/data/serializer/binary_row_serializer_test.cpp

@@ -70,4 +70,38 @@ TEST_F(BinaryRowSerializerTest, SerializeAndDeserialize) {
     EXPECT_EQ(deserialized_row.GetDouble(2), 3.14);
 }
 
+TEST_F(BinaryRowSerializerTest, SkipPreservesNextRow) {
+    BinaryRowSerializer serializer(1, pool_);
+    MemorySegmentOutputStream out(128, pool_);
+    for (int32_t value : {10, 20}) {
+        BinaryRow row(1);
+        BinaryRowWriter writer(&row, 0, pool_.get());
+        writer.WriteInt(0, value);
+        writer.Complete();
+        ASSERT_OK(serializer.Serialize(row, &out));
+    }
+    auto stream =
+        std::make_shared<ByteArrayInputStream>(out.Segments().front().Data(), out.CurrentSize());
+    DataInputStream in(stream);
+    ASSERT_OK(BinaryRowSerializer::Skip(&in));
+    ASSERT_OK_AND_ASSIGN(auto row, serializer.Deserialize(&in));
+    ASSERT_EQ(row.GetInt(0), 20);
+    ASSERT_OK_AND_ASSIGN(int64_t pos, in.GetPos());
+    ASSERT_EQ(pos, out.CurrentSize());
+    ASSERT_NOK(BinaryRowSerializer::Skip(&in));
+}
+
+TEST_F(BinaryRowSerializerTest, SkipRejectsInvalidLength) {
+    for (int32_t length : {-1, 5}) {
+        SCOPED_TRACE(length);
+        MemorySegmentOutputStream out(64, pool_);
+        out.WriteValue<int32_t>(length);
+        out.WriteValue<int32_t>(0);
+        auto stream = std::make_shared<ByteArrayInputStream>(out.Segments().front().Data(),
+                                                             out.CurrentSize());
+        DataInputStream in(stream);
+        ASSERT_NOK_WITH_MSG(BinaryRowSerializer::Skip(&in), "invalid binary row length");
+    }
+}
+
 }  // namespace paimon::test

src/paimon/common/utils/serialization_utils.h

@@ -21,7 +21,9 @@
 
 #include <cstdint>
 #include <cstring>
+#include <limits>
 #include <memory>
+#include <string_view>
 #include <utility>
 
 #include "fmt/format.h"
@@ -75,31 +77,53 @@ class SerializationUtils {
 
     /// Schemaless deserialization for `BinaryRow`.
     static Result<BinaryRow> DeserializeBinaryRow(const std::shared_ptr<Bytes>& bytes) {
-        if (PAIMON_UNLIKELY(bytes->size() < 4)) {
-            return Status::Invalid(fmt::format("bytes size {} is less than 4", bytes->size()));
+        return DeserializeBinaryRow(MemorySegment::Wrap(bytes));
+    }
+
+    /// Borrow a schemaless row. The caller must keep the backing bytes alive while using it.
+    static Result<BinaryRow> DeserializeBinaryRowView(std::string_view bytes) {
+        if (PAIMON_UNLIKELY(bytes.size() > std::numeric_limits<int32_t>::max())) {
+            return Status::Invalid("binary row exceeds maximum segment size");
+        }
+        return DeserializeBinaryRow(
+            MemorySegment::WrapView(bytes.data(), static_cast<int32_t>(bytes.size())));
+    }
+
+    /// Schemaless deserialization for `BinaryRow` from a `DataInputStream`.
+    static Result<BinaryRow> DeserializeBinaryRow(DataInputStream* input, MemoryPool* pool) {
+        int32_t read_length = -1;
+        PAIMON_ASSIGN_OR_RAISE(read_length, input->ReadValue<int32_t>());
+        std::shared_ptr<Bytes> bytes = Bytes::AllocateBytes(read_length, pool);
+        PAIMON_RETURN_NOT_OK(input->ReadBytes(bytes.get()));
+        return DeserializeBinaryRow(bytes);
+    }
+
+ private:
+    static Result<BinaryRow> DeserializeBinaryRow(const MemorySegment& segment) {
+        if (PAIMON_UNLIKELY(segment.Size() < 4)) {
+            return Status::Invalid(fmt::format("bytes size {} is less than 4", segment.Size()));
         }
         // The buffer is byte-filled, so memcpy avoids the strict-aliasing UB of reinterpret_cast.
         int32_t arity;
-        memcpy(&arity, bytes->data(), sizeof(int32_t));
+        memcpy(&arity, segment.Data(), sizeof(int32_t));
         if (SystemByteOrder() == ByteOrder::PAIMON_LITTLE_ENDIAN) {
             arity = EndianSwapValue(arity);
         }
         if (PAIMON_UNLIKELY(arity < 0)) {
             return Status::Invalid("arity is less than 0");
         }
+        const int32_t payload_size = segment.Size() - 4;
+        // Bound the arity before the bitset helper, then compute the total size without overflow.
+        if (PAIMON_UNLIKELY(arity > payload_size / 8 ||
+                            BinaryRow::CalculateBitSetWidthInBytes(arity) +
+                                    static_cast<int64_t>(arity) * 8 >
+                                payload_size)) {
+            return Status::Invalid("binary row fixed fields exceed serialized length");
+        }
         BinaryRow row(arity);
-        row.PointTo(MemorySegment::Wrap(bytes), 4, bytes->size() - 4);
+        row.PointTo(segment, 4, segment.Size() - 4);
         return row;
     }
-
-    /// Schemaless deserialization for `BinaryRow` from a `DataInputStream`.
-    static Result<BinaryRow> DeserializeBinaryRow(DataInputStream* input, MemoryPool* pool) {
-        int32_t read_length = -1;
-        PAIMON_ASSIGN_OR_RAISE(read_length, input->ReadValue<int32_t>());
-        std::shared_ptr<Bytes> bytes = Bytes::AllocateBytes(read_length, pool);
-        PAIMON_RETURN_NOT_OK(input->ReadBytes(bytes.get()));
-        return DeserializeBinaryRow(bytes);
-    }
 };
 
 }  // namespace paimon

src/paimon/common/utils/serialization_utils_test.cpp

@@ -85,4 +85,50 @@ TEST_F(SerializationUtilsTest, TestDeserializeBinaryRowFromStream) {
     ASSERT_EQ(de_row.GetString(2).ToString(), "hello paimon!");
 }
 
+TEST_F(SerializationUtilsTest, DeserializeBorrowedBinaryRow) {
+    auto pool = GetDefaultPool();
+    BinaryRow row(3);
+    BinaryRowWriter writer(&row, 0, pool.get());
+    writer.WriteInt(0, -42);
+    writer.SetNullAt(1);
+    writer.WriteString(2, BinaryString::FromString("partition", pool.get()));
+    writer.Complete();
+    auto bytes = SerializationUtils::SerializeBinaryRow(row, pool.get());
+    // Use an unaligned backing address, as Arrow binary values need not be aligned.
+    std::string backing = "x" + std::string(bytes->data(), bytes->size());
+    std::string_view data(backing.data() + 1, bytes->size());
+    ASSERT_OK_AND_ASSIGN(auto view, SerializationUtils::DeserializeBinaryRowView(data));
+    ASSERT_EQ(view.GetSegment().Data(), data.data());
+    ASSERT_EQ(view.GetInt(0), -42);
+    ASSERT_TRUE(view.IsNullAt(1));
+    ASSERT_EQ(view.GetString(2).ToString(), "partition");
+    ASSERT_EQ(view, row);
+
+    ASSERT_OK_AND_ASSIGN(auto owned, SerializationUtils::DeserializeBinaryRow(bytes));
+    bytes.reset();
+    backing.clear();
+    ASSERT_EQ(owned, row);
+    auto empty_bytes = SerializationUtils::SerializeBinaryRow(BinaryRow::EmptyRow(), pool.get());
+    ASSERT_OK_AND_ASSIGN(auto empty, SerializationUtils::DeserializeBinaryRowView(std::string_view(
+                                         empty_bytes->data(), empty_bytes->size())));
+    ASSERT_EQ(empty, BinaryRow::EmptyRow());
+}
+
+TEST_F(SerializationUtilsTest, RejectInvalidBorrowedBinaryRow) {
+    const std::string negative_arity(4, static_cast<char>(0xff));
+    ASSERT_NOK_WITH_MSG(SerializationUtils::DeserializeBinaryRowView(negative_arity),
+                        "arity is less than 0");
+    const std::string huge_arity = {0x7f, static_cast<char>(0xff), static_cast<char>(0xff),
+                                    static_cast<char>(0xff)};
+    ASSERT_NOK_WITH_MSG(SerializationUtils::DeserializeBinaryRowView(huge_arity),
+                        "fixed fields exceed serialized length");
+    auto bytes =
+        SerializationUtils::SerializeBinaryRow(BinaryRow::EmptyRow(), GetDefaultPool().get());
+    for (size_t size = 0; size < bytes->size(); ++size) {
+        SCOPED_TRACE(size);
+        ASSERT_NOK(
+            SerializationUtils::DeserializeBinaryRowView(std::string_view(bytes->data(), size)));
+    }
+}
+
 }  // namespace paimon::test

src/paimon/core/manifest/manifest_entry_serializer.cpp

@@ -48,6 +48,10 @@ int32_t ManifestEntrySerializer::GetBucket(const InternalRow& row) {
     return row.GetInt(3);
 }
 
+Result<BinaryRow> ManifestEntrySerializer::GetPartitionView(const InternalRow& row) {
+    return SerializationUtils::DeserializeBinaryRowView(row.GetStringView(2));
+}
+
 Result<ManifestEntry> ManifestEntrySerializer::ConvertFrom(int32_t version,
                                                            const InternalRow& row) const {
     PAIMON_RETURN_NOT_OK(ValidateVersion(version));

src/paimon/core/manifest/manifest_entry_serializer.h

@@ -56,6 +56,11 @@ class ManifestEntrySerializer : public VersionedObjectSerializer<ManifestEntry>
     /// Get the bucket from a versioned manifest entry row without fully deserializing it.
     static int32_t GetBucket(const InternalRow& row);
 
+    /// Borrow the partition without copying or constructing file metadata.
+    /// The caller must validate the entry version before accessing this field.
+    /// The row backing memory must remain valid while the returned view is used.
+    static Result<BinaryRow> GetPartitionView(const InternalRow& row);
+
     Result<ManifestEntry> ConvertFrom(int32_t version, const InternalRow& row) const override;
 
     Result<BinaryRow> ToRow(const ManifestEntry& record) const override;

src/paimon/core/manifest/manifest_entry_serializer_test.cpp

@@ -57,9 +57,15 @@ TEST_F(ManifestEntrySerializerTest, TestToFromRow) {
     for (const auto& entry : entries) {
         ASSERT_OK_AND_ASSIGN(auto row, serializer.ToRow(entry));
         ASSERT_EQ(entry.Bucket(), ManifestEntrySerializer::GetBucket(row));
+        ASSERT_OK_AND_ASSIGN(auto partition, ManifestEntrySerializer::GetPartitionView(row));
+        ASSERT_EQ(entry.Partition(), partition);
+        ASSERT_EQ(partition.GetSegment().Data(), row.GetStringView(2).data());
         ASSERT_OK_AND_ASSIGN(auto result_entry, serializer.FromRow(row));
         ASSERT_EQ(entry, result_entry);
         ASSERT_EQ(entry.ToString(), result_entry.ToString());
+        // The final entry owns its partition even after the source row is released.
+        row = BinaryRow::EmptyRow();
+        ASSERT_EQ(entry.Partition(), result_entry.Partition());
     }
 }