diff --git a/docs/source/user_guide/manifest_entry_cache.rst b/docs/source/user_guide/manifest_entry_cache.rst index 0f37138c2..995d77e38 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 @@ -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. diff --git a/src/paimon/core/manifest/snapshot_live_manifest_entries.cpp b/src/paimon/core/manifest/snapshot_live_manifest_entries.cpp index c6d981e10..6f925ba15 100644 --- a/src/paimon/core/manifest/snapshot_live_manifest_entries.cpp +++ b/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(std::max(0, max_snapshots)); @@ -61,12 +61,14 @@ std::shared_ptr ToBytes(const MemorySegmentOutputStream& out, SnapshotLiveManifestEntries::SnapshotLiveManifestEntries(int32_t max_snapshots) : max_snapshots_(max_snapshots) {} -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)); + total_data_files_by_snapshot_[snapshot_id] = total_data_files; EvictIfNeeded(); } @@ -83,6 +85,7 @@ Result> SnapshotLiveManifestEntries::Serialize( ManifestEntrySerializer serializer(pool); for (const auto& [snapshot_id, entries] : entries_by_snapshot_) { out.WriteValue(snapshot_id); + out.WriteValue(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::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.total_data_files_by_snapshot_[snapshot_id] = + total_data_files; snapshot_live_manifest_entries.entries_by_snapshot_[snapshot_id] = std::make_shared>(std::move(entries)); } @@ -122,6 +128,7 @@ Result 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 previous_id; for (int32_t i = 0; i < snapshot_count; ++i) { PAIMON_ASSIGN_OR_RAISE(int64_t id, in.ReadValue()); + PAIMON_ASSIGN_OR_RAISE(int64_t total_data_files, in.ReadValue()); // 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); } diff --git a/src/paimon/core/manifest/snapshot_live_manifest_entries.h b/src/paimon/core/manifest/snapshot_live_manifest_entries.h index 784d31b28..f2d4343ad 100644 --- a/src/paimon/core/manifest/snapshot_live_manifest_entries.h +++ b/src/paimon/core/manifest/snapshot_live_manifest_entries.h @@ -45,11 +45,12 @@ class SnapshotLiveManifestEntries { struct Selection { std::vector 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&& entries); + void Put(int64_t snapshot_id, std::vector&& entries, int64_t total_data_files); size_t Size() const; Result> Serialize(const std::shared_ptr& 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> ReadSnapshot( @@ -71,6 +72,7 @@ class SnapshotLiveManifestEntries { void EvictIfNeeded(); std::map>> entries_by_snapshot_; + std::map total_data_files_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 29d19d7f3..b8dfbfced 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 @@ -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(/*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); @@ -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 = + 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(offset, -1); } MemorySegmentOutputStream out(MemorySegmentOutputStream::DEFAULT_SEGMENT_SIZE, pool); - out.WriteValue(corruption == "magic" ? 0 : 0x534d4543); + out.WriteValue(corruption == "magic" ? 0 : 0x534d4544); out.WriteValue(1); out.WriteValue(snapshot.Id()); + out.WriteValue(1); out.WriteValue(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) { diff --git a/src/paimon/core/operation/file_store_scan.cpp b/src/paimon/core/operation/file_store_scan.cpp index 4e508bab2..ca2de308a 100644 --- a/src/paimon/core/operation/file_store_scan.cpp +++ b/src/paimon/core/operation/file_store_scan.cpp @@ -158,8 +158,8 @@ Result> FileStoreScan::ReadPartitionEntries() const PAIMON_RETURN_NOT_OK( ReadManifests(&snapshot, &all_manifest_file_metas, &filtered_manifest_file_metas)); std::vector manifest_entries; - PAIMON_RETURN_NOT_OK(ReadFileEntries(filtered_manifest_file_metas, &manifest_entries, - /*apply_scan_filter=*/true)); + PAIMON_RETURN_NOT_OK(ReadFileEntries(filtered_manifest_file_metas, /*apply_scan_filter=*/true, + &manifest_entries)); std::unordered_map partitions; PAIMON_RETURN_NOT_OK(PartitionEntry::Merge(manifest_entries, &partitions)); @@ -178,11 +178,10 @@ Result> FileStoreScan::CreatePlan() cons Duration manifest_read_duration; metrics_->SetCounter(ScanMetrics::LAST_SNAPSHOT_CACHE_LOAD_DURATION, 0); metrics_->SetCounter(ScanMetrics::LAST_SNAPSHOT_CACHE_STORE_DURATION, 0); - std::optional snapshot; + PAIMON_ASSIGN_OR_RAISE(std::optional snapshot, ReadSnapshot()); + int64_t all_data_files = 0; 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)); std::vector manifest_entries; std::optional cache_bucket = bucket_filter_; @@ -197,11 +196,19 @@ Result> FileStoreScan::CreatePlan() cons uint64_t lazy_decode_scanned_rows = 0; bool snapshot_cache_hit = false; if (use_snapshot_live_manifest_cache) { - PAIMON_ASSIGN_OR_RAISE(lazy_decode_scanned_rows, - ReadManifestEntriesWithCache( - snapshot.value(), all_manifest_file_metas, cache_bucket.value(), - &manifest_entries, &snapshot_cache_hit)); + PAIMON_ASSIGN_OR_RAISE( + lazy_decode_scanned_rows, + ReadManifestEntriesWithCache(snapshot.value(), cache_bucket.value(), + &all_manifest_file_metas, &filtered_manifest_file_metas, + &manifest_entries, &snapshot_cache_hit, &all_data_files)); } else { + PAIMON_RETURN_NOT_OK( + ReadManifestLists(snapshot, &all_manifest_file_metas, &filtered_manifest_file_metas)); + 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(); + }); 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) { @@ -240,11 +247,6 @@ 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(); - }); 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)); @@ -276,27 +278,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(); @@ -343,8 +349,8 @@ Status FileStoreScan::ReadManifestsWithSnapshot(const Snapshot& snapshot, } Status FileStoreScan::ReadFileEntries(const std::vector& manifest_metas, - std::vector* manifest_entries, - bool apply_scan_filter) const { + bool apply_scan_filter, + std::vector* manifest_entries) const { std::vector>>> futures; for (const auto& meta : manifest_metas) { auto read_meta_task = [this, meta, @@ -389,8 +395,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. Result 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, int32_t bucket, std::vector* all_manifest_metas, + std::vector* filtered_manifest_metas, + std::vector* manifest_entries, bool* cache_hit, int64_t* all_data_files) const { Duration cache_load_duration; const int32_t max_snapshots = core_options_.GetScanManifestEntryCacheMaxSnapshots(); SnapshotLiveManifestEntries cached_entries(max_snapshots); @@ -413,6 +420,7 @@ Result FileStoreScan::ReadManifestEntriesWithCache( } if (selected.ok() && selected.value()) { *cache_hit = true; + *all_data_files = selected.value()->total_data_files; partition_filtered = static_cast(row_filter); candidate_entries = selected.value()->candidate_entries; *manifest_entries = std::move(selected.value()->entries); @@ -432,8 +440,16 @@ Result FileStoreScan::ReadManifestEntriesWithCache( metrics_->ObserveHistogram(ScanMetrics::SNAPSHOT_CACHE_LOAD_DURATION, static_cast(cache_load_duration_ms)); if (!*cache_hit) { + // The selected snapshot is unchanged. Exact hits skip both lists and their decoding. + 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(); + }); 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); } @@ -442,7 +458,7 @@ Result FileStoreScan::ReadManifestEntriesWithCache( ReadAndMergeBucketFileEntries(bucket_manifest_metas, bucket, manifest_entries)); candidate_entries = manifest_entries->size(); 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(); @@ -534,7 +550,7 @@ Status FileStoreScan::ReadAndMergeBucketFileEntries( std::vector unmerged_entries; std::vector entries; - PAIMON_RETURN_NOT_OK(ReadFileEntries(manifest_metas, &entries, /*apply_scan_filter=*/false)); + PAIMON_RETURN_NOT_OK(ReadFileEntries(manifest_metas, /*apply_scan_filter=*/false, &entries)); unmerged_entries.reserve(entries.size()); for (auto& entry : entries) { if (entry.Bucket() == bucket || @@ -566,14 +582,14 @@ Status FileStoreScan::ReadAndMergeFileEntries(const std::vector* merged_entries) const { std::vector unmerged_entries; PAIMON_RETURN_NOT_OK( - ReadFileEntries(manifest_metas, &unmerged_entries, /*apply_scan_filter=*/true)); + ReadFileEntries(manifest_metas, /*apply_scan_filter=*/true, &unmerged_entries)); return MergeLiveEntries(unmerged_entries, merged_entries); } Status FileStoreScan::ReadAndNoMergeFileEntries( const std::vector& manifest_metas, std::vector* manifest_entries) const { - return ReadFileEntries(manifest_metas, manifest_entries, /*apply_scan_filter=*/true); + return ReadFileEntries(manifest_metas, /*apply_scan_filter=*/true, manifest_entries); } bool FileStoreScan::MayContainBucket(const ManifestFileMeta& manifest, int32_t bucket) const { diff --git a/src/paimon/core/operation/file_store_scan.h b/src/paimon/core/operation/file_store_scan.h index 25911d0e5..df8c6d614 100644 --- a/src/paimon/core/operation/file_store_scan.h +++ b/src/paimon/core/operation/file_store_scan.h @@ -276,9 +276,13 @@ class FileStoreScan { const std::shared_ptr& data_schema) const; private: - Status ReadManifests(std::optional* snapshot_ptr, - std::vector* all_manifests_ptr, - std::vector* filtered_manifests_ptr) const; + Status ReadManifests(std::optional* snapshot, + std::vector* all_manifests, + std::vector* filtered_manifests) const; + Result> ReadSnapshot() const; + Status ReadManifestLists(const std::optional& snapshot, + std::vector* all_manifests, + std::vector* filtered_manifests) const; Status ReadManifestsWithSnapshot(const Snapshot& snapshot, std::vector* manifests) const; @@ -289,8 +293,10 @@ class FileStoreScan { /// Read request-filtered entries while caching complete bucket snapshots. /// @return Number of entries considered before request-specific filtering. Result ReadManifestEntriesWithCache( - const Snapshot& snapshot, const std::vector& bucket_manifest_metas, - int32_t bucket, std::vector* manifest_entries, bool* cache_hit) const; + const Snapshot& snapshot, int32_t bucket, std::vector* all_manifest_metas, + std::vector* filtered_manifest_metas, + std::vector* manifest_entries, bool* cache_hit, + int64_t* all_data_files) const; std::shared_ptr CreateSnapshotLiveManifestEntriesCacheKey(int32_t bucket) const; Status StoreSnapshotLiveManifestEntries(int32_t bucket, const SnapshotLiveManifestEntries& entries) const; @@ -306,8 +312,8 @@ class FileStoreScan { std::vector* manifest_entries) const; Status ReadFileEntries(const std::vector& manifest_metas, - std::vector* manifest_entries, - bool apply_scan_filter) const; + bool apply_scan_filter, + std::vector* manifest_entries) const; bool MayContainBucket(const ManifestFileMeta& manifest, int32_t bucket) const; diff --git a/src/paimon/core/operation/file_store_scan_test.cpp b/src/paimon/core/operation/file_store_scan_test.cpp index 5c53f138d..eb49f3541 100644 --- a/src/paimon/core/operation/file_store_scan_test.cpp +++ b/src/paimon/core/operation/file_store_scan_test.cpp @@ -232,8 +232,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=*/1); + entries.Put(/*snapshot_id=*/3, {}, /*total_data_files=*/0); ASSERT_OK_AND_ASSIGN(auto bytes, entries.Serialize(GetDefaultPool())); ASSERT_OK_AND_ASSIGN(auto deserialized, @@ -273,9 +273,9 @@ TEST_F(FileStoreScanTest, SelectExactRetainedSnapshot) { 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, {}); + entries.Put(1, {records[0]}, 1); + entries.Put(3, {records[1], records[2]}, 17); + entries.Put(5, {}, 0); ASSERT_OK_AND_ASSIGN(auto bytes, entries.Serialize(pool)); const MemorySegment segment = MemorySegment::Wrap(bytes); ASSERT_OK_AND_ASSIGN(auto first, @@ -300,6 +300,7 @@ TEST_F(FileStoreScanTest, SelectExactRetainedSnapshot) { pool)); ASSERT_TRUE(selected); ASSERT_EQ(selected->candidate_entries, 2); + ASSERT_EQ(selected->total_data_files, 17); ASSERT_EQ(filter_calls, 2); ASSERT_EQ(selected->entries, std::vector({records[2]})); ASSERT_OK_AND_ASSIGN(auto absent, @@ -326,7 +327,7 @@ TEST_F(FileStoreScanTest, SelectExactRetainedSnapshot) { pool), "injected filter error"); - entries.Put(7, {}); + entries.Put(7, {}, 0); 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. @@ -336,6 +337,7 @@ TEST_F(FileStoreScanTest, SelectExactRetainedSnapshot) { ASSERT_OK_AND_ASSIGN(auto retained, SnapshotLiveManifestEntries::ReadSnapshot( MemorySegment::Wrap(updated), 3, 4, nullptr, pool)); ASSERT_TRUE(retained); + ASSERT_EQ(retained->total_data_files, 17); ASSERT_EQ(retained->entries, std::vector({records[1], records[2]})); } @@ -349,7 +351,7 @@ TEST_F(FileStoreScanTest, SkipDecodingFilteredManifestEntries) { ManifestEntry excluded(FileKind(static_cast(2)), BinaryRowGenerator::GenerateRow({10}, pool.get()), 0, 1, file); SnapshotLiveManifestEntries entries(1); - entries.Put(1, {excluded}); + entries.Put(1, {excluded}, 1); ASSERT_OK_AND_ASSIGN(auto bytes, entries.Serialize(pool)); const MemorySegment segment = MemorySegment::Wrap(bytes); ASSERT_OK_AND_ASSIGN(auto selected, SnapshotLiveManifestEntries::ReadSnapshot( @@ -373,9 +375,10 @@ TEST_F(FileStoreScanTest, RejectUnsupportedSnapshotEntryVersionBeforeFiltering) BinaryRow row = BinaryRowGenerator::GenerateRow({3}, pool.get()); BinaryRowSerializer serializer(1, pool); MemorySegmentOutputStream out(64, pool); - out.WriteValue(0x534d4543); + out.WriteValue(0x534d4544); out.WriteValue(1); out.WriteValue(1); + out.WriteValue(7); out.WriteValue(1); ASSERT_OK(serializer.Serialize(row, &out)); bool filter_called = false; @@ -400,16 +403,17 @@ TEST_F(FileStoreScanTest, RejectInvalidSnapshotSelectionHeader) { out.WriteValue(magic); out.WriteValue(snapshots); out.WriteValue(1); + out.WriteValue(7); out.WriteValue(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); + segment = header(0x534d4544, -1, 0); ASSERT_NOK_WITH_MSG(SnapshotLiveManifestEntries::ReadSnapshot(segment, 1, 3, nullptr, pool), "snapshot count is negative"); - segment = header(0x534d4543, 1, -1); + segment = header(0x534d4544, 1, -1); ASSERT_NOK_WITH_MSG(SnapshotLiveManifestEntries::ReadSnapshot(segment, 1, 3, nullptr, pool), "entry count is negative"); auto truncated = MemorySegment::WrapView(segment.Data(), 7);