diff --git a/docs/source/user_guide/manifest_entry_cache.rst b/docs/source/user_guide/manifest_entry_cache.rst index 97b33cdec..5ed6f0a5f 100644 --- 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 index 8982209a6..583b81541 100644 --- 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(std::max(0, max_snapshots)); @@ -68,15 +69,17 @@ std::optional SnapshotLiveManifestEntries::L return std::optional(); } --iter; - return Entry{iter->first, iter->second}; + return iter->second; } -void SnapshotLiveManifestEntries::Put(int64_t snapshot_id, std::vector&& entries) { +void SnapshotLiveManifestEntries::Put(int64_t snapshot_id, std::vector&& entries, + int64_t total_data_files) { if (NormalizeMaxSnapshots(max_snapshots_) == 0) { return; } entries_by_snapshot_[snapshot_id] = - std::make_shared>(std::move(entries)); + Entry{snapshot_id, total_data_files, + std::make_shared>(std::move(entries))}; EvictIfNeeded(); } @@ -93,7 +96,8 @@ Result> SnapshotLiveManifestEntries::Serialize( ManifestEntrySerializer serializer(pool); for (const auto& [snapshot_id, entries] : entries_by_snapshot_) { out.WriteValue(snapshot_id); - PAIMON_RETURN_NOT_OK(serializer.SerializeList(*entries, &out)); + out.WriteValue(entries.total_data_files); + PAIMON_RETURN_NOT_OK(serializer.SerializeList(*entries.entries, &out)); } return ToBytes(out, pool); } @@ -121,9 +125,9 @@ Result SnapshotLiveManifestEntries::Deserialize( ManifestEntrySerializer serializer(pool); for (int32_t i = 0; i < snapshot_count; i++) { PAIMON_ASSIGN_OR_RAISE(int64_t snapshot_id, in.ReadValue()); + PAIMON_ASSIGN_OR_RAISE(int64_t total_data_files, in.ReadValue()); PAIMON_ASSIGN_OR_RAISE(std::vector entries, serializer.DeserializeList(&in)); - snapshot_live_manifest_entries.entries_by_snapshot_[snapshot_id] = - std::make_shared>(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 index acf1d8e77..bdd79acdf 100644 --- 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> entries; }; explicit SnapshotLiveManifestEntries(int32_t max_snapshots); std::optional LatestBeforeOrEqual(int64_t snapshot_id) const; - void Put(int64_t snapshot_id, std::vector&& entries); + void Put(int64_t snapshot_id, std::vector&& entries, + int64_t total_data_files = -1); size_t Size() const; Result> Serialize(const std::shared_ptr& pool) const; @@ -59,7 +61,7 @@ class SnapshotLiveManifestEntries { private: void EvictIfNeeded(); - std::map>> entries_by_snapshot_; + std::map 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 index 908f9bbb1..c34b41f40 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 @@ -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(/*max_weight=*/16 * 1024 * 1024); + auto cache = std::make_shared( + std::map{{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 = + 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 index 45376b229..2955b127d 100644 --- a/src/paimon/core/operation/file_store_scan.cpp +++ b/src/paimon/core/operation/file_store_scan.cpp @@ -143,8 +143,8 @@ Result> FileStoreScan::CreatePlan() cons std::optional snapshot; std::vector all_manifest_file_metas; std::vector 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 manifest_entries; std::optional cache_bucket = bucket_filter_; @@ -159,9 +159,9 @@ Result> 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 filtered_entries; filtered_entries.reserve(manifest_entries.size()); @@ -173,6 +173,8 @@ Result> 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> 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(scan_duration_ms)); @@ -247,27 +251,31 @@ Result> FileStoreScan::CreatePlan() cons std::move(manifest_entries)); } -Status FileStoreScan::ReadManifests(std::optional* snapshot_ptr, - std::vector* all_manifests_ptr, - std::vector* 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> FileStoreScan::ReadSnapshot() const { + if (specified_snapshot_) { + return specified_snapshot_; } - if (snapshot == std::nullopt) { - all_manifests = std::vector(); - filtered_manifests = std::vector(); + return snapshot_manager_->LatestSnapshot(); +} + +Status FileStoreScan::ReadManifests(std::optional* snapshot, + std::vector* all_manifests, + std::vector* filtered_manifests) const { + PAIMON_ASSIGN_OR_RAISE(*snapshot, ReadSnapshot()); + return ReadManifestLists(*snapshot, all_manifests, filtered_manifests); +} + +Status FileStoreScan::ReadManifestLists(const std::optional& snapshot, + std::vector* all_manifests, + std::vector* 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& 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& all_manifest_metas, - int32_t bucket, std::vector* manifest_entries, bool* cache_hit) const { + const Snapshot& snapshot, std::vector* all_manifest_metas, + std::vector* filtered_manifest_metas, int32_t bucket, + std::vector* 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,17 +380,26 @@ Status FileStoreScan::ReadManifestEntriesWithCache( static_cast(cache_load_duration_ms)); std::optional 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 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); } @@ -389,7 +407,7 @@ Status FileStoreScan::ReadManifestEntriesWithCache( PAIMON_RETURN_NOT_OK( ReadAndMergeBucketFileEntries(bucket_manifest_metas, bucket, manifest_entries)); std::vector 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 index 0eef29096..b46d04d97 100644 --- 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& evolution); private: + Result> ReadSnapshot() const; + Status ReadManifestLists(const std::optional& snapshot, + std::vector* all_manifests, + std::vector* filtered_manifests) const; + Status ReadManifests(std::optional* snapshot_ptr, std::vector* all_manifests_ptr, std::vector* filtered_manifests_ptr) const; @@ -269,10 +274,11 @@ class FileStoreScan { std::vector* manifest_entries) const; Status ReadManifestEntriesWithCache(const Snapshot& snapshot, - const std::vector& bucket_manifest_metas, + std::vector* all_manifest_metas, + std::vector* filtered_manifest_metas, int32_t bucket, std::vector* manifest_entries, - bool* cache_hit) const; + bool* cache_hit, int64_t* all_data_files) const; std::shared_ptr CreateSnapshotLiveManifestEntriesCacheKey(int32_t bucket) const; Result 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 index c4f0289e7..49dc1a9a1 100644 --- 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),