diff --git a/docs/source/user_guide/manifest_entry_cache.rst b/docs/source/user_guide/manifest_entry_cache.rst
index 97b33cdec..0f37138c2 100644
--- a/docs/source/user_guide/manifest_entry_cache.rst
+++ b/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
 -----------
diff --git a/include/paimon/defs.h b/include/paimon/defs.h
index 8c88c3de6..88cb78bda 100644
--- a/include/paimon/defs.h
+++ b/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.
diff --git a/src/paimon/common/data/serializer/binary_row_serializer.cpp b/src/paimon/common/data/serializer/binary_row_serializer.cpp
index 56075ac0d..5c406a4e9 100644
--- a/src/paimon/common/data/serializer/binary_row_serializer.cpp
+++ b/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
diff --git a/src/paimon/common/data/serializer/binary_row_serializer.h b/src/paimon/common/data/serializer/binary_row_serializer.h
index b1f0a05a0..1b8b90f38 100644
--- a/src/paimon/common/data/serializer/binary_row_serializer.h
+++ b/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:
diff --git a/src/paimon/common/data/serializer/binary_row_serializer_test.cpp b/src/paimon/common/data/serializer/binary_row_serializer_test.cpp
index 51a6d3aaa..298d77fa9 100644
--- a/src/paimon/common/data/serializer/binary_row_serializer_test.cpp
+++ b/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
diff --git a/src/paimon/common/utils/serialization_utils.h b/src/paimon/common/utils/serialization_utils.h
index 449766ca6..8c5d7a736 100644
--- a/src/paimon/common/utils/serialization_utils.h
+++ b/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
diff --git a/src/paimon/common/utils/serialization_utils_test.cpp b/src/paimon/common/utils/serialization_utils_test.cpp
index 56a39a295..3df436a96 100644
--- a/src/paimon/common/utils/serialization_utils_test.cpp
+++ b/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
diff --git a/src/paimon/core/manifest/manifest_entry_serializer.cpp b/src/paimon/core/manifest/manifest_entry_serializer.cpp
index 5e1a2be08..fec2b6a50 100644
--- a/src/paimon/core/manifest/manifest_entry_serializer.cpp
+++ b/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));
diff --git a/src/paimon/core/manifest/manifest_entry_serializer.h b/src/paimon/core/manifest/manifest_entry_serializer.h
index 9d4ddcfa7..5f334aadf 100644
--- a/src/paimon/core/manifest/manifest_entry_serializer.h
+++ b/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;
diff --git a/src/paimon/core/manifest/manifest_entry_serializer_test.cpp b/src/paimon/core/manifest/manifest_entry_serializer_test.cpp
index a31f4bf73..2838829e6 100644
--- a/src/paimon/core/manifest/manifest_entry_serializer_test.cpp
+++ b/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());
     }
 }
 
diff --git a/src/paimon/core/manifest/manifest_file.cpp b/src/paimon/core/manifest/manifest_file.cpp
index c0a71a2f8..f47a63a9c 100644
--- a/src/paimon/core/manifest/manifest_file.cpp
+++ b/src/paimon/core/manifest/manifest_file.cpp
@@ -134,53 +134,69 @@ Status ManifestFile::ReadBucketEntries(const std::string& file_name, int32_t buc
         });
 }
 
-Status ManifestFile::ReadRowRangeEntries(
-    const std::string& file_name, const RowRangeIndex& row_ranges,
-    const std::function<Result<bool>(const ManifestEntry&)>& filter,
-    std::optional<int64_t> file_size, std::vector<ManifestEntry>* entries) const {
+Status ManifestFile::ReadSelectedEntries(
+    const std::string& file_name, std::optional<int64_t> file_size, const RowRangeIndex* row_ranges,
+    const std::function<Result<bool>(const InternalRow&)>& row_filter,
+    const std::function<Result<bool>(const ManifestEntry&)>& entry_filter,
+    std::vector<ManifestEntry>* entries) const {
+    if (!row_ranges && !row_filter) {
+        return Read(file_name, entry_filter, file_size, entries);
+    }
     return ReadArrowBatches(
         file_name, file_size,
-        [this, &row_ranges, &filter,
+        [this, row_ranges, &row_filter, &entry_filter,
          entries](const std::shared_ptr<arrow::StructArray>& batch) -> Status {
-            // ManifestMetaReader has aligned both the entry and its nested file schema. Probe
-            // the two range columns without allocating DataFileMeta, stats or binary keys.
-            // The serialized entry has a leading _VERSION field.
-            const auto& file_column = batch->field(ManifestEntry::kFileFieldIndex + 1);
-            if (file_column->type_id() != arrow::Type::STRUCT) {
-                return Status::Invalid("Manifest entry file metadata must be a struct");
-            }
-            auto files = checked_pointer_cast<arrow::StructArray>(file_column);
-            const auto& count_column = files->field(DataFileMeta::kRowCountFieldIndex);
-            const auto& first_column = files->field(DataFileMeta::kFirstRowIdFieldIndex);
-            if (count_column->type_id() != arrow::Type::INT64 ||
-                first_column->type_id() != arrow::Type::INT64) {
-                return Status::Invalid("Manifest entry row range must contain int64 fields");
+            // ManifestMetaReader has aligned both the entry and its nested file schema.
+            // Only row-range scans need to probe the file columns before entry decoding.
+            std::shared_ptr<arrow::StructArray> files;
+            std::shared_ptr<arrow::Int64Array> counts;
+            std::shared_ptr<arrow::Int64Array> first_ids;
+            if (row_ranges) {
+                const auto& file_column = batch->field(ManifestEntry::kFileFieldIndex + 1);
+                if (file_column->type_id() != arrow::Type::STRUCT) {
+                    return Status::Invalid("Manifest entry file metadata must be a struct");
+                }
+                files = checked_pointer_cast<arrow::StructArray>(file_column);
+                const auto& count_column = files->field(DataFileMeta::kRowCountFieldIndex);
+                const auto& first_column = files->field(DataFileMeta::kFirstRowIdFieldIndex);
+                if (count_column->type_id() != arrow::Type::INT64 ||
+                    first_column->type_id() != arrow::Type::INT64) {
+                    return Status::Invalid("Manifest entry row range must contain int64 fields");
+                }
+                counts = checked_pointer_cast<arrow::Int64Array>(count_column);
+                first_ids = checked_pointer_cast<arrow::Int64Array>(first_column);
             }
-            auto counts = checked_pointer_cast<arrow::Int64Array>(count_column);
-            auto first_ids = checked_pointer_cast<arrow::Int64Array>(first_column);
             ColumnarRow row(batch->fields(), pool_, /*row_id=*/0);
             for (int64_t i = 0; i < batch->length(); ++i) {
                 row.SetRowId(i);
                 PAIMON_RETURN_NOT_OK(
                     ManifestEntrySerializer::ValidateVersion(row.GetInt(kVersionFieldIndex)));
-                if (files->IsNull(i)) {
-                    return Status::Invalid(
-                        "ManifestEntry convert from row failed, with null DataFileMeta");
+                if (row_ranges) {
+                    if (files->IsNull(i)) {
+                        return Status::Invalid(
+                            "ManifestEntry convert from row failed, with null DataFileMeta");
+                    }
+                    if (!first_ids->IsNull(i) && !counts->IsNull(i)) {
+                        const int64_t first = first_ids->Value(i);
+                        const int64_t count = counts->Value(i);
+                        // Missing or invalid ranges cannot prove an entry is irrelevant. In
+                        // particular, do not overflow when computing the inclusive upper bound.
+                        if (first >= 0 && count > 0 &&
+                            first <= std::numeric_limits<int64_t>::max() - (count - 1) &&
+                            !row_ranges->Intersects(first, first + (count - 1))) {
+                            continue;
+                        }
+                    }
                 }
-                if (!first_ids->IsNull(i) && !counts->IsNull(i)) {
-                    const int64_t first = first_ids->Value(i);
-                    const int64_t count = counts->Value(i);
-                    // Missing or invalid ranges cannot prove an entry is irrelevant. In
-                    // particular, do not overflow when computing the inclusive upper bound.
-                    if (first >= 0 && count > 0 &&
-                        first <= std::numeric_limits<int64_t>::max() - (count - 1) &&
-                        !row_ranges.Intersects(first, first + (count - 1))) {
+                if (row_filter) {
+                    PAIMON_ASSIGN_OR_RAISE(bool keep, row_filter(row));
+                    if (!keep) {
                         continue;
                     }
                 }
                 PAIMON_ASSIGN_OR_RAISE(ManifestEntry entry, serializer_->FromRow(row));
-                if (filter) {
-                    PAIMON_ASSIGN_OR_RAISE(bool keep, filter(entry));
+                if (entry_filter) {
+                    PAIMON_ASSIGN_OR_RAISE(bool keep, entry_filter(entry));
                     if (!keep) {
                         continue;
                     }
diff --git a/src/paimon/core/manifest/manifest_file.h b/src/paimon/core/manifest/manifest_file.h
index c37882b54..22ffba497 100644
--- a/src/paimon/core/manifest/manifest_file.h
+++ b/src/paimon/core/manifest/manifest_file.h
@@ -78,13 +78,16 @@ class ManifestFile : public ObjectsFile<ManifestEntry> {
                              std::optional<int64_t> file_size,
                              std::vector<ManifestEntry>* entries) const;
 
-    /// Read entries intersecting row ID ranges before constructing their file metadata.
-    /// Unknown row ranges are retained. Add and Delete entries use the same selection, and
-    /// the ordinary entry filter still runs on every retained entry.
-    Status ReadRowRangeEntries(const std::string& file_name, const RowRangeIndex& row_ranges,
-                               const std::function<Result<bool>(const ManifestEntry&)>& filter,
-                               std::optional<int64_t> file_size,
-                               std::vector<ManifestEntry>* entries) const;
+    /// Select entries by row ranges and a row filter before constructing file metadata.
+    /// A null row_ranges or row_filter disables that selection. Unknown ranges are retained.
+    /// Add and Delete entries use the same selection. The entry filter runs after decoding.
+    /// The reader validates the entry version before inspecting row ranges or invoking the filter.
+    Status ReadSelectedEntries(
+        const std::string& file_name, std::optional<int64_t> file_size,
+        const RowRangeIndex* row_ranges,
+        const std::function<Result<bool>(const InternalRow&)>& row_filter,
+        const std::function<Result<bool>(const ManifestEntry&)>& entry_filter,
+        std::vector<ManifestEntry>* entries) const;
 
  private:
     Status PrepareBucketRead(int32_t bucket, const std::optional<int32_t>& expected_total_buckets,
diff --git a/src/paimon/core/manifest/manifest_row_range_test.cpp b/src/paimon/core/manifest/manifest_row_range_test.cpp
index a4f60fea1..db2dd75cd 100644
--- a/src/paimon/core/manifest/manifest_row_range_test.cpp
+++ b/src/paimon/core/manifest/manifest_row_range_test.cpp
@@ -46,6 +46,7 @@
 #include "paimon/format/format_writer.h"
 #include "paimon/format/writer_builder.h"
 #include "paimon/fs/local/local_file_system.h"
+#include "paimon/testing/utils/binary_row_generator.h"
 #include "paimon/testing/utils/counting_cache_test_utils.h"
 #include "paimon/testing/utils/testharness.h"
 #include "paimon/utils/row_range_index.h"
@@ -106,6 +107,65 @@ class RowRangeManifestFileTest : public ::testing::Test {
     }
 };
 
+TEST_F(RowRangeManifestFileTest, PartitionFilterComposesWithRowRanges) {
+    for (bool cache_enabled : {false, true}) {
+        SCOPED_TRACE(cache_enabled);
+        auto dir = UniqueTestDirectory::Create();
+        ASSERT_TRUE(dir);
+        ASSERT_OK_AND_ASSIGN(std::unique_ptr<ManifestFile> manifest,
+                             CreateManifest(dir->Str(), dir->GetFileSystem(), cache_enabled));
+        auto pool = GetDefaultPool();
+        auto partition10 = BinaryRowGenerator::GenerateRow({10}, pool.get());
+        auto partition20 = BinaryRowGenerator::GenerateRow({20}, pool.get());
+        ASSERT_OK_AND_ASSIGN(ManifestEntry first, Entry("a.parquet", 100, 10));
+        ASSERT_OK_AND_ASSIGN(ManifestEntry second, Entry("b.parquet", 100, 10));
+        ASSERT_OK_AND_ASSIGN(ManifestEntry third, Entry("c.parquet", 200, 10));
+        ManifestEntry a(FileKind::Add(), partition10, 0, 1, first.File());
+        ManifestEntry deleted(FileKind::Delete(), partition10, 0, 1, first.File());
+        // Decoding the excluded partition would reject its invalid file kind.
+        ManifestEntry b(FileKind(static_cast<int8_t>(2)), partition20, 0, 1, second.File());
+        ManifestEntry c(FileKind::Add(), partition10, 0, 1, third.File());
+        std::vector<ManifestEntry> source = {a, deleted, b, c};
+        using WrittenFile = std::pair<std::string, int64_t>;
+        ASSERT_OK_AND_ASSIGN(WrittenFile written, manifest->WriteWithoutRolling(source));
+        auto row_filter = [](const InternalRow& row) -> Result<bool> {
+            PAIMON_ASSIGN_OR_RAISE(BinaryRow partition,
+                                   ManifestEntrySerializer::GetPartitionView(row));
+            return partition.GetInt(0) == 10;
+        };
+        std::vector<ManifestEntry> selected;
+        ASSERT_OK(manifest->ReadSelectedEntries(
+            written.first, written.second, /*row_ranges=*/nullptr, row_filter, nullptr, &selected));
+        ASSERT_EQ(std::vector<ManifestEntry>({a, deleted, c}), selected);
+        std::vector<ManifestEntry> live;
+        ASSERT_OK(FileStoreScan::MergeLiveEntries(selected, &live));
+        ASSERT_EQ(std::vector<ManifestEntry>({c}), live);
+
+        ASSERT_OK_AND_ASSIGN(RowRangeIndex ranges, RowRangeIndex::Create({Range(100, 100)}));
+        selected.clear();
+        ASSERT_OK(manifest->ReadSelectedEntries(written.first, written.second, &ranges, row_filter,
+                                                nullptr, &selected));
+        ASSERT_EQ(std::vector<ManifestEntry>({a, deleted}), selected);
+        live.clear();
+        ASSERT_OK(FileStoreScan::MergeLiveEntries(selected, &live));
+        ASSERT_TRUE(live.empty());
+
+        auto error_filter = [](const InternalRow&) -> Result<bool> {
+            return Status::IOError("injected row filter error");
+        };
+        selected.clear();
+        ASSERT_NOK_WITH_MSG(
+            manifest->ReadSelectedEntries(written.first, written.second, /*row_ranges=*/nullptr,
+                                          error_filter, nullptr, &selected),
+            "injected row filter error");
+        // The unfiltered path still decodes the excluded entry from the complete batches.
+        selected.clear();
+        ASSERT_NOK_WITH_MSG(manifest->ReadSelectedEntries(written.first, written.second, nullptr,
+                                                          nullptr, nullptr, &selected),
+                            "Unsupported byte value 2 for file kind.");
+    }
+}
+
 TEST_F(RowRangeManifestFileTest, ArrowCacheReuseEvictionAndConcurrentReaders) {
     auto dir = UniqueTestDirectory::Create();
     ASSERT_TRUE(dir);
@@ -123,8 +183,8 @@ TEST_F(RowRangeManifestFileTest, ArrowCacheReuseEvictionAndConcurrentReaders) {
         PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<ManifestFile> reader,
                                CreateManifest(dir->Str(), dir->GetFileSystem(), true, cache));
         std::vector<ManifestEntry> entries;
-        PAIMON_RETURN_NOT_OK(
-            reader->ReadRowRangeEntries(written.first, ranges, nullptr, written.second, &entries));
+        PAIMON_RETURN_NOT_OK(reader->ReadSelectedEntries(
+            written.first, written.second, &ranges, /*row_filter=*/nullptr, nullptr, &entries));
         if (entries != expected) {
             return Status::Invalid("cached manifest result differs from source");
         }
@@ -148,7 +208,8 @@ TEST_F(RowRangeManifestFileTest, ArrowCacheReuseEvictionAndConcurrentReaders) {
     // A different immutable manifest must not reuse the previous file's cached content.
     ASSERT_OK_AND_ASSIGN(WrittenFile next, manifest->WriteWithoutRolling({a}));
     std::vector<ManifestEntry> entries;
-    ASSERT_OK(manifest->ReadRowRangeEntries(next.first, ranges, nullptr, next.second, &entries));
+    ASSERT_OK(manifest->ReadSelectedEntries(next.first, next.second, &ranges,
+                                            /*row_filter=*/nullptr, nullptr, &entries));
     ASSERT_TRUE(entries.empty());
     ASSERT_EQ(3, cache->SupplierCallCount(cache_kind));
 
@@ -156,15 +217,15 @@ TEST_F(RowRangeManifestFileTest, ArrowCacheReuseEvictionAndConcurrentReaders) {
         std::make_shared<CountingRoutingCache>(CacheKind::DATA_FILE_FOOTER, 1024 * 1024);
     ASSERT_OK_AND_ASSIGN(std::unique_ptr<ManifestFile> unsupported,
                          CreateManifest(dir->Str(), dir->GetFileSystem(), true, unsupported_cache));
-    ASSERT_OK(
-        unsupported->ReadRowRangeEntries(written.first, ranges, nullptr, written.second, &entries));
+    ASSERT_OK(unsupported->ReadSelectedEntries(written.first, written.second, &ranges,
+                                               /*row_filter=*/nullptr, nullptr, &entries));
     ASSERT_EQ(expected, entries);
     entries.clear();
     auto small_cache = std::make_shared<LruCache>(1);
     ASSERT_OK_AND_ASSIGN(std::unique_ptr<ManifestFile> uncached,
                          CreateManifest(dir->Str(), dir->GetFileSystem(), true, small_cache));
-    ASSERT_OK(
-        uncached->ReadRowRangeEntries(written.first, ranges, nullptr, written.second, &entries));
+    ASSERT_OK(uncached->ReadSelectedEntries(written.first, written.second, &ranges,
+                                            /*row_filter=*/nullptr, nullptr, &entries));
     ASSERT_EQ(expected, entries);
     ASSERT_LE(small_cache->GetCurrentWeight(), 1);
     small_cache->InvalidateAll();
@@ -302,7 +363,8 @@ TEST_F(RowRangeManifestFileTest, OrcCachePreservesManifestMetadata) {
     // This legacy manifest has unknown row IDs, so every entry must be retained.
     ASSERT_OK_AND_ASSIGN(RowRangeIndex ranges, RowRangeIndex::Create({Range(0, 0)}));
     std::vector<ManifestEntry> without_cache;
-    ASSERT_OK(uncached->ReadRowRangeEntries(name, ranges, nullptr, std::nullopt, &without_cache));
+    ASSERT_OK(uncached->ReadSelectedEntries(name, std::nullopt, &ranges, /*row_filter=*/nullptr,
+                                            nullptr, &without_cache));
     ASSERT_EQ(expected, without_cache);
 
     auto cache = std::make_shared<CountingRoutingCache>(CacheKind::MANIFEST, 1024 * 1024);
@@ -313,7 +375,8 @@ TEST_F(RowRangeManifestFileTest, OrcCachePreservesManifestMetadata) {
             std::unique_ptr<ManifestFile> reader,
             ManifestFile::Create(fs, format, "zstd", paths, 1024, pool, options, schema));
         std::vector<ManifestEntry> actual;
-        ASSERT_OK(reader->ReadRowRangeEntries(name, ranges, nullptr, std::nullopt, &actual));
+        ASSERT_OK(reader->ReadSelectedEntries(name, std::nullopt, &ranges, /*row_filter=*/nullptr,
+                                              nullptr, &actual));
         ASSERT_EQ(expected, actual);
         ASSERT_EQ(1, cache->SupplierCallCount(CacheKind::MANIFEST));
     }
@@ -330,8 +393,8 @@ TEST_F(RowRangeManifestFileTest, EmptyManifestCacheReuse) {
     ASSERT_OK_AND_ASSIGN(RowRangeIndex ranges, RowRangeIndex::Create({Range(0, 0)}));
     for (int32_t attempt = 0; attempt < 2; ++attempt) {
         std::vector<ManifestEntry> entries;
-        ASSERT_OK(manifest->ReadRowRangeEntries(written.first, ranges, nullptr, written.second,
-                                                &entries));
+        ASSERT_OK(manifest->ReadSelectedEntries(written.first, written.second, &ranges,
+                                                /*row_filter=*/nullptr, nullptr, &entries));
         ASSERT_TRUE(entries.empty());
         ASSERT_EQ(1, cache->SupplierCallCount(CacheKind::MANIFEST));
     }
@@ -362,8 +425,8 @@ TEST_F(RowRangeManifestFileTest, AllReadPathsShareOneDecodedEntry) {
                                                       written.second, &actual));
                 ASSERT_EQ(source, actual);
             } else {
-                ASSERT_OK(manifest->ReadRowRangeEntries(written.first, ranges, nullptr,
-                                                        written.second, &actual));
+                ASSERT_OK(manifest->ReadSelectedEntries(written.first, written.second, &ranges,
+                                                        /*row_filter=*/nullptr, nullptr, &actual));
                 ASSERT_EQ(std::vector<ManifestEntry>{b}, actual);
             }
         };
@@ -582,8 +645,8 @@ TEST_F(RowRangeManifestFileTest, BoundariesUnknownRangesAndDeleteMerging) {
             std::vector<ManifestEntry> ordinary;
             ASSERT_OK(manifest->Read(written.first, filter, written.second, &ordinary));
             std::vector<ManifestEntry> selected;
-            ASSERT_OK(manifest->ReadRowRangeEntries(written.first, ranges, filter, written.second,
-                                                    &selected));
+            ASSERT_OK(manifest->ReadSelectedEntries(written.first, written.second, &ranges,
+                                                    /*row_filter=*/nullptr, filter, &selected));
             ASSERT_EQ(ordinary, selected);
             std::vector<ManifestEntry> ordinary_live;
             std::vector<ManifestEntry> selected_live;
@@ -596,15 +659,15 @@ TEST_F(RowRangeManifestFileTest, BoundariesUnknownRangesAndDeleteMerging) {
         }
         ASSERT_OK_AND_ASSIGN(RowRangeIndex ranges, RowRangeIndex::Create({Range(100, 100)}));
         std::vector<ManifestEntry> entries;
-        ASSERT_NOK_WITH_MSG(manifest->ReadRowRangeEntries(
-                                written.first, ranges,
+        ASSERT_NOK_WITH_MSG(manifest->ReadSelectedEntries(
+                                written.first, written.second, &ranges, /*row_filter=*/nullptr,
                                 [](const ManifestEntry&) -> Result<bool> {
                                     return Status::IOError("filter error");
                                 },
-                                written.second, &entries),
+                                &entries),
                             "filter error");
-        ASSERT_NOK(manifest->ReadRowRangeEntries("missing-manifest", ranges, nullptr, std::nullopt,
-                                                 &entries));
+        ASSERT_NOK(manifest->ReadSelectedEntries("missing-manifest", std::nullopt, &ranges,
+                                                 /*row_filter=*/nullptr, nullptr, &entries));
     }
 }
 
@@ -626,8 +689,8 @@ TEST_F(RowRangeManifestFileTest, UncertainAndOverflowingRangesAreRetained) {
     ASSERT_OK_AND_ASSIGN(WrittenFile written, manifest->WriteWithoutRolling(source));
     ASSERT_OK_AND_ASSIGN(RowRangeIndex ranges, RowRangeIndex::Create({Range(0, 0)}));
     std::vector<ManifestEntry> actual;
-    ASSERT_OK(
-        manifest->ReadRowRangeEntries(written.first, ranges, nullptr, written.second, &actual));
+    ASSERT_OK(manifest->ReadSelectedEntries(written.first, written.second, &ranges,
+                                            /*row_filter=*/nullptr, nullptr, &actual));
     source.erase(source.begin() + 1);
     ASSERT_EQ(source, actual);
 }
@@ -684,11 +747,22 @@ TEST_F(RowRangeManifestFileTest, SchemaEvolutionAndVersionValidation) {
             std::vector<ManifestEntry> actual;
             if (mode >= 2) {
                 ASSERT_NOK_WITH_MSG(
-                    manifest->ReadRowRangeEntries(name, ranges, nullptr, std::nullopt, &actual),
+                    manifest->ReadSelectedEntries(name, std::nullopt, &ranges,
+                                                  /*row_filter=*/nullptr, nullptr, &actual),
                     (mode == 2 ? "not compatible" : "Unsupported version: 999"));
+                bool filter_called = false;
+                ASSERT_NOK_WITH_MSG(manifest->ReadSelectedEntries(
+                                        name, std::nullopt, /*row_ranges=*/nullptr,
+                                        [&filter_called](const InternalRow&) -> Result<bool> {
+                                            filter_called = true;
+                                            return false;
+                                        },
+                                        nullptr, &actual),
+                                    (mode == 2 ? "not compatible" : "Unsupported version: 999"));
+                ASSERT_FALSE(filter_called);
             } else {
-                ASSERT_OK(
-                    manifest->ReadRowRangeEntries(name, ranges, nullptr, std::nullopt, &actual));
+                ASSERT_OK(manifest->ReadSelectedEntries(name, std::nullopt, &ranges,
+                                                        /*row_filter=*/nullptr, nullptr, &actual));
                 ASSERT_EQ(mode == 0 ? 1 : 0, actual.size());
                 if (mode == 0) {
                     ASSERT_FALSE(actual[0].File()->first_row_id.has_value());
diff --git a/src/paimon/core/manifest/snapshot_live_manifest_entries.cpp b/src/paimon/core/manifest/snapshot_live_manifest_entries.cpp
index 8982209a6..c6d981e10 100644
--- a/src/paimon/core/manifest/snapshot_live_manifest_entries.cpp
+++ b/src/paimon/core/manifest/snapshot_live_manifest_entries.cpp
@@ -61,16 +61,6 @@ std::shared_ptr<Bytes> ToBytes(const MemorySegmentOutputStream& out,
 SnapshotLiveManifestEntries::SnapshotLiveManifestEntries(int32_t max_snapshots)
     : max_snapshots_(max_snapshots) {}
 
-std::optional<SnapshotLiveManifestEntries::Entry> SnapshotLiveManifestEntries::LatestBeforeOrEqual(
-    int64_t snapshot_id) const {
-    auto iter = entries_by_snapshot_.upper_bound(snapshot_id);
-    if (iter == entries_by_snapshot_.begin()) {
-        return std::optional<Entry>();
-    }
-    --iter;
-    return Entry{iter->first, iter->second};
-}
-
 void SnapshotLiveManifestEntries::Put(int64_t snapshot_id, std::vector<ManifestEntry>&& entries) {
     if (NormalizeMaxSnapshots(max_snapshots_) == 0) {
         return;
@@ -136,4 +126,67 @@ void SnapshotLiveManifestEntries::EvictIfNeeded() {
     }
 }
 
+Result<std::optional<SnapshotLiveManifestEntries::Selection>>
+SnapshotLiveManifestEntries::ReadSnapshot(
+    const MemorySegment& segment, int64_t snapshot_id, int32_t max_snapshots,
+    const std::function<Result<bool>(const InternalRow&)>& row_filter,
+    const std::shared_ptr<MemoryPool>& pool) {
+    std::optional<Selection> selected;
+    if (segment.Data() == nullptr || segment.Size() == 0 || max_snapshots <= 0) {
+        return selected;
+    }
+    auto bytes = segment.GetOrCreateHeapMemory(pool.get());
+    auto stream = std::make_shared<ByteArrayInputStream>(bytes->data(), bytes->size());
+    DataInputStream in(stream);
+    PAIMON_ASSIGN_OR_RAISE(int32_t magic, in.ReadValue<int32_t>());
+    if (magic != kMagic) {
+        return Status::Invalid("invalid snapshot live manifest entries magic");
+    }
+    PAIMON_ASSIGN_OR_RAISE(int32_t snapshot_count, in.ReadValue<int32_t>());
+    if (snapshot_count < 0) {
+        return Status::Invalid("snapshot live manifest entries snapshot count is negative");
+    }
+    ManifestEntrySerializer serializer(pool);
+    BinaryRowSerializer row_serializer(serializer.NumFields(), pool);
+    std::optional<int64_t> previous_id;
+    for (int32_t i = 0; i < snapshot_count; ++i) {
+        PAIMON_ASSIGN_OR_RAISE(int64_t id, in.ReadValue<int64_t>());
+        // Serialize emits snapshots in ascending order. Validate this before using the position
+        // to honor a retention limit smaller than the one used when the cache was populated.
+        if (previous_id && id <= previous_id.value()) {
+            return Status::Invalid("snapshot live manifest entries are not ordered");
+        }
+        previous_id = id;
+        const bool keep = id == snapshot_id && i >= snapshot_count - max_snapshots;
+        PAIMON_ASSIGN_OR_RAISE(int32_t entry_count, in.ReadValue<int32_t>());
+        if (entry_count < 0) {
+            return Status::Invalid("snapshot live manifest entries entry count is negative");
+        }
+        if (keep) {
+            selected.emplace();
+            selected->candidate_entries = entry_count;
+            if (!row_filter) {
+                selected->entries.reserve(entry_count);
+            }
+        }
+        for (int32_t j = 0; j < entry_count; ++j) {
+            if (!keep) {
+                PAIMON_RETURN_NOT_OK(BinaryRowSerializer::Skip(&in));
+                continue;
+            }
+            PAIMON_ASSIGN_OR_RAISE(BinaryRow row, row_serializer.Deserialize(&in));
+            if (row_filter) {
+                PAIMON_RETURN_NOT_OK(ManifestEntrySerializer::ValidateVersion(row.GetInt(0)));
+                PAIMON_ASSIGN_OR_RAISE(bool matches, row_filter(row));
+                if (!matches) {
+                    continue;
+                }
+            }
+            PAIMON_ASSIGN_OR_RAISE(ManifestEntry entry, serializer.FromRow(row));
+            selected->entries.emplace_back(std::move(entry));
+        }
+    }
+    return selected;
+}
+
 }  // namespace paimon
diff --git a/src/paimon/core/manifest/snapshot_live_manifest_entries.h b/src/paimon/core/manifest/snapshot_live_manifest_entries.h
index acf1d8e77..784d31b28 100644
--- a/src/paimon/core/manifest/snapshot_live_manifest_entries.h
+++ b/src/paimon/core/manifest/snapshot_live_manifest_entries.h
@@ -21,6 +21,7 @@
 
 #include <cstddef>
 #include <cstdint>
+#include <functional>
 #include <map>
 #include <memory>
 #include <optional>
@@ -31,6 +32,7 @@
 
 namespace paimon {
 class Bytes;
+class InternalRow;
 class MemoryPool;
 class MemorySegment;
 
@@ -40,14 +42,13 @@ class MemorySegment;
 /// cache; callers are responsible for storing the serialized bytes in the cache layer.
 class SnapshotLiveManifestEntries {
  public:
-    struct Entry {
-        int64_t snapshot_id;
-        std::shared_ptr<const std::vector<ManifestEntry>> entries;
+    struct Selection {
+        std::vector<ManifestEntry> entries;
+        uint64_t candidate_entries = 0;
     };
 
     explicit SnapshotLiveManifestEntries(int32_t max_snapshots);
 
-    std::optional<Entry> LatestBeforeOrEqual(int64_t snapshot_id) const;
     void Put(int64_t snapshot_id, std::vector<ManifestEntry>&& entries);
     size_t Size() const;
 
@@ -56,6 +57,16 @@ class SnapshotLiveManifestEntries {
                                                            int32_t max_snapshots,
                                                            const std::shared_ptr<MemoryPool>& pool);
 
+    /// Select an exact retained snapshot, applying a row filter before constructing file metadata.
+    /// The reader validates the entry version before invoking the filter, as in
+    /// ManifestFile::ReadSelectedEntries. The serialized cache value is unchanged. A Selection is
+    /// deliberately not a cache value: query-filtered entries must never replace the complete
+    /// bucket's cached entries.
+    static Result<std::optional<Selection>> ReadSnapshot(
+        const MemorySegment& segment, int64_t snapshot_id, int32_t max_snapshots,
+        const std::function<Result<bool>(const InternalRow&)>& row_filter,
+        const std::shared_ptr<MemoryPool>& pool);
+
  private:
     void EvictIfNeeded();
 
diff --git a/src/paimon/core/operation/append_only_file_store_scan_test.cpp b/src/paimon/core/operation/append_only_file_store_scan_test.cpp
index 1bb5bda01..29d19d7f3 100644
--- a/src/paimon/core/operation/append_only_file_store_scan_test.cpp
+++ b/src/paimon/core/operation/append_only_file_store_scan_test.cpp
@@ -29,9 +29,12 @@
 
 #include "arrow/type.h"
 #include "gtest/gtest.h"
+#include "paimon/cache/cache.h"
 #include "paimon/common/data/binary_row.h"
 #include "paimon/common/data/binary_row_writer.h"
+#include "paimon/common/data/serializer/binary_row_serializer.h"
 #include "paimon/common/io/cache/lru_cache.h"
+#include "paimon/common/io/memory_segment_output_stream.h"
 #include "paimon/common/utils/math.h"
 #include "paimon/core/bucket/default_bucket_function.h"
 #include "paimon/core/io/meta_to_arrow_array_converter.h"
@@ -50,6 +53,7 @@
 #include "paimon/executor.h"
 #include "paimon/format/file_format_factory.h"
 #include "paimon/fs/local/local_file_system.h"
+#include "paimon/memory/bytes.h"
 #include "paimon/memory/memory_pool.h"
 #include "paimon/metrics.h"
 #include "paimon/predicate/literal.h"
@@ -61,6 +65,7 @@
 #include "paimon/testing/mock/mock_file_batch_reader.h"
 #include "paimon/testing/mock/mock_format_reader_builder.h"
 #include "paimon/testing/utils/binary_row_generator.h"
+#include "paimon/testing/utils/counting_cache_test_utils.h"
 #include "paimon/testing/utils/test_helper.h"
 #include "paimon/testing/utils/testharness.h"
 #include "paimon/testing/utils/timezone_guard.h"
@@ -689,14 +694,61 @@ TEST(AppendOnlyFileStoreScanTest, TestScanDurationMetric) {
 
 namespace {
 
+class CountingPartitionPredicate : public PredicateFilter {
+ public:
+    explicit CountingPartitionPredicate(const std::shared_ptr<PredicateFilter>& predicate,
+                                        const Status& first_row_error = Status::OK())
+        : predicate_(predicate), first_row_error_(first_row_error) {}
+
+    bool operator==(const Predicate& other) const override {
+        return *predicate_ == other;
+    }
+    const Function& GetFunction() const override {
+        return predicate_->GetFunction();
+    }
+    std::shared_ptr<Predicate> Negate() const override {
+        return predicate_->Negate();
+    }
+    std::string ToString() const override {
+        return predicate_->ToString();
+    }
+    Result<std::vector<char>> Test(const arrow::Array& array,
+                                   arrow::MemoryPool* pool) const override {
+        return predicate_->Test(array, pool);
+    }
+    Result<bool> Test(const std::shared_ptr<arrow::Schema>& schema,
+                      const InternalRow& row) const override {
+        if (++row_calls_ == 1 && !first_row_error_.ok()) {
+            return first_row_error_;
+        }
+        return predicate_->Test(schema, row);
+    }
+    Result<bool> Test(const std::shared_ptr<arrow::Schema>& schema, int64_t row_count,
+                      const InternalRow& min_values, const InternalRow& max_values,
+                      const InternalArray& null_counts) const override {
+        ++stats_calls_;
+        // Retain every manifest so even an empty result exercises entry filtering.
+        return true;
+    }
+
+    mutable std::atomic<int32_t> row_calls_{0};
+    mutable std::atomic<int32_t> stats_calls_{0};
+
+ private:
+    std::shared_ptr<PredicateFilter> predicate_;
+    Status first_row_error_;
+};
+
 std::shared_ptr<FileStoreScan> BuildScan(const std::string& table_path,
                                          const std::shared_ptr<Cache>& cache,
                                          const std::optional<int32_t>& bucket = std::nullopt,
                                          const std::shared_ptr<Predicate>& predicate = nullptr,
-                                         bool manifest_entry_lazy_decode_enabled = true) {
+                                         bool manifest_entry_lazy_decode_enabled = true,
+                                         int32_t max_snapshots = 8) {
     ScanContextBuilder context_builder(table_path);
     context_builder.AddOption(Options::FILE_FORMAT, "orc")
-        .AddOption(Options::SCAN_MANIFEST_ENTRY_CACHE_MAX_SNAPSHOTS, "8")
+        .AddOption(Options::SCAN_MANIFEST_ENTRY_CACHE_MAX_SNAPSHOTS,
+                   fmt::format("{}", max_snapshots))
         .AddOption(Options::SCAN_MANIFEST_ENTRY_LAZY_DECODE_ENABLED,
                    manifest_entry_lazy_decode_enabled ? "true" : "false")
         .WithCache(cache);
@@ -811,7 +863,8 @@ TEST(AppendOnlyFileStoreScanTest, TestSnapshotLiveManifestCachePath) {
 TEST(AppendOnlyFileStoreScanTest, TestSnapshotLiveManifestCacheRebuildOnMiss) {
     TimezoneGuard guard("Asia/Shanghai");
     std::string table_path = paimon::test::GetDataDir() + "/orc/append_09.db/append_09/";
-    auto cache = std::make_shared<LruCache>(/*max_weight=*/16 * 1024 * 1024);
+    auto cache =
+        std::make_shared<CountingRoutingCache>(CacheKind::SNAPSHOT_LIVE_MANIFEST, 16 * 1024 * 1024);
 
     // Seed the cache with an earlier snapshot.
     auto scan_base = BuildScan(table_path, cache, /*bucket=*/0);
@@ -819,7 +872,7 @@ TEST(AppendOnlyFileStoreScanTest, TestSnapshotLiveManifestCacheRebuildOnMiss) {
                          scan_base->GetSnapshotManager()->LoadSnapshot(/*snapshot_id=*/3));
     scan_base->WithSnapshot(snapshot_3);
     ASSERT_OK_AND_ASSIGN(auto plan_base, scan_base->CreatePlan());
-    (void)plan_base;
+    ASSERT_EQ(cache->GetCount(CacheKind::SNAPSHOT_LIVE_MANIFEST), 1);
 
     // Advance to a newer snapshot: cache miss rebuilds the target snapshot bucket.
     auto scan_next = BuildScan(table_path, cache, /*bucket=*/0);
@@ -827,11 +880,30 @@ TEST(AppendOnlyFileStoreScanTest, TestSnapshotLiveManifestCacheRebuildOnMiss) {
                          scan_next->GetSnapshotManager()->LoadSnapshot(/*snapshot_id=*/5));
     scan_next->WithSnapshot(snapshot_5);
     ASSERT_OK_AND_ASSIGN(auto plan_next, scan_next->CreatePlan());
+    ASSERT_EQ(cache->GetCount(CacheKind::SNAPSHOT_LIVE_MANIFEST), 2);
 
     auto scan_expected = BuildScan(table_path, /*cache=*/nullptr, /*bucket=*/0);
     scan_expected->WithSnapshot(snapshot_5);
     ASSERT_OK_AND_ASSIGN(auto plan_expected, scan_expected->CreatePlan());
     ASSERT_EQ(SortedFileNames(plan_expected->Files()), SortedFileNames(plan_next->Files()));
+
+    auto scan_retained = BuildScan(table_path, cache, /*bucket=*/0);
+    scan_retained->WithSnapshot(snapshot_3);
+    ASSERT_OK_AND_ASSIGN(auto plan_retained, scan_retained->CreatePlan());
+    ASSERT_EQ(SortedFileNames(plan_base->Files()), SortedFileNames(plan_retained->Files()));
+    ASSERT_EQ(cache->GetCount(CacheKind::SNAPSHOT_LIVE_MANIFEST), 3);
+    ASSERT_OK_AND_ASSIGN(uint64_t hit, scan_retained->GetScanMetrics()->GetCounter(
+                                           ScanMetrics::LAST_SNAPSHOT_CACHE_HIT));
+    ASSERT_EQ(hit, 1);
+    auto scan_evicted = BuildScan(table_path, cache, /*bucket=*/0, /*predicate=*/nullptr,
+                                  /*manifest_entry_lazy_decode_enabled=*/true, /*max_snapshots=*/1);
+    scan_evicted->WithSnapshot(snapshot_3);
+    ASSERT_OK_AND_ASSIGN(auto plan_evicted, scan_evicted->CreatePlan());
+    ASSERT_EQ(SortedFileNames(plan_base->Files()), SortedFileNames(plan_evicted->Files()));
+    ASSERT_EQ(cache->GetCount(CacheKind::SNAPSHOT_LIVE_MANIFEST), 4);
+    ASSERT_OK_AND_ASSIGN(
+        hit, scan_evicted->GetScanMetrics()->GetCounter(ScanMetrics::LAST_SNAPSHOT_CACHE_HIT));
+    ASSERT_EQ(hit, 0);
 }
 
 TEST(AppendOnlyFileStoreScanTest, TestSnapshotLiveManifestCacheFallbackWithoutLazyDecode) {
@@ -851,4 +923,254 @@ TEST(AppendOnlyFileStoreScanTest, TestSnapshotLiveManifestCacheFallbackWithoutLa
     ASSERT_OK_AND_ASSIGN(auto plan_expected, scan_expected->CreatePlan());
     ASSERT_EQ(SortedFileNames(plan_expected->Files()), SortedFileNames(plan_fallback->Files()));
 }
+
+TEST(AppendOnlyFileStoreScanTest, SnapshotCachePropagatesPartitionPredicateError) {
+    TimezoneGuard guard("Asia/Shanghai");
+    const std::string table_path = GetDataDir() + "/orc/append_09.db/append_09/";
+    auto cache = std::make_shared<LruCache>(16 * 1024 * 1024);
+    auto warm = BuildScan(table_path, cache, /*bucket=*/0);
+    ASSERT_OK_AND_ASSIGN(Snapshot snapshot, warm->GetSnapshotManager()->LoadSnapshot(5));
+    warm->WithSnapshot(snapshot);
+    ASSERT_OK_AND_ASSIGN(auto warm_plan, warm->CreatePlan());
+    auto files = warm_plan->Files();
+    ASSERT_FALSE(files.empty());
+    const int32_t partition = files.front().Partition().GetInt(0);
+    auto predicate = PredicateBuilder::Equal(1, "f1", FieldType::INT, Literal(partition));
+    for (bool lazy_decode : {false, true}) {
+        SCOPED_TRACE(lazy_decode);
+        for (const Status& error : {Status::IOError("injected partition predicate error"),
+                                    Status::Invalid("injected partition predicate error")}) {
+            SCOPED_TRACE(error.ToString());
+            auto scan = BuildScan(table_path, cache, /*bucket=*/0, predicate, lazy_decode);
+            scan->WithSnapshot(snapshot);
+            auto failing =
+                std::make_shared<CountingPartitionPredicate>(scan->GetPartitionPredicate(), error);
+            scan->partition_filter_ = failing;
+            // A transient predicate failure must reach the caller, without an internal retry.
+            auto result = scan->CreatePlan();
+            ASSERT_FALSE(result.ok());
+            ASSERT_EQ(result.status().code(), error.code());
+            ASSERT_NE(result.status().message().find(error.message()), std::string::npos);
+            ASSERT_EQ(failing->row_calls_.load(), 1);
+            // The failure must leave the complete cached snapshot available for the next request.
+            ASSERT_OK_AND_ASSIGN(auto plan, scan->CreatePlan());
+            ASSERT_FALSE(plan->Files().empty());
+            for (const auto& entry : plan->Files()) {
+                ASSERT_EQ(entry.Partition().GetInt(0), partition);
+            }
+            ASSERT_OK_AND_ASSIGN(uint64_t hit, scan->GetScanMetrics()->GetCounter(
+                                                   ScanMetrics::LAST_SNAPSHOT_CACHE_HIT));
+            ASSERT_EQ(hit, 1);
+        }
+    }
+}
+
+TEST(AppendOnlyFileStoreScanTest, SnapshotCacheRebuildsCorruptValues) {
+    TimezoneGuard guard("Asia/Shanghai");
+    const std::string table_path = GetDataDir() + "/orc/append_09.db/append_09/";
+    auto source = BuildScan(table_path, /*cache=*/nullptr, /*bucket=*/0);
+    ASSERT_OK_AND_ASSIGN(Snapshot snapshot, source->GetSnapshotManager()->LoadSnapshot(5));
+    source->WithSnapshot(snapshot);
+    ASSERT_OK_AND_ASSIGN(auto source_plan, source->CreatePlan());
+    auto files = source_plan->Files();
+    ASSERT_FALSE(files.empty());
+    const ManifestEntry first = files.front();
+    const auto all_names = SortedFileNames(std::move(files));
+    auto predicate =
+        PredicateBuilder::Equal(1, "f1", FieldType::INT, Literal(first.Partition().GetInt(0)));
+    auto expected = BuildScan(table_path, /*cache=*/nullptr, /*bucket=*/0, predicate);
+    expected->WithSnapshot(snapshot);
+    ASSERT_OK_AND_ASSIGN(auto expected_plan, expected->CreatePlan());
+    const auto expected_names = SortedFileNames(expected_plan->Files());
+    auto pool = GetDefaultPool();
+    ManifestEntrySerializer serializer(pool);
+    BinaryRowSerializer row_serializer(serializer.NumFields(), pool);
+    for (const std::string corruption : {"magic", "version", "truncated", "partition"}) {
+        SCOPED_TRACE(corruption);
+        ASSERT_OK_AND_ASSIGN(BinaryRow row, serializer.ToRow(first));
+        if (corruption == "version") {
+            row.SetInt(0, 999);
+        } else if (corruption == "partition") {
+            // A negative partition arity fails before predicate evaluation, regardless of mode.
+            auto segment = row.GetSegment();
+            const int32_t offset =
+                static_cast<int32_t>(row.GetStringView(2).data() - segment.Data());
+            segment.PutValue<int32_t>(offset, -1);
+        }
+        MemorySegmentOutputStream out(MemorySegmentOutputStream::DEFAULT_SEGMENT_SIZE, pool);
+        out.WriteValue<int32_t>(corruption == "magic" ? 0 : 0x534d4543);
+        out.WriteValue<int32_t>(1);
+        out.WriteValue<int64_t>(snapshot.Id());
+        out.WriteValue<int32_t>(1);
+        ASSERT_OK(row_serializer.Serialize(row, &out));
+        ASSERT_EQ(out.Segments().size(), 1);
+        const int64_t size = out.CurrentSize() - (corruption == "truncated" ? 1 : 0);
+        std::shared_ptr<Bytes> bytes =
+            Bytes::AllocateBytes(std::string(out.Segments().front().Data(), size), pool.get());
+        const auto segment = MemorySegment::Wrap(bytes);
+        ASSERT_NOK(SnapshotLiveManifestEntries::ReadSnapshot(segment, snapshot.Id(), 8,
+                                                             /*row_filter=*/nullptr, pool));
+        for (bool lazy_decode : {false, true}) {
+            SCOPED_TRACE(lazy_decode);
+            auto cache = std::make_shared<CountingRoutingCache>(CacheKind::SNAPSHOT_LIVE_MANIFEST,
+                                                                16 * 1024 * 1024);
+            auto scan = BuildScan(table_path, cache, /*bucket=*/0, predicate, lazy_decode);
+            scan->WithSnapshot(snapshot);
+            ASSERT_OK(cache->Put(scan->CreateSnapshotLiveManifestEntriesCacheKey(0),
+                                 std::make_shared<CacheValue>(segment, CacheCallback())));
+            for (int32_t attempt = 0; attempt < 2; ++attempt) {
+                ASSERT_OK_AND_ASSIGN(auto plan, scan->CreatePlan());
+                ASSERT_EQ(SortedFileNames(plan->Files()), expected_names);
+                ASSERT_OK_AND_ASSIGN(uint64_t hit, scan->GetScanMetrics()->GetCounter(
+                                                       ScanMetrics::LAST_SNAPSHOT_CACHE_HIT));
+                ASSERT_EQ(hit, attempt);
+                ASSERT_EQ(cache->GetCount(CacheKind::SNAPSHOT_LIVE_MANIFEST), attempt + 1);
+            }
+            auto unfiltered = BuildScan(table_path, cache, /*bucket=*/0, nullptr, lazy_decode);
+            unfiltered->WithSnapshot(snapshot);
+            ASSERT_OK_AND_ASSIGN(auto all, unfiltered->CreatePlan());
+            ASSERT_EQ(SortedFileNames(all->Files()), all_names);
+        }
+    }
+}
+
+TEST(AppendOnlyFileStoreScanTest, PartitionPruningAcrossCacheModes) {
+    auto dir = UniqueTestDirectory::Create();
+    ASSERT_TRUE(dir);
+    auto fields =
+        arrow::FieldVector{arrow::field("f0", arrow::int32()), arrow::field("f1", arrow::int32())};
+    std::map<std::string, std::string> options = {
+        {Options::BUCKET, "1"}, {Options::BUCKET_KEY, "f0"}, {Options::FILE_FORMAT, "orc"}};
+    ASSERT_OK_AND_ASSIGN(auto helper,
+                         TestHelper::Create(dir->Str(), arrow::schema(fields), {"f1"}, {}, options,
+                                            /*is_streaming_mode=*/false));
+    std::vector<std::unique_ptr<RecordBatch>> batches;
+    for (int32_t partition : {10, 20}) {
+        ASSERT_OK_AND_ASSIGN(
+            auto batch,
+            TestHelper::MakeRecordBatch(arrow::struct_(fields), fmt::format("[[1, {}]]", partition),
+                                        {{"f1", fmt::format("{}", partition)}}, /*bucket=*/0,
+                                        {RecordBatch::RowKind::INSERT}));
+        batches.push_back(std::move(batch));
+    }
+    ASSERT_OK(helper->WriteAndCommit(std::move(batches), 0, std::nullopt));
+    const std::string table_path = dir->Str() + "/foo.db/bar";
+    for (bool lazy_decode : {false, true}) {
+        SCOPED_TRACE(lazy_decode);
+        // Exercise uncached reads, manifest Arrow caching, and bucket snapshot caching.
+        for (int32_t mode : {0, 1, 2}) {
+            SCOPED_TRACE(mode);
+            auto cache = std::make_shared<CountingRoutingCache>(std::map<CacheKind, int64_t>{
+                {CacheKind::MANIFEST, 16 * 1024 * 1024},
+                {CacheKind::SNAPSHOT_LIVE_MANIFEST, 16 * 1024 * 1024}});
+            std::shared_ptr<Cache> scan_cache = mode == 0 ? nullptr : cache;
+            std::optional<int32_t> bucket = mode == 2 ? std::optional<int32_t>(0) : std::nullopt;
+            int32_t scans = 0;
+            int64_t manifest_loads = 0;
+            // A filtered cold read must retain other partitions for subsequent cached reads.
+            for (int32_t partition : {10, 20, 99, 10}) {
+                SCOPED_TRACE(partition);
+                auto predicate =
+                    PredicateBuilder::Equal(1, "f1", FieldType::INT, Literal(partition));
+                auto scan = BuildScan(table_path, scan_cache, bucket, predicate, lazy_decode);
+                auto counting =
+                    std::make_shared<CountingPartitionPredicate>(scan->GetPartitionPredicate());
+                scan->partition_filter_ = counting;
+                ASSERT_OK_AND_ASSIGN(auto plan, scan->CreatePlan());
+                ASSERT_EQ(counting->row_calls_.load(), 2);
+                ASSERT_GT(counting->stats_calls_.load(), 0);
+                ASSERT_EQ(plan->Files().size(), partition == 99 ? 0 : 1);
+                if (partition != 99) {
+                    ASSERT_EQ(plan->Files()[0].Partition().GetInt(0), partition);
+                }
+                auto metrics = scan->GetScanMetrics();
+                ASSERT_OK_AND_ASSIGN(uint64_t hit,
+                                     metrics->GetCounter(ScanMetrics::LAST_SNAPSHOT_CACHE_HIT));
+                ASSERT_EQ(hit, mode == 2 && scans > 0 ? 1 : 0);
+                if (mode == 2) {
+                    ASSERT_EQ(cache->GetCount(CacheKind::SNAPSHOT_LIVE_MANIFEST), scans + 1);
+                    ASSERT_OK_AND_ASSIGN(
+                        uint64_t scanned,
+                        metrics->GetCounter(ScanMetrics::LAST_LAZY_DECODE_SCANNED_ROWS));
+                    ASSERT_EQ(scanned, 2);
+                }
+                if (mode == 1) {
+                    if (scans == 0) {
+                        manifest_loads = cache->SupplierCallCount(CacheKind::MANIFEST);
+                        ASSERT_GT(manifest_loads, 0);
+                    }
+                    ASSERT_EQ(cache->SupplierCallCount(CacheKind::MANIFEST), manifest_loads);
+                    ASSERT_EQ(cache->GetCount(CacheKind::MANIFEST), (scans + 1) * manifest_loads);
+                }
+                ++scans;
+            }
+            auto unfiltered = BuildScan(table_path, scan_cache, bucket, nullptr, lazy_decode);
+            ASSERT_OK_AND_ASSIGN(auto all, unfiltered->CreatePlan());
+            ASSERT_EQ(all->Files().size(), 2);
+        }
+    }
+}
+
+TEST(AppendOnlyFileStoreScanTest, PartitionPruningSkipsExcludedMetadataInCreatePlan) {
+    auto dir = UniqueTestDirectory::Create();
+    ASSERT_TRUE(dir);
+    auto fields =
+        arrow::FieldVector{arrow::field("f0", arrow::int32()), arrow::field("f1", arrow::int32())};
+    ASSERT_OK_AND_ASSIGN(
+        auto helper,
+        TestHelper::Create(
+            dir->Str(), arrow::schema(fields), {"f1"}, {},
+            {{Options::BUCKET, "1"}, {Options::BUCKET_KEY, "f0"}, {Options::FILE_FORMAT, "orc"}},
+            /*is_streaming_mode=*/false));
+    const std::string table_path = dir->Str() + "/foo.db/bar";
+    auto pool = GetDefaultPool();
+    ASSERT_OK_AND_ASSIGN(
+        auto file,
+        DataFileMeta::ForAppend("file.orc", 10, 1, SimpleStats::EmptyStats(), 0, 0, 0, std::nullopt,
+                                std::nullopt, std::nullopt, std::nullopt, std::nullopt));
+    ManifestEntry kept(FileKind::Add(), BinaryRowGenerator::GenerateRow({10}, pool.get()), 0, 1,
+                       file);
+    ManifestEntry excluded(FileKind(static_cast<int8_t>(2)),
+                           BinaryRowGenerator::GenerateRow({20}, pool.get()), 0, 1, file);
+    auto writer = BuildScan(table_path, nullptr);
+    using WrittenFile = std::pair<std::string, int64_t>;
+    ASSERT_OK_AND_ASSIGN(WrittenFile written,
+                         writer->manifest_file_->WriteWithoutRolling({kept, excluded}));
+    // Both partitions share one manifest, whose statistics cannot prune the requested partition.
+    ManifestFileMeta meta(written.first, written.second, 2, 0,
+                          BinaryRowGenerator::GenerateStats({10}, {20}, {0}, pool.get()), 0, 0, 0,
+                          0, 0, std::nullopt, std::nullopt);
+    ASSERT_OK_AND_ASSIGN(WrittenFile base, writer->manifest_list_->Write({}));
+    ASSERT_OK_AND_ASSIGN(WrittenFile delta, writer->manifest_list_->Write({meta}));
+    Snapshot snapshot(1, 0, base.first, base.second, delta.first, delta.second, std::nullopt,
+                      std::nullopt, std::nullopt, "test", 0, Snapshot::CommitKind::Append(), 0, 2,
+                      2, std::nullopt, std::nullopt, std::nullopt, std::nullopt, std::nullopt);
+    auto predicate = PredicateBuilder::Equal(1, "f1", FieldType::INT, Literal(10));
+    for (int32_t mode : {0, 1, 2}) {
+        SCOPED_TRACE(mode);
+        std::shared_ptr<Cache> cache =
+            mode == 0 ? nullptr : std::make_shared<LruCache>(16 * 1024 * 1024);
+        std::optional<int32_t> bucket = mode == 2 ? std::optional<int32_t>(0) : std::nullopt;
+        auto scan = BuildScan(table_path, cache, bucket, predicate);
+        scan->WithSnapshot(snapshot);
+        if (mode == 2) {
+            // Seed the complete snapshot directly: rebuilding must decode every partition.
+            SnapshotLiveManifestEntries entries(8);
+            entries.Put(1, {kept, excluded});
+            ASSERT_OK(scan->StoreSnapshotLiveManifestEntries(0, entries));
+        }
+        for (int32_t attempt = 0; attempt < 2; ++attempt) {
+            ASSERT_OK_AND_ASSIGN(auto plan, scan->CreatePlan());
+            ASSERT_EQ(plan->Files(), std::vector<ManifestEntry>{kept});
+            ASSERT_OK_AND_ASSIGN(uint64_t hit, scan->GetScanMetrics()->GetCounter(
+                                                   ScanMetrics::LAST_SNAPSHOT_CACHE_HIT));
+            ASSERT_EQ(hit, mode == 2 ? 1 : 0);
+        }
+        auto full_decode = BuildScan(table_path, cache, bucket, predicate,
+                                     /*manifest_entry_lazy_decode_enabled=*/false);
+        full_decode->WithSnapshot(snapshot);
+        ASSERT_NOK_WITH_MSG(full_decode->CreatePlan(), "Unsupported byte value 2 for file kind.");
+    }
+}
+
 }  // namespace paimon::test
diff --git a/src/paimon/core/operation/file_store_scan.cpp b/src/paimon/core/operation/file_store_scan.cpp
index e7475d178..e5f132335 100644
--- a/src/paimon/core/operation/file_store_scan.cpp
+++ b/src/paimon/core/operation/file_store_scan.cpp
@@ -42,6 +42,7 @@
 #include "paimon/core/io/file_index_options.h"
 #include "paimon/core/manifest/file_entry.h"
 #include "paimon/core/manifest/file_kind.h"
+#include "paimon/core/manifest/manifest_entry_serializer.h"
 #include "paimon/core/manifest/manifest_file.h"
 #include "paimon/core/manifest/manifest_file_meta.h"
 #include "paimon/core/manifest/manifest_list.h"
@@ -196,19 +197,10 @@ Result<std::shared_ptr<FileStoreScan::RawPlan>> FileStoreScan::CreatePlan() cons
     uint64_t lazy_decode_scanned_rows = 0;
     bool snapshot_cache_hit = false;
     if (use_snapshot_live_manifest_cache) {
-        PAIMON_RETURN_NOT_OK(ReadManifestEntriesWithCache(snapshot.value(), all_manifest_file_metas,
-                                                          cache_bucket.value(), &manifest_entries,
-                                                          &snapshot_cache_hit));
-        lazy_decode_scanned_rows = manifest_entries.size();
-        std::vector<ManifestEntry> filtered_entries;
-        filtered_entries.reserve(manifest_entries.size());
-        for (auto& entry : manifest_entries) {
-            PAIMON_ASSIGN_OR_RAISE(bool keep, FilterManifestEntry(entry));
-            if (keep) {
-                filtered_entries.emplace_back(std::move(entry));
-            }
-        }
-        manifest_entries = std::move(filtered_entries);
+        PAIMON_ASSIGN_OR_RAISE(lazy_decode_scanned_rows,
+                               ReadManifestEntriesWithCache(
+                                   snapshot.value(), all_manifest_file_metas, cache_bucket.value(),
+                                   &manifest_entries, &snapshot_cache_hit));
     } else {
         lazy_decode_scanned_rows = std::accumulate(
             filtered_manifest_file_metas.begin(), filtered_manifest_file_metas.end(), uint64_t{0},
@@ -396,44 +388,81 @@ Status FileStoreScan::ReadManifestEntries(const std::vector<ManifestFileMeta>& m
 // bucket counts, with the current count and schema in the cache key. Exact hits
 // can be returned directly; cache misses rebuild the target snapshot bucket from the target
 // snapshot's data manifests.
-Status FileStoreScan::ReadManifestEntriesWithCache(
+Result<uint64_t> FileStoreScan::ReadManifestEntriesWithCache(
     const Snapshot& snapshot, const std::vector<ManifestFileMeta>& all_manifest_metas,
     int32_t bucket, std::vector<ManifestEntry>* manifest_entries, bool* cache_hit) const {
     Duration cache_load_duration;
-    PAIMON_ASSIGN_OR_RAISE(SnapshotLiveManifestEntries cached_entries,
-                           LoadSnapshotLiveManifestEntries(bucket));
+    const int32_t max_snapshots = core_options_.GetScanManifestEntryCacheMaxSnapshots();
+    SnapshotLiveManifestEntries cached_entries(max_snapshots);
+    bool predicate_failed = false;
+    auto row_filter = CreateManifestEntryRowFilter(&predicate_failed);
+    bool partition_filtered = false;
+    uint64_t candidate_entries = 0;
+    *cache_hit = false;
+    Result<std::shared_ptr<CacheValue>> value = core_options_.GetCache()->Get(
+        CreateSnapshotLiveManifestEntriesCacheKey(bucket),
+        [](const std::shared_ptr<CacheKey>&) -> Result<std::shared_ptr<CacheValue>> {
+            return std::shared_ptr<CacheValue>();
+        });
+    if (value.ok() && value.value()) {
+        Result<std::optional<SnapshotLiveManifestEntries::Selection>> selected =
+            SnapshotLiveManifestEntries::ReadSnapshot(value.value()->GetSegment(), snapshot.Id(),
+                                                      max_snapshots, row_filter, pool_);
+        if (predicate_failed) {
+            return selected.status();
+        }
+        if (selected.ok() && selected.value()) {
+            *cache_hit = true;
+            partition_filtered = static_cast<bool>(row_filter);
+            candidate_entries = selected.value()->candidate_entries;
+            *manifest_entries = std::move(selected.value()->entries);
+        } else if (selected.ok()) {
+            // Reuse the same cache value to retain complete historical snapshots on a miss.
+            // Query-filtered entries must never be written back into the cache.
+            Result<SnapshotLiveManifestEntries> deserialized =
+                SnapshotLiveManifestEntries::Deserialize(value.value()->GetSegment(), max_snapshots,
+                                                         pool_);
+            if (deserialized.ok()) {
+                cached_entries = std::move(deserialized.value());
+            }
+        }
+    }
     const uint64_t cache_load_duration_ms = cache_load_duration.Get();
     metrics_->SetCounter(ScanMetrics::LAST_SNAPSHOT_CACHE_LOAD_DURATION, cache_load_duration_ms);
     metrics_->ObserveHistogram(ScanMetrics::SNAPSHOT_CACHE_LOAD_DURATION,
                                static_cast<double>(cache_load_duration_ms));
-    std::optional<SnapshotLiveManifestEntries::Entry> cached =
-        cached_entries.LatestBeforeOrEqual(snapshot.Id());
-    if (cached && cached->snapshot_id == snapshot.Id()) {
-        *cache_hit = true;
-        *manifest_entries = *cached->entries;
-        return Status::OK();
+    if (!*cache_hit) {
+        std::vector<ManifestFileMeta> bucket_manifest_metas;
+        for (const auto& meta : all_manifest_metas) {
+            if ((!bucket_filter_ && bucket_selector_) || MayContainBucket(meta, bucket)) {
+                bucket_manifest_metas.push_back(meta);
+            }
+        }
+        PAIMON_RETURN_NOT_OK(
+            ReadAndMergeBucketFileEntries(bucket_manifest_metas, bucket, manifest_entries));
+        candidate_entries = manifest_entries->size();
+        std::vector<ManifestEntry> cache_entries = *manifest_entries;
+        cached_entries.Put(snapshot.Id(), std::move(cache_entries));
+        Duration cache_store_duration;
+        PAIMON_RETURN_NOT_OK(StoreSnapshotLiveManifestEntries(bucket, cached_entries));
+        const uint64_t cache_store_duration_ms = cache_store_duration.Get();
+        metrics_->SetCounter(ScanMetrics::LAST_SNAPSHOT_CACHE_STORE_DURATION,
+                             cache_store_duration_ms);
+        metrics_->ObserveHistogram(ScanMetrics::SNAPSHOT_CACHE_STORE_DURATION,
+                                   static_cast<double>(cache_store_duration_ms));
     }
-    *cache_hit = false;
-
-    // Rebuild the target snapshot bucket from all manifests and write the live entries back to the
-    // cache.
-    std::vector<ManifestFileMeta> bucket_manifest_metas;
-    for (const auto& meta : all_manifest_metas) {
-        if ((!bucket_filter_ && bucket_selector_) || MayContainBucket(meta, bucket)) {
-            bucket_manifest_metas.push_back(meta);
+    std::vector<ManifestEntry> filtered_entries;
+    filtered_entries.reserve(manifest_entries->size());
+    for (auto& entry : *manifest_entries) {
+        PAIMON_ASSIGN_OR_RAISE(bool keep, partition_filtered
+                                              ? FilterNonPartitionManifestEntry(entry)
+                                              : FilterManifestEntry(entry));
+        if (keep) {
+            filtered_entries.emplace_back(std::move(entry));
         }
     }
-    PAIMON_RETURN_NOT_OK(
-        ReadAndMergeBucketFileEntries(bucket_manifest_metas, bucket, manifest_entries));
-    std::vector<ManifestEntry> cache_entries = *manifest_entries;
-    cached_entries.Put(snapshot.Id(), std::move(cache_entries));
-    Duration cache_store_duration;
-    PAIMON_RETURN_NOT_OK(StoreSnapshotLiveManifestEntries(bucket, cached_entries));
-    const uint64_t cache_store_duration_ms = cache_store_duration.Get();
-    metrics_->SetCounter(ScanMetrics::LAST_SNAPSHOT_CACHE_STORE_DURATION, cache_store_duration_ms);
-    metrics_->ObserveHistogram(ScanMetrics::SNAPSHOT_CACHE_STORE_DURATION,
-                               static_cast<double>(cache_store_duration_ms));
-    return Status::OK();
+    *manifest_entries = std::move(filtered_entries);
+    return candidate_entries;
 }
 
 std::shared_ptr<CacheKey> FileStoreScan::CreateSnapshotLiveManifestEntriesCacheKey(
@@ -447,26 +476,6 @@ std::shared_ptr<CacheKey> FileStoreScan::CreateSnapshotLiveManifestEntriesCacheK
         table_path_, BranchManager::NormalizeBranch(core_options_.GetBranch()), bucket);
 }
 
-Result<SnapshotLiveManifestEntries> FileStoreScan::LoadSnapshotLiveManifestEntries(
-    int32_t bucket) const {
-    auto supplier = [](const std::shared_ptr<CacheKey>&) -> Result<std::shared_ptr<CacheValue>> {
-        return std::shared_ptr<CacheValue>();
-    };
-    std::shared_ptr<CacheKey> cache_key = CreateSnapshotLiveManifestEntriesCacheKey(bucket);
-    const auto max_snapshots = core_options_.GetScanManifestEntryCacheMaxSnapshots();
-    Result<std::shared_ptr<CacheValue>> cache_result =
-        core_options_.GetCache()->Get(cache_key, supplier);
-    if (!cache_result.ok() || !cache_result.value()) {
-        return SnapshotLiveManifestEntries(max_snapshots);
-    }
-    Result<SnapshotLiveManifestEntries> deserialized = SnapshotLiveManifestEntries::Deserialize(
-        cache_result.value()->GetSegment(), max_snapshots, pool_);
-    if (!deserialized.ok()) {
-        return SnapshotLiveManifestEntries(max_snapshots);
-    }
-    return std::move(deserialized.value());
-}
-
 Status FileStoreScan::StoreSnapshotLiveManifestEntries(
     int32_t bucket, const SnapshotLiveManifestEntries& entries) const {
     Result<std::shared_ptr<Bytes>> bytes_result = entries.Serialize(pool_);
@@ -619,25 +628,35 @@ bool FileStoreScan::FilterManifestByRowRanges(const ManifestFileMeta& manifest)
 
 Status FileStoreScan::ReadManifestFileMeta(const ManifestFileMeta& manifest,
                                            std::vector<ManifestEntry>* entries) const {
+    const RowRangeIndex* row_ranges = nullptr;
     if (row_range_index_ && core_options_.DataEvolutionEnabled() &&
         core_options_.ScanManifestEntryLazyDecodeEnabled()) {
-        return manifest_file_->ReadRowRangeEntries(
-            manifest.FileName(), row_range_index_.value(),
-            [this](const ManifestEntry& entry) -> Result<bool> {
-                return FilterManifestEntry(entry);
-            },
-            manifest.FileSize(), entries);
-    }
-    std::vector<ManifestEntry> unfiltered_entries;
-    PAIMON_RETURN_NOT_OK(manifest_file_->Read(
-        manifest.FileName(),
-        [this](const ManifestEntry& entry) -> Result<bool> { return FilterManifestEntry(entry); },
-        manifest.FileSize(), &unfiltered_entries));
-    entries->reserve(entries->size() + unfiltered_entries.size());
-    for (auto& entry : unfiltered_entries) {
-        entries->emplace_back(std::move(entry));
+        row_ranges = &row_range_index_.value();
+    }
+    auto row_filter = CreateManifestEntryRowFilter(/*predicate_failed=*/nullptr);
+    return manifest_file_->ReadSelectedEntries(
+        manifest.FileName(), manifest.FileSize(), row_ranges, row_filter,
+        [this, partition_filtered =
+                   static_cast<bool>(row_filter)](const ManifestEntry& entry) -> Result<bool> {
+            return partition_filtered ? FilterNonPartitionManifestEntry(entry)
+                                      : FilterManifestEntry(entry);
+        },
+        entries);
+}
+
+std::function<Result<bool>(const InternalRow&)> FileStoreScan::CreateManifestEntryRowFilter(
+    bool* predicate_failed) const {
+    if (!partition_filter_ || !core_options_.ScanManifestEntryLazyDecodeEnabled()) {
+        return nullptr;
     }
-    return Status::OK();
+    return [this, predicate_failed](const InternalRow& row) -> Result<bool> {
+        PAIMON_ASSIGN_OR_RAISE(BinaryRow partition, ManifestEntrySerializer::GetPartitionView(row));
+        Result<bool> result = partition_filter_->Test(partition_schema_, partition);
+        if (!result.ok() && predicate_failed) {
+            *predicate_failed = true;
+        }
+        return result;
+    };
 }
 
 Result<bool> FileStoreScan::FilterManifestEntry(const ManifestEntry& entry) const {
@@ -648,6 +667,10 @@ Result<bool> FileStoreScan::FilterManifestEntry(const ManifestEntry& entry) cons
             return false;
         }
     }
+    return FilterNonPartitionManifestEntry(entry);
+}
+
+Result<bool> FileStoreScan::FilterNonPartitionManifestEntry(const ManifestEntry& entry) const {
     if (only_read_real_buckets_ && entry.Bucket() < 0) {
         return false;
     }
diff --git a/src/paimon/core/operation/file_store_scan.h b/src/paimon/core/operation/file_store_scan.h
index 5fe128da1..b73ad095d 100644
--- a/src/paimon/core/operation/file_store_scan.h
+++ b/src/paimon/core/operation/file_store_scan.h
@@ -63,6 +63,7 @@ namespace paimon {
 class CacheKey;
 class Executor;
 class FileKind;
+class InternalRow;
 class ManifestFile;
 class ManifestFileMeta;
 class ManifestList;
@@ -276,13 +277,12 @@ class FileStoreScan {
     Status ReadManifestEntries(const std::vector<ManifestFileMeta>& manifest_metas,
                                std::vector<ManifestEntry>* manifest_entries) const;
 
-    Status ReadManifestEntriesWithCache(const Snapshot& snapshot,
-                                        const std::vector<ManifestFileMeta>& bucket_manifest_metas,
-                                        int32_t bucket,
-                                        std::vector<ManifestEntry>* manifest_entries,
-                                        bool* cache_hit) const;
+    /// Read request-filtered entries while caching complete bucket snapshots.
+    /// @return Number of entries considered before request-specific filtering.
+    Result<uint64_t> ReadManifestEntriesWithCache(
+        const Snapshot& snapshot, const std::vector<ManifestFileMeta>& bucket_manifest_metas,
+        int32_t bucket, std::vector<ManifestEntry>* manifest_entries, bool* cache_hit) const;
     std::shared_ptr<CacheKey> CreateSnapshotLiveManifestEntriesCacheKey(int32_t bucket) const;
-    Result<SnapshotLiveManifestEntries> LoadSnapshotLiveManifestEntries(int32_t bucket) const;
     Status StoreSnapshotLiveManifestEntries(int32_t bucket,
                                             const SnapshotLiveManifestEntries& entries) const;
 
@@ -309,7 +309,13 @@ class FileStoreScan {
     Status ReadManifestFileMeta(const ManifestFileMeta& manifest,
                                 std::vector<ManifestEntry>* entries) const;
 
+    /// If provided, predicate_failed must initially be false and outlive the synchronous callback.
+    /// Only predicate evaluation errors set it; partition decoding errors leave it unchanged.
+    std::function<Result<bool>(const InternalRow&)> CreateManifestEntryRowFilter(
+        bool* predicate_failed) const;
+
     Result<bool> FilterManifestEntry(const ManifestEntry& entry) const;
+    Result<bool> FilterNonPartitionManifestEntry(const ManifestEntry& entry) const;
 
  protected:
     std::shared_ptr<MemoryPool> pool_;
diff --git a/src/paimon/core/operation/file_store_scan_test.cpp b/src/paimon/core/operation/file_store_scan_test.cpp
index c4f0289e7..ef784f763 100644
--- a/src/paimon/core/operation/file_store_scan_test.cpp
+++ b/src/paimon/core/operation/file_store_scan_test.cpp
@@ -21,14 +21,19 @@
 #include <string>
 
 #include "arrow/type.h"
+#include "fmt/format.h"
 #include "gtest/gtest.h"
+#include "paimon/common/io/memory_segment_output_stream.h"
 #include "paimon/core/io/data_file_meta.h"
 #include "paimon/core/manifest/file_kind.h"
 #include "paimon/core/manifest/file_source.h"
+#include "paimon/core/manifest/manifest_entry_serializer.h"
 #include "paimon/core/manifest/snapshot_live_manifest_entries.h"
 #include "paimon/core/stats/simple_stats.h"
 #include "paimon/data/timestamp.h"
+#include "paimon/memory/bytes.h"
 #include "paimon/memory/memory_segment.h"
+#include "paimon/testing/utils/binary_row_generator.h"
 #include "paimon/testing/utils/testharness.h"
 
 namespace paimon::test {
@@ -190,54 +195,6 @@ TEST_F(FileStoreScanTest, TestFilterManifestByRowRanges) {
     ASSERT_TRUE(file_store_scan->FilterManifestByRowRanges(manifest2));
 }
 
-TEST_F(FileStoreScanTest, TestSnapshotLiveManifestEntries) {
-    std::vector<ManifestEntry> snapshot1;
-    ASSERT_OK_AND_ASSIGN(
-        auto file1,
-        DataFileMeta::ForAppend("file-1", /*file_size=*/10, /*row_count=*/1,
-                                SimpleStats::EmptyStats(), /*min_sequence_number=*/0,
-                                /*max_sequence_number=*/0, /*schema_id=*/0,
-                                /*file_source=*/std::nullopt, /*value_stats_cols=*/std::nullopt,
-                                /*external_path=*/std::nullopt, /*first_row_id=*/std::nullopt,
-                                /*write_cols=*/std::nullopt));
-    snapshot1.emplace_back(FileKind::Add(), BinaryRow::EmptyRow(), /*bucket=*/0,
-                           /*total_buckets=*/1, file1);
-    SnapshotLiveManifestEntries entries(/*max_snapshots=*/2);
-    entries.Put(/*snapshot_id=*/1, std::move(snapshot1));
-    ASSERT_EQ(entries.Size(), 1);
-    auto hit = entries.LatestBeforeOrEqual(/*snapshot_id=*/1);
-    ASSERT_TRUE(hit);
-    ASSERT_EQ(hit->snapshot_id, 1);
-    ASSERT_EQ(hit->entries->size(), 1);
-    ASSERT_EQ((*hit->entries)[0].FileName(), "file-1");
-    auto latest_before_2 = entries.LatestBeforeOrEqual(/*snapshot_id=*/2);
-    ASSERT_TRUE(latest_before_2);
-    ASSERT_EQ(latest_before_2->snapshot_id, 1);
-
-    std::vector<ManifestEntry> snapshot3;
-    ASSERT_OK_AND_ASSIGN(
-        auto file3,
-        DataFileMeta::ForAppend("file-3", /*file_size=*/10, /*row_count=*/1,
-                                SimpleStats::EmptyStats(), /*min_sequence_number=*/0,
-                                /*max_sequence_number=*/0, /*schema_id=*/0,
-                                /*file_source=*/std::nullopt, /*value_stats_cols=*/std::nullopt,
-                                /*external_path=*/std::nullopt, /*first_row_id=*/std::nullopt,
-                                /*write_cols=*/std::nullopt));
-    snapshot3.emplace_back(FileKind::Add(), BinaryRow::EmptyRow(), /*bucket=*/0,
-                           /*total_buckets=*/1, file3);
-    entries.Put(/*snapshot_id=*/3, std::move(snapshot3));
-
-    auto latest_before_4 = entries.LatestBeforeOrEqual(/*snapshot_id=*/4);
-    ASSERT_TRUE(latest_before_4);
-    ASSERT_EQ(latest_before_4->snapshot_id, 3);
-
-    entries.Put(/*snapshot_id=*/5, {});
-    ASSERT_EQ(entries.Size(), 2);
-    ASSERT_FALSE(entries.LatestBeforeOrEqual(/*snapshot_id=*/1));
-    ASSERT_TRUE(entries.LatestBeforeOrEqual(/*snapshot_id=*/3));
-    ASSERT_TRUE(entries.LatestBeforeOrEqual(/*snapshot_id=*/5));
-}
-
 TEST_F(FileStoreScanTest, TestSnapshotLiveManifestEntriesSerialization) {
     std::vector<ManifestEntry> manifest_entries;
     ASSERT_OK_AND_ASSIGN(
@@ -259,18 +216,180 @@ TEST_F(FileStoreScanTest, TestSnapshotLiveManifestEntriesSerialization) {
                          SnapshotLiveManifestEntries::Deserialize(
                              MemorySegment::Wrap(bytes), /*max_snapshots=*/2, GetDefaultPool()));
     ASSERT_EQ(deserialized.Size(), 2);
-    auto hit = deserialized.LatestBeforeOrEqual(/*snapshot_id=*/2);
-    ASSERT_TRUE(hit);
-    ASSERT_EQ(hit->snapshot_id, 1);
-    ASSERT_EQ(hit->entries->size(), 1);
-    ASSERT_EQ((*hit->entries)[0].FileName(), "file-1");
-    ASSERT_EQ(deserialized.LatestBeforeOrEqual(/*snapshot_id=*/4)->snapshot_id, 3);
+    ASSERT_OK_AND_ASSIGN(auto round_trip, deserialized.Serialize(GetDefaultPool()));
+    ASSERT_EQ(*bytes, *round_trip);
 
     ASSERT_OK_AND_ASSIGN(auto evicted_deserialized, SnapshotLiveManifestEntries::Deserialize(
                                                         MemorySegment::Wrap(bytes),
                                                         /*max_snapshots=*/1, GetDefaultPool()));
     ASSERT_EQ(evicted_deserialized.Size(), 1);
-    ASSERT_FALSE(evicted_deserialized.LatestBeforeOrEqual(/*snapshot_id=*/1));
-    ASSERT_EQ(evicted_deserialized.LatestBeforeOrEqual(/*snapshot_id=*/3)->snapshot_id, 3);
+    ASSERT_OK_AND_ASSIGN(auto retained, evicted_deserialized.Serialize(GetDefaultPool()));
+    // Use a larger read limit to verify that Deserialize itself evicted the older snapshot.
+    ASSERT_OK_AND_ASSIGN(
+        auto evicted, SnapshotLiveManifestEntries::ReadSnapshot(MemorySegment::Wrap(retained), 1, 2,
+                                                                nullptr, GetDefaultPool()));
+    ASSERT_FALSE(evicted);
+    ASSERT_OK_AND_ASSIGN(
+        auto empty, SnapshotLiveManifestEntries::ReadSnapshot(MemorySegment::Wrap(retained), 3, 2,
+                                                              nullptr, GetDefaultPool()));
+    ASSERT_TRUE(empty);
+    ASSERT_TRUE(empty->entries.empty());
 }
+
+TEST_F(FileStoreScanTest, SelectExactRetainedSnapshot) {
+    auto pool = GetDefaultPool();
+    std::vector<ManifestEntry> records;
+    for (int32_t partition = 0; partition < 3; ++partition) {
+        ASSERT_OK_AND_ASSIGN(
+            auto file,
+            DataFileMeta::ForAppend(fmt::format("file-{}", partition), 10, partition + 1,
+                                    SimpleStats::EmptyStats(), 0, 0, 0, std::nullopt, std::nullopt,
+                                    std::nullopt, std::nullopt, std::nullopt));
+        records.emplace_back(FileKind::Add(),
+                             BinaryRowGenerator::GenerateRow({partition}, pool.get()), 0, 1, file);
+    }
+    SnapshotLiveManifestEntries entries(3);
+    entries.Put(1, {records[0]});
+    entries.Put(3, {records[1], records[2]});
+    entries.Put(5, {});
+    ASSERT_OK_AND_ASSIGN(auto bytes, entries.Serialize(pool));
+    const MemorySegment segment = MemorySegment::Wrap(bytes);
+    ASSERT_OK_AND_ASSIGN(auto first,
+                         SnapshotLiveManifestEntries::ReadSnapshot(segment, 1, 3, nullptr, pool));
+    ASSERT_TRUE(first);
+    ASSERT_EQ(first->entries, std::vector<ManifestEntry>({records[0]}));
+    ASSERT_OK_AND_ASSIGN(auto third,
+                         SnapshotLiveManifestEntries::ReadSnapshot(segment, 3, 3, nullptr, pool));
+    ASSERT_TRUE(third);
+    ASSERT_EQ(third->entries, std::vector<ManifestEntry>({records[1], records[2]}));
+    int32_t filter_calls = 0;
+    ASSERT_OK_AND_ASSIGN(auto selected,
+                         SnapshotLiveManifestEntries::ReadSnapshot(
+                             segment, 3, 2,
+                             [&filter_calls](const InternalRow& row) -> Result<bool> {
+                                 ++filter_calls;
+                                 PAIMON_ASSIGN_OR_RAISE(
+                                     BinaryRow partition,
+                                     ManifestEntrySerializer::GetPartitionView(row));
+                                 return partition.GetInt(0) == 2;
+                             },
+                             pool));
+    ASSERT_TRUE(selected);
+    ASSERT_EQ(selected->candidate_entries, 2);
+    ASSERT_EQ(filter_calls, 2);
+    ASSERT_EQ(selected->entries, std::vector<ManifestEntry>({records[2]}));
+    ASSERT_OK_AND_ASSIGN(auto absent,
+                         SnapshotLiveManifestEntries::ReadSnapshot(segment, 2, 3, nullptr, pool));
+    ASSERT_FALSE(absent);
+    ASSERT_OK_AND_ASSIGN(auto evicted,
+                         SnapshotLiveManifestEntries::ReadSnapshot(segment, 3, 1, nullptr, pool));
+    ASSERT_FALSE(evicted);
+    ASSERT_OK_AND_ASSIGN(auto disabled,
+                         SnapshotLiveManifestEntries::ReadSnapshot(segment, 3, 0, nullptr, pool));
+    ASSERT_FALSE(disabled);
+    // Skipping other snapshots must not allocate row payloads in the reader's pool.
+    std::shared_ptr<MemoryPool> read_pool = GetMemoryPool();
+    ASSERT_OK_AND_ASSIGN(
+        auto empty, SnapshotLiveManifestEntries::ReadSnapshot(segment, 5, 1, nullptr, read_pool));
+    ASSERT_TRUE(empty);
+    ASSERT_TRUE(empty->entries.empty());
+    ASSERT_EQ(read_pool->MaxMemoryUsage(), 0);
+    ASSERT_NOK_WITH_MSG(SnapshotLiveManifestEntries::ReadSnapshot(
+                            segment, 3, 3,
+                            [](const InternalRow&) -> Result<bool> {
+                                return Status::IOError("injected filter error");
+                            },
+                            pool),
+                        "injected filter error");
+
+    entries.Put(7, {});
+    ASSERT_EQ(entries.Size(), 3);
+    ASSERT_OK_AND_ASSIGN(auto updated, entries.Serialize(pool));
+    // Put must evict the oldest snapshot even when the reader permits more snapshots.
+    ASSERT_OK_AND_ASSIGN(auto oldest, SnapshotLiveManifestEntries::ReadSnapshot(
+                                          MemorySegment::Wrap(updated), 1, 4, nullptr, pool));
+    ASSERT_FALSE(oldest);
+    ASSERT_OK_AND_ASSIGN(auto retained, SnapshotLiveManifestEntries::ReadSnapshot(
+                                            MemorySegment::Wrap(updated), 3, 4, nullptr, pool));
+    ASSERT_TRUE(retained);
+    ASSERT_EQ(retained->entries, std::vector<ManifestEntry>({records[1], records[2]}));
+}
+
+TEST_F(FileStoreScanTest, SkipDecodingFilteredManifestEntries) {
+    auto pool = GetDefaultPool();
+    ASSERT_OK_AND_ASSIGN(
+        auto file,
+        DataFileMeta::ForAppend("file", 10, 1, SimpleStats::EmptyStats(), 0, 0, 0, std::nullopt,
+                                std::nullopt, std::nullopt, std::nullopt, std::nullopt));
+    // Like the bucket-pruning tests, use an invalid kind to detect full entry decoding.
+    ManifestEntry excluded(FileKind(static_cast<int8_t>(2)),
+                           BinaryRowGenerator::GenerateRow({10}, pool.get()), 0, 1, file);
+    SnapshotLiveManifestEntries entries(1);
+    entries.Put(1, {excluded});
+    ASSERT_OK_AND_ASSIGN(auto bytes, entries.Serialize(pool));
+    const MemorySegment segment = MemorySegment::Wrap(bytes);
+    ASSERT_OK_AND_ASSIGN(auto selected, SnapshotLiveManifestEntries::ReadSnapshot(
+                                            segment, 1, 1,
+                                            [](const InternalRow& row) -> Result<bool> {
+                                                PAIMON_ASSIGN_OR_RAISE(
+                                                    BinaryRow partition,
+                                                    ManifestEntrySerializer::GetPartitionView(row));
+                                                return partition.GetInt(0) == 20;
+                                            },
+                                            pool));
+    ASSERT_TRUE(selected);
+    ASSERT_TRUE(selected->entries.empty());
+    ASSERT_NOK_WITH_MSG(SnapshotLiveManifestEntries::ReadSnapshot(segment, 1, 1, nullptr, pool),
+                        "Unsupported byte value 2 for file kind.");
+}
+
+TEST_F(FileStoreScanTest, RejectUnsupportedSnapshotEntryVersionBeforeFiltering) {
+    auto pool = GetDefaultPool();
+    // Only the version is present: the reader must reject it before the filter accesses fields.
+    BinaryRow row = BinaryRowGenerator::GenerateRow({3}, pool.get());
+    BinaryRowSerializer serializer(1, pool);
+    MemorySegmentOutputStream out(64, pool);
+    out.WriteValue<int32_t>(0x534d4543);
+    out.WriteValue<int32_t>(1);
+    out.WriteValue<int64_t>(1);
+    out.WriteValue<int32_t>(1);
+    ASSERT_OK(serializer.Serialize(row, &out));
+    bool filter_called = false;
+    ASSERT_NOK_WITH_MSG(SnapshotLiveManifestEntries::ReadSnapshot(
+                            out.Segments().front(), 1, 1,
+                            [&filter_called](const InternalRow&) -> Result<bool> {
+                                filter_called = true;
+                                return false;
+                            },
+                            pool),
+                        "Unsupported version: 3");
+    ASSERT_FALSE(filter_called);
+}
+
+TEST_F(FileStoreScanTest, RejectInvalidSnapshotSelectionHeader) {
+    auto pool = GetDefaultPool();
+    ASSERT_OK_AND_ASSIGN(auto absent, SnapshotLiveManifestEntries::ReadSnapshot(MemorySegment(), 1,
+                                                                                3, nullptr, pool));
+    ASSERT_FALSE(absent);
+    auto header = [&pool](int32_t magic, int32_t snapshots, int32_t entries) {
+        MemorySegmentOutputStream out(64, pool);
+        out.WriteValue<int32_t>(magic);
+        out.WriteValue<int32_t>(snapshots);
+        out.WriteValue<int64_t>(1);
+        out.WriteValue<int32_t>(entries);
+        return out.Segments().front();
+    };
+    auto segment = header(0, 1, 0);
+    ASSERT_NOK_WITH_MSG(SnapshotLiveManifestEntries::ReadSnapshot(segment, 1, 3, nullptr, pool),
+                        "invalid snapshot live manifest entries magic");
+    segment = header(0x534d4543, -1, 0);
+    ASSERT_NOK_WITH_MSG(SnapshotLiveManifestEntries::ReadSnapshot(segment, 1, 3, nullptr, pool),
+                        "snapshot count is negative");
+    segment = header(0x534d4543, 1, -1);
+    ASSERT_NOK_WITH_MSG(SnapshotLiveManifestEntries::ReadSnapshot(segment, 1, 3, nullptr, pool),
+                        "entry count is negative");
+    auto truncated = MemorySegment::WrapView(segment.Data(), 7);
+    ASSERT_NOK(SnapshotLiveManifestEntries::ReadSnapshot(truncated, 1, 3, nullptr, pool));
+}
+
 }  // namespace paimon::test