apache · wangyong9999 · Sep 19, 2026
diff --git a/docs/source/user_guide/manifest_entry_cache.rst b/docs/source/user_guide/manifest_entry_cache.rst
@@ -28,8 +28,8 @@ scans that target the same bucket.
 
 The cache stores decoded and merged live manifest entries by table path, branch,
 and bucket for ``ScanMode::ALL``. Each cache value can retain several snapshot
-results for that bucket. Exact snapshot hits are served from the cache; cache
-misses rebuild the target snapshot bucket from the target snapshot's data
+results for that bucket. Exact snapshot hits are served from the cache without reading or decoding the
+manifest lists again; cache misses rebuild the target snapshot bucket from the target snapshot's data
 manifests and store the rebuilt live entries.
 
 Request-specific filters are not stored in the cache. Partition, level, and
@@ -89,3 +89,7 @@ The scan metrics expose existing counters for the last scan:
 
 - ``lastScannedManifests``: how many manifest files were loaded during this
   scan before manifest entry decoding.
+
+On an exact cache hit, ``lastScannedManifests`` is zero. The cached table file
+count preserves the skipped-file metric even when manifest lists are bypassed.
+This changes only transient cache values; persisted table formats are unchanged.
diff --git a/src/paimon/core/manifest/snapshot_live_manifest_entries.cpp b/src/paimon/core/manifest/snapshot_live_manifest_entries.cpp
@@ -34,7 +34,8 @@
 namespace paimon {
 namespace {
 
-constexpr int32_t kMagic = 0x534d4543;  // SMEC
+// Cache-only format: include the table file count used by scan metrics.
+constexpr int32_t kMagic = 0x534d4544;  // SMED
 
 size_t NormalizeMaxSnapshots(int32_t max_snapshots) {
     return static_cast<size_t>(std::max(0, max_snapshots));
@@ -68,15 +69,17 @@ std::optional<SnapshotLiveManifestEntries::Entry> SnapshotLiveManifestEntries::L
         return std::optional<Entry>();
     }
     --iter;
-    return Entry{iter->first, iter->second};
+    return iter->second;
 }
 
-void SnapshotLiveManifestEntries::Put(int64_t snapshot_id, std::vector<ManifestEntry>&& entries) {
+void SnapshotLiveManifestEntries::Put(int64_t snapshot_id, std::vector<ManifestEntry>&& entries,
+                                      int64_t total_data_files) {
     if (NormalizeMaxSnapshots(max_snapshots_) == 0) {
         return;
     }
     entries_by_snapshot_[snapshot_id] =
-        std::make_shared<const std::vector<ManifestEntry>>(std::move(entries));
+        Entry{snapshot_id, total_data_files,
+              std::make_shared<const std::vector<ManifestEntry>>(std::move(entries))};
     EvictIfNeeded();
 }
 
@@ -93,7 +96,8 @@ Result<std::shared_ptr<Bytes>> SnapshotLiveManifestEntries::Serialize(
     ManifestEntrySerializer serializer(pool);
     for (const auto& [snapshot_id, entries] : entries_by_snapshot_) {
         out.WriteValue<int64_t>(snapshot_id);
-        PAIMON_RETURN_NOT_OK(serializer.SerializeList(*entries, &out));
+        out.WriteValue<int64_t>(entries.total_data_files);
+        PAIMON_RETURN_NOT_OK(serializer.SerializeList(*entries.entries, &out));
     }
     return ToBytes(out, pool);
 }
@@ -121,9 +125,9 @@ Result<SnapshotLiveManifestEntries> SnapshotLiveManifestEntries::Deserialize(
     ManifestEntrySerializer serializer(pool);
     for (int32_t i = 0; i < snapshot_count; i++) {
         PAIMON_ASSIGN_OR_RAISE(int64_t snapshot_id, in.ReadValue<int64_t>());
+        PAIMON_ASSIGN_OR_RAISE(int64_t total_data_files, in.ReadValue<int64_t>());
         PAIMON_ASSIGN_OR_RAISE(std::vector<ManifestEntry> entries, serializer.DeserializeList(&in));
-        snapshot_live_manifest_entries.entries_by_snapshot_[snapshot_id] =
-            std::make_shared<const std::vector<ManifestEntry>>(std::move(entries));
+        snapshot_live_manifest_entries.Put(snapshot_id, std::move(entries), total_data_files);
     }
     snapshot_live_manifest_entries.EvictIfNeeded();
     return snapshot_live_manifest_entries;

diff --git a/src/paimon/core/manifest/snapshot_live_manifest_entries.h b/src/paimon/core/manifest/snapshot_live_manifest_entries.h
@@ -42,13 +42,15 @@ class SnapshotLiveManifestEntries {
  public:
     struct Entry {
         int64_t snapshot_id;
+        int64_t total_data_files;
         std::shared_ptr<const std::vector<ManifestEntry>> entries;
     };
 
     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);
+    void Put(int64_t snapshot_id, std::vector<ManifestEntry>&& entries,
+             int64_t total_data_files = -1);
     size_t Size() const;
 
     Result<std::shared_ptr<Bytes>> Serialize(const std::shared_ptr<MemoryPool>& pool) const;
@@ -59,7 +61,7 @@ class SnapshotLiveManifestEntries {
  private:
     void EvictIfNeeded();
 
-    std::map<int64_t, std::shared_ptr<const std::vector<ManifestEntry>>> entries_by_snapshot_;
+    std::map<int64_t, Entry> entries_by_snapshot_;
     int32_t max_snapshots_;
 };
 

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
@@ -61,6 +61,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"
@@ -754,7 +755,9 @@ TEST(AppendOnlyFileStoreScanTest, TestDropStatsAfterFiltering) {
 TEST(AppendOnlyFileStoreScanTest, TestSnapshotLiveManifestCachePath) {
     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>(
+        std::map<CacheKind, int64_t>{{CacheKind::MANIFEST, 16 * 1024 * 1024},
+                                     {CacheKind::SNAPSHOT_LIVE_MANIFEST, 16 * 1024 * 1024}});
 
     // First scan on snapshot 5: cache miss, entries rebuilt from all manifests.
     auto scan_first = BuildScan(table_path, cache, /*bucket=*/0);
@@ -778,6 +781,11 @@ TEST(AppendOnlyFileStoreScanTest, TestSnapshotLiveManifestCachePath) {
     ASSERT_EQ(first_cache_hit, 0);
     ASSERT_EQ(first_cache_misses, 1);
 
+    const int64_t manifest_gets = cache->GetCount(CacheKind::MANIFEST);
+    ASSERT_GT(manifest_gets, 0);
+    ASSERT_OK_AND_ASSIGN(uint64_t first_skipped,
+                         first_metrics->GetCounter(ScanMetrics::LAST_SCAN_SKIPPED_TABLE_FILES));
+
     // Second scan on the same snapshot should read the same bucket live entries from cache.
     auto scan_second = BuildScan(table_path, cache, /*bucket=*/0);
     scan_second->WithSnapshot(snapshot_5);
@@ -794,9 +802,36 @@ TEST(AppendOnlyFileStoreScanTest, TestSnapshotLiveManifestCachePath) {
         uint64_t materialized_rows,
         second_metrics->GetCounter(ScanMetrics::LAST_LAZY_DECODE_MATERIALIZED_ROWS));
     ASSERT_EQ(second_cache_hit, 1);
+    ASSERT_EQ(cache->GetCount(CacheKind::MANIFEST), manifest_gets);
+    ASSERT_OK_AND_ASSIGN(uint64_t second_scanned,
+                         second_metrics->GetCounter(ScanMetrics::LAST_SCANNED_MANIFESTS));
+    ASSERT_OK_AND_ASSIGN(uint64_t second_skipped,
+                         second_metrics->GetCounter(ScanMetrics::LAST_SCAN_SKIPPED_TABLE_FILES));
+    ASSERT_EQ(second_scanned, 0);
+    ASSERT_EQ(second_skipped, first_skipped);
     ASSERT_EQ(second_cache_hits, 1);
     ASSERT_GE(scanned_rows, materialized_rows);
     ASSERT_OK(second_metrics->GetHistogramStats(ScanMetrics::SNAPSHOT_CACHE_LOAD_DURATION));
+
+    // A different predicate must still filter the cached, unfiltered bucket entries.
+    std::shared_ptr<Predicate> predicate =
+        PredicateBuilder::Equal(0, "f0", FieldType::STRING, Literal(FieldType::STRING, "David", 5));
+    auto filtered = BuildScan(table_path, cache, /*bucket=*/0, predicate);
+    filtered->WithSnapshot(snapshot_5);
+    ASSERT_OK_AND_ASSIGN(auto filtered_plan, filtered->CreatePlan());
+    auto expected = BuildScan(table_path, nullptr, /*bucket=*/0, predicate);
+    expected->WithSnapshot(snapshot_5);
+    ASSERT_OK_AND_ASSIGN(auto expected_plan, expected->CreatePlan());
+    ASSERT_EQ(SortedFileNames(filtered_plan->Files()), SortedFileNames(expected_plan->Files()));
+    ASSERT_EQ(cache->GetCount(CacheKind::MANIFEST), manifest_gets);
+
+    // Eviction must rebuild from source; an exact-hit shortcut cannot survive invalidation.
+    cache->InvalidateAll();
+    auto rebuilt = BuildScan(table_path, cache, /*bucket=*/0);
+    rebuilt->WithSnapshot(snapshot_5);
+    ASSERT_OK_AND_ASSIGN(auto rebuilt_plan, rebuilt->CreatePlan());
+    ASSERT_EQ(first_file_names, SortedFileNames(rebuilt_plan->Files()));
+    ASSERT_GT(cache->GetCount(CacheKind::MANIFEST), manifest_gets);
 }
 
 TEST(AppendOnlyFileStoreScanTest, TestSnapshotLiveManifestCacheRebuildOnMiss) {

diff --git a/src/paimon/core/operation/file_store_scan.cpp b/src/paimon/core/operation/file_store_scan.cpp
@@ -143,8 +143,8 @@ Result<std::shared_ptr<FileStoreScan::RawPlan>> FileStoreScan::CreatePlan() cons
     std::optional<Snapshot> snapshot;
     std::vector<ManifestFileMeta> all_manifest_file_metas;
     std::vector<ManifestFileMeta> filtered_manifest_file_metas;
-    PAIMON_RETURN_NOT_OK(
-        ReadManifests(&snapshot, &all_manifest_file_metas, &filtered_manifest_file_metas));
+    PAIMON_ASSIGN_OR_RAISE(snapshot, ReadSnapshot());
+    int64_t all_data_files = 0;
 
     std::vector<ManifestEntry> manifest_entries;
     std::optional<int32_t> cache_bucket = bucket_filter_;
@@ -159,9 +159,9 @@ 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));
+        PAIMON_RETURN_NOT_OK(ReadManifestEntriesWithCache(
+            snapshot.value(), &all_manifest_file_metas, &filtered_manifest_file_metas,
+            cache_bucket.value(), &manifest_entries, &snapshot_cache_hit, &all_data_files));
         lazy_decode_scanned_rows = manifest_entries.size();
         std::vector<ManifestEntry> filtered_entries;
         filtered_entries.reserve(manifest_entries.size());
@@ -173,6 +173,8 @@ Result<std::shared_ptr<FileStoreScan::RawPlan>> FileStoreScan::CreatePlan() cons
         }
         manifest_entries = std::move(filtered_entries);
     } else {
+        PAIMON_RETURN_NOT_OK(
+            ReadManifestLists(snapshot, &all_manifest_file_metas, &filtered_manifest_file_metas));
         lazy_decode_scanned_rows = std::accumulate(
             filtered_manifest_file_metas.begin(), filtered_manifest_file_metas.end(), uint64_t{0},
             [](uint64_t sum, const ManifestFileMeta& meta) {
@@ -211,11 +213,13 @@ Result<std::shared_ptr<FileStoreScan::RawPlan>> FileStoreScan::CreatePlan() cons
             entry = entry.CopyWithoutStats();
         }
     }
-    const int64_t all_data_files = std::accumulate(
-        all_manifest_file_metas.begin(), all_manifest_file_metas.end(), int64_t{0},
-        [](const int64_t sum, const ManifestFileMeta& manifest_file_meta) {
-            return sum + manifest_file_meta.NumAddedFiles() - manifest_file_meta.NumDeletedFiles();
-        });
+    if (!snapshot_cache_hit) {
+        all_data_files =
+            std::accumulate(all_manifest_file_metas.begin(), all_manifest_file_metas.end(),
+                            int64_t{0}, [](int64_t sum, const ManifestFileMeta& meta) {
+                                return sum + meta.NumAddedFiles() - meta.NumDeletedFiles();
+                            });
+    }
     const uint64_t scan_duration_ms = duration.Get();
     metrics_->SetCounter(ScanMetrics::LAST_SCAN_DURATION, scan_duration_ms);
     metrics_->ObserveHistogram(ScanMetrics::SCAN_DURATION, static_cast<double>(scan_duration_ms));
@@ -247,27 +251,31 @@ Result<std::shared_ptr<FileStoreScan::RawPlan>> FileStoreScan::CreatePlan() cons
                                                     std::move(manifest_entries));
 }
 
-Status FileStoreScan::ReadManifests(std::optional<Snapshot>* snapshot_ptr,
-                                    std::vector<ManifestFileMeta>* all_manifests_ptr,
-                                    std::vector<ManifestFileMeta>* filter_manifests_ptr) const {
-    auto& snapshot = *snapshot_ptr;
-    auto& all_manifests = *all_manifests_ptr;
-    auto& filtered_manifests = *filter_manifests_ptr;
-    if (specified_snapshot_ != std::nullopt) {
-        snapshot = specified_snapshot_;
-    } else {
-        PAIMON_ASSIGN_OR_RAISE(snapshot, snapshot_manager_->LatestSnapshot());
+Result<std::optional<Snapshot>> FileStoreScan::ReadSnapshot() const {
+    if (specified_snapshot_) {
+        return specified_snapshot_;
     }
-    if (snapshot == std::nullopt) {
-        all_manifests = std::vector<ManifestFileMeta>();
-        filtered_manifests = std::vector<ManifestFileMeta>();
+    return snapshot_manager_->LatestSnapshot();
+}
+
+Status FileStoreScan::ReadManifests(std::optional<Snapshot>* snapshot,
+                                    std::vector<ManifestFileMeta>* all_manifests,
+                                    std::vector<ManifestFileMeta>* filtered_manifests) const {
+    PAIMON_ASSIGN_OR_RAISE(*snapshot, ReadSnapshot());
+    return ReadManifestLists(*snapshot, all_manifests, filtered_manifests);
+}
+
+Status FileStoreScan::ReadManifestLists(const std::optional<Snapshot>& snapshot,
+                                        std::vector<ManifestFileMeta>* all_manifests,
+                                        std::vector<ManifestFileMeta>* filtered_manifests) const {
+    if (!snapshot) {
         return Status::OK();
     }
-    PAIMON_RETURN_NOT_OK(ReadManifestsWithSnapshot(snapshot.value(), &all_manifests));
-    for (const auto& meta : all_manifests) {
-        PAIMON_ASSIGN_OR_RAISE(bool filter_meta_result, FilterManifestFileMeta(meta));
-        if (filter_meta_result) {
-            filtered_manifests.push_back(meta);
+    PAIMON_RETURN_NOT_OK(ReadManifestsWithSnapshot(snapshot.value(), all_manifests));
+    for (const auto& meta : *all_manifests) {
+        PAIMON_ASSIGN_OR_RAISE(bool keep, FilterManifestFileMeta(meta));
+        if (keep) {
+            filtered_manifests->push_back(meta);
         }
     }
     return Status::OK();
@@ -360,8 +368,9 @@ Status FileStoreScan::ReadManifestEntries(const std::vector<ManifestFileMeta>& m
 // can be returned directly; cache misses rebuild the target snapshot bucket from the target
 // snapshot's data manifests.
 Status FileStoreScan::ReadManifestEntriesWithCache(
-    const Snapshot& snapshot, const std::vector<ManifestFileMeta>& all_manifest_metas,
-    int32_t bucket, std::vector<ManifestEntry>* manifest_entries, bool* cache_hit) const {
+    const Snapshot& snapshot, std::vector<ManifestFileMeta>* all_manifest_metas,
+    std::vector<ManifestFileMeta>* filtered_manifest_metas, int32_t bucket,
+    std::vector<ManifestEntry>* manifest_entries, bool* cache_hit, int64_t* all_data_files) const {
     Duration cache_load_duration;
     PAIMON_ASSIGN_OR_RAISE(SnapshotLiveManifestEntries cached_entries,
                            LoadSnapshotLiveManifestEntries(bucket));
@@ -371,25 +380,34 @@ Status FileStoreScan::ReadManifestEntriesWithCache(
                                static_cast<double>(cache_load_duration_ms));
     std::optional<SnapshotLiveManifestEntries::Entry> cached =
         cached_entries.LatestBeforeOrEqual(snapshot.Id());
-    if (cached && cached->snapshot_id == snapshot.Id()) {
+    if (cached && cached->snapshot_id == snapshot.Id() && cached->total_data_files >= 0) {
+        *all_data_files = cached->total_data_files;
         *cache_hit = true;
         *manifest_entries = *cached->entries;
         return Status::OK();
     }
     *cache_hit = false;
+    // Exact hits need neither manifest-list IO nor decoding. A miss resolves lists from
+    // the same snapshot selected above, then populates the unfiltered bucket cache.
+    PAIMON_RETURN_NOT_OK(ReadManifestLists(snapshot, all_manifest_metas, filtered_manifest_metas));
+    *all_data_files =
+        std::accumulate(all_manifest_metas->begin(), all_manifest_metas->end(), int64_t{0},
+                        [](int64_t sum, const ManifestFileMeta& meta) {
+                            return sum + meta.NumAddedFiles() - meta.NumDeletedFiles();
+                        });
 
     // 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) {
+    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));
     std::vector<ManifestEntry> cache_entries = *manifest_entries;
-    cached_entries.Put(snapshot.Id(), std::move(cache_entries));
+    cached_entries.Put(snapshot.Id(), std::move(cache_entries), *all_data_files);
     Duration cache_store_duration;
     PAIMON_RETURN_NOT_OK(StoreSnapshotLiveManifestEntries(bucket, cached_entries));
     const uint64_t cache_store_duration_ms = cache_store_duration.Get();

diff --git a/src/paimon/core/operation/file_store_scan.h b/src/paimon/core/operation/file_store_scan.h
@@ -258,6 +258,11 @@ class FileStoreScan {
         const std::shared_ptr<SimpleStatsEvolution>& evolution);
 
  private:
+    Result<std::optional<Snapshot>> ReadSnapshot() const;
+    Status ReadManifestLists(const std::optional<Snapshot>& snapshot,
+                             std::vector<ManifestFileMeta>* all_manifests,
+                             std::vector<ManifestFileMeta>* filtered_manifests) const;
+
     Status ReadManifests(std::optional<Snapshot>* snapshot_ptr,
                          std::vector<ManifestFileMeta>* all_manifests_ptr,
                          std::vector<ManifestFileMeta>* filtered_manifests_ptr) const;
@@ -269,10 +274,11 @@ class FileStoreScan {
                                std::vector<ManifestEntry>* manifest_entries) const;
 
     Status ReadManifestEntriesWithCache(const Snapshot& snapshot,
-                                        const std::vector<ManifestFileMeta>& bucket_manifest_metas,
+                                        std::vector<ManifestFileMeta>* all_manifest_metas,
+                                        std::vector<ManifestFileMeta>* filtered_manifest_metas,
                                         int32_t bucket,
                                         std::vector<ManifestEntry>* manifest_entries,
-                                        bool* cache_hit) const;
+                                        bool* cache_hit, int64_t* all_data_files) const;
     std::shared_ptr<CacheKey> CreateSnapshotLiveManifestEntriesCacheKey(int32_t bucket) const;
     Result<SnapshotLiveManifestEntries> LoadSnapshotLiveManifestEntries(int32_t bucket) const;
     Status StoreSnapshotLiveManifestEntries(int32_t bucket,

diff --git a/src/paimon/core/operation/file_store_scan_test.cpp b/src/paimon/core/operation/file_store_scan_test.cpp
@@ -251,8 +251,8 @@ TEST_F(FileStoreScanTest, TestSnapshotLiveManifestEntriesSerialization) {
     manifest_entries.emplace_back(FileKind::Add(), BinaryRow::EmptyRow(), /*bucket=*/0,
                                   /*total_buckets=*/1, file1);
     SnapshotLiveManifestEntries entries(/*max_snapshots=*/2);
-    entries.Put(/*snapshot_id=*/1, std::move(manifest_entries));
-    entries.Put(/*snapshot_id=*/3, {});
+    entries.Put(/*snapshot_id=*/1, std::move(manifest_entries), /*total_data_files=*/17);
+    entries.Put(/*snapshot_id=*/3, {}, /*total_data_files=*/0);
 
     ASSERT_OK_AND_ASSIGN(auto bytes, entries.Serialize(GetDefaultPool()));
     ASSERT_OK_AND_ASSIGN(auto deserialized,
@@ -262,9 +262,11 @@ TEST_F(FileStoreScanTest, TestSnapshotLiveManifestEntriesSerialization) {
     auto hit = deserialized.LatestBeforeOrEqual(/*snapshot_id=*/2);
     ASSERT_TRUE(hit);
     ASSERT_EQ(hit->snapshot_id, 1);
+    ASSERT_EQ(hit->total_data_files, 17);
     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_EQ(deserialized.LatestBeforeOrEqual(/*snapshot_id=*/4)->total_data_files, 0);
 
     ASSERT_OK_AND_ASSIGN(auto evicted_deserialized, SnapshotLiveManifestEntries::Deserialize(
                                                         MemorySegment::Wrap(bytes),