perf(scan): skip manifest lists on exact snapshot cache hits

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
@@ -120,3 +120,6 @@ 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.

src/paimon/core/manifest/snapshot_live_manifest_entries.cpp

@@ -34,7 +34,7 @@
 namespace paimon {
 namespace {
 
-constexpr int32_t kMagic = 0x534d4543;  // SMEC
+constexpr int32_t kMagic = 0x534d4544;  // SMED: includes per-snapshot total file counts
 
 size_t NormalizeMaxSnapshots(int32_t max_snapshots) {
     return static_cast<size_t>(std::max(0, max_snapshots));
@@ -61,12 +61,14 @@ std::shared_ptr<Bytes> ToBytes(const MemorySegmentOutputStream& out,
 SnapshotLiveManifestEntries::SnapshotLiveManifestEntries(int32_t max_snapshots)
     : max_snapshots_(max_snapshots) {}
 
-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));
+    total_data_files_by_snapshot_[snapshot_id] = total_data_files;
     EvictIfNeeded();
 }
 
@@ -83,6 +85,7 @@ 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);
+        out.WriteValue<int64_t>(total_data_files_by_snapshot_.at(snapshot_id));
         PAIMON_RETURN_NOT_OK(serializer.SerializeList(*entries, &out));
     }
     return ToBytes(out, pool);
@@ -111,7 +114,10 @@ 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.total_data_files_by_snapshot_[snapshot_id] =
+            total_data_files;
         snapshot_live_manifest_entries.entries_by_snapshot_[snapshot_id] =
             std::make_shared<const std::vector<ManifestEntry>>(std::move(entries));
     }
@@ -122,6 +128,7 @@ Result<SnapshotLiveManifestEntries> SnapshotLiveManifestEntries::Deserialize(
 void SnapshotLiveManifestEntries::EvictIfNeeded() {
     size_t max_snapshots = NormalizeMaxSnapshots(max_snapshots_);
     while (entries_by_snapshot_.size() > max_snapshots) {
+        total_data_files_by_snapshot_.erase(entries_by_snapshot_.begin()->first);
         entries_by_snapshot_.erase(entries_by_snapshot_.begin());
     }
 }
@@ -151,6 +158,7 @@ SnapshotLiveManifestEntries::ReadSnapshot(
     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>());
+        PAIMON_ASSIGN_OR_RAISE(int64_t total_data_files, 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()) {
@@ -165,6 +173,7 @@ SnapshotLiveManifestEntries::ReadSnapshot(
         if (keep) {
             selected.emplace();
             selected->candidate_entries = entry_count;
+            selected->total_data_files = total_data_files;
             if (!row_filter) {
                 selected->entries.reserve(entry_count);
             }

src/paimon/core/manifest/snapshot_live_manifest_entries.h

@@ -45,11 +45,12 @@ class SnapshotLiveManifestEntries {
     struct Selection {
         std::vector<ManifestEntry> entries;
         uint64_t candidate_entries = 0;
+        int64_t total_data_files = 0;
     };
 
     explicit SnapshotLiveManifestEntries(int32_t max_snapshots);
 
-    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);
     size_t Size() const;
 
     Result<std::shared_ptr<Bytes>> Serialize(const std::shared_ptr<MemoryPool>& pool) const;
@@ -59,7 +60,7 @@ class SnapshotLiveManifestEntries {
 
     /// 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
+    /// ManifestFile::ReadSelectedEntries. 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(
@@ -71,6 +72,7 @@ class SnapshotLiveManifestEntries {
     void EvictIfNeeded();
 
     std::map<int64_t, std::shared_ptr<const std::vector<ManifestEntry>>> entries_by_snapshot_;
+    std::map<int64_t, int64_t> total_data_files_by_snapshot_;
     int32_t max_snapshots_;
 };
 

src/paimon/core/operation/append_only_file_store_scan_test.cpp

@@ -815,7 +815,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);
@@ -839,6 +841,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);
@@ -855,9 +862,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) {
@@ -998,9 +1032,10 @@ TEST(AppendOnlyFileStoreScanTest, SnapshotCacheRebuildsCorruptValues) {
             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>(corruption == "magic" ? 0 : 0x534d4544);
         out.WriteValue<int32_t>(1);
         out.WriteValue<int64_t>(snapshot.Id());
+        out.WriteValue<int64_t>(1);
         out.WriteValue<int32_t>(1);
         ASSERT_OK(row_serializer.Serialize(row, &out));
         ASSERT_EQ(out.Segments().size(), 1);
@@ -1078,7 +1113,11 @@ TEST(AppendOnlyFileStoreScanTest, PartitionPruningAcrossCacheModes) {
                 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);
+                if (mode == 2 && scans > 0) {
+                    ASSERT_EQ(counting->stats_calls_.load(), 0);
+                } else {
+                    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);
@@ -1156,7 +1195,7 @@ TEST(AppendOnlyFileStoreScanTest, PartitionPruningSkipsExcludedMetadataInCreateP
         if (mode == 2) {
             // Seed the complete snapshot directly: rebuilding must decode every partition.
             SnapshotLiveManifestEntries entries(8);
-            entries.Put(1, {kept, excluded});
+            entries.Put(1, {kept, excluded}, /*total_data_files=*/2);
             ASSERT_OK(scan->StoreSnapshotLiveManifestEntries(0, entries));
         }
         for (int32_t attempt = 0; attempt < 2; ++attempt) {