From af9509dcb534236fe06144887a63c62996fa5869 Mon Sep 17 00:00:00 2001 From: "wangyong.alen" Date: Mon, 31 Aug 2026 04:46:12 -0400 Subject: [PATCH 1/8] perf(scan): reuse global index plan resources --- .../source/data_evolution_batch_scan.cpp | 30 ++++++++++++------- .../table/source/data_evolution_batch_scan.h | 3 ++ src/paimon/core/table/source/table_scan.cpp | 2 +- test/inte/global_index_test.cpp | 22 +++++++++++++- 4 files changed, 45 insertions(+), 12 deletions(-) diff --git a/src/paimon/core/table/source/data_evolution_batch_scan.cpp b/src/paimon/core/table/source/data_evolution_batch_scan.cpp index ed09b693b..2f86769ad 100644 --- a/src/paimon/core/table/source/data_evolution_batch_scan.cpp +++ b/src/paimon/core/table/source/data_evolution_batch_scan.cpp @@ -25,19 +25,21 @@ #include "paimon/core/global_index/global_index_scan_impl.h" #include "paimon/core/global_index/indexed_split_impl.h" #include "paimon/core/table/source/data_split_impl.h" +#include "paimon/core/utils/snapshot_manager.h" #include "paimon/global_index/bitmap_global_index_result.h" -#include "paimon/global_index/global_index_scan.h" namespace paimon { DataEvolutionBatchScan::DataEvolutionBatchScan( const std::string& table_path, const std::shared_ptr& snapshot_reader, std::unique_ptr&& batch_scan, + const std::shared_ptr& table_schema, const std::shared_ptr& global_index_result, const CoreOptions& core_options, const std::shared_ptr& pool, const std::shared_ptr& executor) : AbstractTableScan(core_options, snapshot_reader), pool_(pool), table_path_(table_path), batch_scan_(std::move(batch_scan)), + table_schema_(table_schema), global_index_result_(global_index_result), executor_(executor) {} @@ -144,17 +146,25 @@ Result> DataEvolutionBatchScan::EvalGlobalInd } auto partition_filter = batch_scan_->GetPartitionPredicate(); // TODO(lisizhuo.lsz): support time travel - PAIMON_ASSIGN_OR_RAISE( - std::unique_ptr index_scan, - GlobalIndexScan::Create(table_path_, core_options_.GetScanSnapshotId(), partition_filter, - core_options_.ToMap(), core_options_.GetFileSystem(), executor_, - pool_)); - auto index_scan_impl = dynamic_cast(index_scan.get()); - if (!index_scan_impl) { - return Status::Invalid("invalid GlobalIndexScan, cannot cast to GlobalIndexScanImpl"); + std::optional snapshot; + const std::shared_ptr& snapshot_manager = + snapshot_reader_->GetSnapshotManager(); + if (const std::optional& snapshot_id = core_options_.GetScanSnapshotId()) { + PAIMON_ASSIGN_OR_RAISE(Snapshot loaded_snapshot, + snapshot_manager->LoadSnapshot(snapshot_id.value())); + snapshot = std::move(loaded_snapshot); + } else { + PAIMON_ASSIGN_OR_RAISE(snapshot, snapshot_manager->LatestSnapshot()); + } + if (!snapshot) { + return Status::Invalid("not found latest snapshot"); } - return index_scan_impl->Scan(predicate); + PAIMON_ASSIGN_OR_RAISE( + std::unique_ptr index_scan, + GlobalIndexScanImpl::Create(table_path_, table_schema_, snapshot.value(), partition_filter, + core_options_, executor_, pool_)); + return index_scan->Scan(predicate); } } // namespace paimon diff --git a/src/paimon/core/table/source/data_evolution_batch_scan.h b/src/paimon/core/table/source/data_evolution_batch_scan.h index cfa297857..546ae1011 100644 --- a/src/paimon/core/table/source/data_evolution_batch_scan.h +++ b/src/paimon/core/table/source/data_evolution_batch_scan.h @@ -24,6 +24,7 @@ #include #include +#include "paimon/core/schema/table_schema.h" #include "paimon/core/table/source/abstract_table_scan.h" #include "paimon/core/table/source/data_table_batch_scan.h" #include "paimon/result.h" @@ -35,6 +36,7 @@ class DataEvolutionBatchScan : public AbstractTableScan { DataEvolutionBatchScan(const std::string& table_path, const std::shared_ptr& snapshot_reader, std::unique_ptr&& batch_scan, + const std::shared_ptr& table_schema, const std::shared_ptr& global_index_result, const CoreOptions& core_options, const std::shared_ptr& pool, const std::shared_ptr& executor); @@ -55,6 +57,7 @@ class DataEvolutionBatchScan : public AbstractTableScan { std::shared_ptr pool_; std::string table_path_; std::unique_ptr batch_scan_; + std::shared_ptr table_schema_; std::shared_ptr global_index_result_; std::shared_ptr executor_; }; diff --git a/src/paimon/core/table/source/table_scan.cpp b/src/paimon/core/table/source/table_scan.cpp index 95af2a23b..e4db33fb4 100644 --- a/src/paimon/core/table/source/table_scan.cpp +++ b/src/paimon/core/table/source/table_scan.cpp @@ -361,7 +361,7 @@ Result> NewDataTableScan(const std::shared_ptr( - context->GetPath(), snapshot_reader, std::move(batch_scan), + context->GetPath(), snapshot_reader, std::move(batch_scan), table_schema, context->GetGlobalIndexResult(), core_options, context->GetMemoryPool(), context->GetExecutor()); } diff --git a/test/inte/global_index_test.cpp b/test/inte/global_index_test.cpp index 84ab22df8..91b559e9f 100644 --- a/test/inte/global_index_test.cpp +++ b/test/inte/global_index_test.cpp @@ -39,6 +39,7 @@ #include "paimon/result.h" #include "paimon/status.h" #include "paimon/testing/utils/binary_row_generator.h" +#include "paimon/testing/utils/counting_cache_test_utils.h" #include "paimon/testing/utils/io_exception_helper.h" #include "paimon/testing/utils/test_helper.h" #include "paimon/testing/utils/testharness.h" @@ -167,12 +168,16 @@ class GlobalIndexTest : public ::testing::Test, public ::testing::WithParamInter Result> ScanGlobalIndexAndData( const std::string& table_path, const std::shared_ptr& predicate, const std::map& options = {}, - const std::shared_ptr& index_result = nullptr) const { + const std::shared_ptr& index_result = nullptr, + const std::shared_ptr& cache = nullptr) const { ScanContextBuilder scan_context_builder(table_path); scan_context_builder.SetPredicate(predicate) .SetOptions(options) .SetGlobalIndexResult(index_result) .WithFileSystem(fs_); + if (cache) { + scan_context_builder.WithCache(cache); + } PAIMON_ASSIGN_OR_RAISE(auto scan_context, scan_context_builder.Finish()); PAIMON_ASSIGN_OR_RAISE(auto table_scan, TableScan::Create(std::move(scan_context))); PAIMON_ASSIGN_OR_RAISE(auto result_plan, table_scan->CreatePlan()); @@ -1422,6 +1427,21 @@ TEST_P(GlobalIndexTest, TestDataEvolutionBatchScan) { ASSERT_OK(WriteIndex(table_path, /*partition_filters=*/{}, "f0", "bitmap", /*options=*/{}, Range(0, 7))); + { + auto cache = std::make_shared(CacheKind::MANIFEST, 64 * 1024 * 1024); + auto predicate = + PredicateBuilder::Equal(/*field_index=*/0, /*field_name=*/"f0", FieldType::STRING, + Literal(FieldType::STRING, "Alice", 5)); + ASSERT_OK(ScanGlobalIndexAndData(table_path, predicate, /*options=*/{}, + /*index_result=*/nullptr, cache)); + ASSERT_GE(cache->GetCount(CacheKind::MANIFEST), 2); + int64_t first_supplier_calls = cache->SupplierCallCount(CacheKind::MANIFEST); + + ASSERT_OK(ScanGlobalIndexAndData(table_path, predicate, /*options=*/{}, + /*index_result=*/nullptr, cache)); + ASSERT_EQ(cache->SupplierCallCount(CacheKind::MANIFEST), first_supplier_calls); + } + // scan and read with global index { auto predicate = From 5e0147b9cc87c198b34235c4043c18417dac9d15 Mon Sep 17 00:00:00 2001 From: "wangyong.alen" Date: Mon, 31 Aug 2026 05:43:18 -0400 Subject: [PATCH 2/8] perf(scan): cache whole-snapshot manifest entries --- include/paimon/cache/cache.h | 2 + src/paimon/common/io/cache/cache_key.cpp | 17 +++++-- src/paimon/common/io/cache/lru_cache_test.cpp | 2 + .../append_only_file_store_scan_test.cpp | 29 ++++++++++++ src/paimon/core/operation/file_store_scan.cpp | 47 +++++++++++-------- src/paimon/core/operation/file_store_scan.h | 12 +++-- 6 files changed, 81 insertions(+), 28 deletions(-) diff --git a/include/paimon/cache/cache.h b/include/paimon/cache/cache.h index 5bff2a120..e5e5c67fd 100644 --- a/include/paimon/cache/cache.h +++ b/include/paimon/cache/cache.h @@ -44,6 +44,8 @@ class PAIMON_EXPORT CacheKey { int32_t length, bool is_index); static std::shared_ptr ForKind(const std::string& file_path, int64_t position, int32_t length, CacheKind kind); + static std::shared_ptr ForSnapshotLiveManifestEntries(const std::string& table_path, + const std::string& branch); static std::shared_ptr ForSnapshotLiveManifestEntries(const std::string& table_path, const std::string& branch, int32_t bucket); diff --git a/src/paimon/common/io/cache/cache_key.cpp b/src/paimon/common/io/cache/cache_key.cpp index c84c8f908..ae23eb303 100644 --- a/src/paimon/common/io/cache/cache_key.cpp +++ b/src/paimon/common/io/cache/cache_key.cpp @@ -18,13 +18,15 @@ #include "paimon/common/io/cache/cache_key.h" +#include + namespace paimon { namespace { class SnapshotLiveManifestEntriesCacheKey : public CacheKey { public: SnapshotLiveManifestEntriesCacheKey(const std::string& table_path, const std::string& branch, - int32_t bucket) + const std::optional& bucket) : CacheKey(CacheKind::SNAPSHOT_LIVE_MANIFEST), table_path_(table_path), branch_(branch), @@ -47,7 +49,11 @@ class SnapshotLiveManifestEntriesCacheKey : public CacheKey { size_t seed = 0; seed ^= std::hash{}(table_path_) + HASH_CONSTANT + (seed << 6) + (seed >> 2); seed ^= std::hash{}(branch_) + HASH_CONSTANT + (seed << 6) + (seed >> 2); - seed ^= std::hash{}(bucket_) + HASH_CONSTANT + (seed << 6) + (seed >> 2); + seed ^= std::hash{}(bucket_.has_value()) + HASH_CONSTANT + (seed << 6) + (seed >> 2); + if (bucket_) { + seed ^= + std::hash{}(bucket_.value()) + HASH_CONSTANT + (seed << 6) + (seed >> 2); + } seed ^= std::hash{}(static_cast(GetKind())) + HASH_CONSTANT + (seed << 6) + (seed >> 2); return seed; @@ -58,7 +64,7 @@ class SnapshotLiveManifestEntriesCacheKey : public CacheKey { const std::string table_path_; const std::string branch_; - const int32_t bucket_; + const std::optional bucket_; }; } // namespace @@ -76,6 +82,11 @@ std::shared_ptr CacheKey::ForKind(const std::string& file_path, int64_ return key; } +std::shared_ptr CacheKey::ForSnapshotLiveManifestEntries(const std::string& table_path, + const std::string& branch) { + return std::make_shared(table_path, branch, std::nullopt); +} + std::shared_ptr CacheKey::ForSnapshotLiveManifestEntries(const std::string& table_path, const std::string& branch, int32_t bucket) { diff --git a/src/paimon/common/io/cache/lru_cache_test.cpp b/src/paimon/common/io/cache/lru_cache_test.cpp index 1d644c70e..bae436228 100644 --- a/src/paimon/common/io/cache/lru_cache_test.cpp +++ b/src/paimon/common/io/cache/lru_cache_test.cpp @@ -391,12 +391,14 @@ TEST_F(LruCacheTest, TestForSnapshotLiveManifestEntries) { auto bucket_key = CacheKey::ForSnapshotLiveManifestEntries("table_path", "main", 1); auto hash_in_path_key = CacheKey::ForSnapshotLiveManifestEntries("table#path", "main", 0); auto hash_in_branch_key = CacheKey::ForSnapshotLiveManifestEntries("table", "path#main", 0); + auto all_buckets_key = CacheKey::ForSnapshotLiveManifestEntries("table_path", "main"); ASSERT_EQ(CacheKind::SNAPSHOT_LIVE_MANIFEST, main_key->GetKind()); ASSERT_TRUE(CacheKeyEqual()(main_key, same_key)); ASSERT_FALSE(CacheKeyEqual()(main_key, branch_key)); ASSERT_FALSE(CacheKeyEqual()(main_key, table_key)); ASSERT_FALSE(CacheKeyEqual()(main_key, bucket_key)); + ASSERT_FALSE(CacheKeyEqual()(main_key, all_buckets_key)); ASSERT_FALSE(CacheKeyEqual()(hash_in_path_key, hash_in_branch_key)); } 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 d72dcaf0d..f5865a0c2 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 @@ -317,6 +317,35 @@ TEST(AppendOnlyFileStoreScanTest, TestSnapshotLiveManifestCachePath) { ASSERT_OK(second_metrics->GetHistogramStats(ScanMetrics::SNAPSHOT_CACHE_LOAD_DURATION)); } +TEST(AppendOnlyFileStoreScanTest, TestSnapshotLiveManifestCacheWithoutBucketFilter) { + 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 scan_first = BuildScan(table_path, cache); + ASSERT_OK_AND_ASSIGN(Snapshot snapshot_5, + scan_first->GetSnapshotManager()->LoadSnapshot(/*snapshot_id=*/5)); + scan_first->WithSnapshot(snapshot_5); + ASSERT_OK_AND_ASSIGN(auto plan_first, scan_first->CreatePlan()); + std::vector first_file_names = SortedFileNames(plan_first->Files()); + std::shared_ptr first_metrics = scan_first->GetScanMetrics(); + ASSERT_OK_AND_ASSIGN(uint64_t first_cache_enabled, + first_metrics->GetCounter(ScanMetrics::LAST_SNAPSHOT_CACHE_ENABLED)); + ASSERT_OK_AND_ASSIGN(uint64_t first_cache_hit, + first_metrics->GetCounter(ScanMetrics::LAST_SNAPSHOT_CACHE_HIT)); + ASSERT_EQ(first_cache_enabled, 1); + ASSERT_EQ(first_cache_hit, 0); + + auto scan_second = BuildScan(table_path, cache); + scan_second->WithSnapshot(snapshot_5); + ASSERT_OK_AND_ASSIGN(auto plan_second, scan_second->CreatePlan()); + ASSERT_EQ(first_file_names, SortedFileNames(plan_second->Files())); + std::shared_ptr second_metrics = scan_second->GetScanMetrics(); + ASSERT_OK_AND_ASSIGN(uint64_t second_cache_hit, + second_metrics->GetCounter(ScanMetrics::LAST_SNAPSHOT_CACHE_HIT)); + ASSERT_EQ(second_cache_hit, 1); +} + TEST(AppendOnlyFileStoreScanTest, TestSnapshotLiveManifestCacheRebuildOnMiss) { TimezoneGuard guard("Asia/Shanghai"); std::string table_path = paimon::test::GetDataDir() + "/orc/append_09.db/append_09/"; diff --git a/src/paimon/core/operation/file_store_scan.cpp b/src/paimon/core/operation/file_store_scan.cpp index befe7eb97..f65430dbf 100644 --- a/src/paimon/core/operation/file_store_scan.cpp +++ b/src/paimon/core/operation/file_store_scan.cpp @@ -150,12 +150,12 @@ Result> FileStoreScan::CreatePlan() cons snapshot.has_value() && scan_mode_ == ScanMode::ALL && core_options_.GetScanManifestEntryCacheMaxSnapshots() > 0 && core_options_.GetCache() != nullptr && !table_path_.empty() && - !row_range_index_.has_value() && bucket_filter_.has_value(); + !row_range_index_.has_value(); 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, - bucket_filter_.value(), &manifest_entries, + bucket_filter_, &manifest_entries, &snapshot_cache_hit)); lazy_decode_scanned_rows = manifest_entries.size(); std::vector filtered_entries; @@ -324,13 +324,13 @@ Status FileStoreScan::ReadManifestEntries(const std::vector& m return ReadAndNoMergeFileEntries(manifest_metas, manifest_entries); } -// Cache merged live manifest entries for one bucket before applying scan filters. Each cache value -// keeps a bounded number of snapshot results for the same table/branch/bucket. Exact snapshot hits -// can be returned directly; cache misses rebuild the target snapshot bucket from the target -// snapshot's data manifests. +// Cache merged live manifest entries before applying scan filters. Each cache value keeps a +// bounded number of snapshot results for the same table/branch and optional bucket. Exact snapshot +// hits can be returned directly; cache misses rebuild the target snapshot from its 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 std::optional& bucket, std::vector* manifest_entries, + bool* cache_hit) const { Duration cache_load_duration; PAIMON_ASSIGN_OR_RAISE(SnapshotLiveManifestEntries cached_entries, LoadSnapshotLiveManifestEntries(bucket)); @@ -347,16 +347,21 @@ Status FileStoreScan::ReadManifestEntriesWithCache( } *cache_hit = false; - // 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) { - if (MayContainBucket(meta, bucket)) { - bucket_manifest_metas.push_back(meta); + if (bucket) { + std::vector bucket_manifest_metas; + for (const auto& meta : all_manifest_metas) { + if (MayContainBucket(meta, bucket.value())) { + bucket_manifest_metas.push_back(meta); + } } + PAIMON_RETURN_NOT_OK( + ReadAndMergeBucketFileEntries(bucket_manifest_metas, bucket.value(), manifest_entries)); + } else { + std::vector unmerged_entries; + PAIMON_RETURN_NOT_OK( + ReadFileEntries(all_manifest_metas, &unmerged_entries, /*apply_scan_filter=*/false)); + PAIMON_RETURN_NOT_OK(MergeLiveEntries(unmerged_entries, manifest_entries)); } - 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)); Duration cache_store_duration; @@ -368,13 +373,15 @@ Status FileStoreScan::ReadManifestEntriesWithCache( return Status::OK(); } -std::shared_ptr FileStoreScan::SnapshotLiveManifestEntriesCacheKey(int32_t bucket) const { - return CacheKey::ForSnapshotLiveManifestEntries( - table_path_, BranchManager::NormalizeBranch(core_options_.GetBranch()), bucket); +std::shared_ptr FileStoreScan::SnapshotLiveManifestEntriesCacheKey( + const std::optional& bucket) const { + const std::string branch = BranchManager::NormalizeBranch(core_options_.GetBranch()); + return bucket ? CacheKey::ForSnapshotLiveManifestEntries(table_path_, branch, bucket.value()) + : CacheKey::ForSnapshotLiveManifestEntries(table_path_, branch); } Result FileStoreScan::LoadSnapshotLiveManifestEntries( - int32_t bucket) const { + const std::optional& bucket) const { auto supplier = [](const std::shared_ptr&) -> Result> { return std::shared_ptr(); }; @@ -394,7 +401,7 @@ Result FileStoreScan::LoadSnapshotLiveManifestEntri } Status FileStoreScan::StoreSnapshotLiveManifestEntries( - int32_t bucket, const SnapshotLiveManifestEntries& entries) const { + const std::optional& bucket, const SnapshotLiveManifestEntries& entries) const { Result> bytes_result = entries.Serialize(pool_); if (!bytes_result.ok()) { return Status::OK(); diff --git a/src/paimon/core/operation/file_store_scan.h b/src/paimon/core/operation/file_store_scan.h index ff53e0c9e..962637341 100644 --- a/src/paimon/core/operation/file_store_scan.h +++ b/src/paimon/core/operation/file_store_scan.h @@ -260,13 +260,15 @@ class FileStoreScan { std::vector* manifest_entries) const; Status ReadManifestEntriesWithCache(const Snapshot& snapshot, - const std::vector& bucket_manifest_metas, - int32_t bucket, + const std::vector& all_manifest_metas, + const std::optional& bucket, std::vector* manifest_entries, bool* cache_hit) const; - std::shared_ptr SnapshotLiveManifestEntriesCacheKey(int32_t bucket) const; - Result LoadSnapshotLiveManifestEntries(int32_t bucket) const; - Status StoreSnapshotLiveManifestEntries(int32_t bucket, + std::shared_ptr SnapshotLiveManifestEntriesCacheKey( + const std::optional& bucket) const; + Result LoadSnapshotLiveManifestEntries( + const std::optional& bucket) const; + Status StoreSnapshotLiveManifestEntries(const std::optional& bucket, const SnapshotLiveManifestEntries& entries) const; Status ReadAndMergeBucketFileEntries(const std::vector& manifest_metas, From 43268ddcff4aeeaa8a641495407b5f39b7fcbd9e Mon Sep 17 00:00:00 2001 From: "wangyong.alen" Date: Mon, 31 Aug 2026 06:17:53 -0400 Subject: [PATCH 3/8] perf(index): reuse btree blocks across queries --- .../btree/btree_global_indexer.cpp | 27 +++++++- .../btree/lazy_filtered_btree_reader.cpp | 24 ++++--- .../btree/lazy_filtered_btree_reader_test.cpp | 34 +++++++++ src/paimon/common/sst/block_cache.h | 69 +++++++++++++++---- src/paimon/common/sst/block_cache_test.cpp | 40 +++++++++++ 5 files changed, 169 insertions(+), 25 deletions(-) diff --git a/src/paimon/common/global_index/btree/btree_global_indexer.cpp b/src/paimon/common/global_index/btree/btree_global_indexer.cpp index 8996fe59d..277856eab 100644 --- a/src/paimon/common/global_index/btree/btree_global_indexer.cpp +++ b/src/paimon/common/global_index/btree/btree_global_indexer.cpp @@ -19,8 +19,11 @@ #include "paimon/common/global_index/btree/btree_global_indexer.h" #include +#include #include +#include #include +#include #include "arrow/c/bridge.h" #include "fmt/format.h" @@ -45,6 +48,27 @@ #include "paimon/utils/roaring_bitmap64.h" namespace paimon { +namespace { + +std::shared_ptr GetSharedCacheManager(int64_t cache_size, + double high_priority_pool_ratio) { + using CacheConfig = std::pair; + static std::mutex mutex; + static std::map> cache_managers; + + std::lock_guard lock(mutex); + CacheConfig config(cache_size, high_priority_pool_ratio); + auto iter = cache_managers.find(config); + if (iter != cache_managers.end()) { + return iter->second; + } + auto cache_manager = std::make_shared(cache_size, high_priority_pool_ratio); + cache_managers.emplace(config, cache_manager); + return cache_manager; +} + +} // namespace + Result> BTreeGlobalIndexer::Create( const std::map& options) { // parse cache options @@ -57,7 +81,8 @@ Result> BTreeGlobalIndexer::Create( double high_priority_pool_ratio, OptionsUtils::GetValueFromMap(options, BtreeDefs::kBtreeIndexHighPriorityPoolRatio, BtreeDefs::kDefaultBtreeIndexHighPriorityPoolRatio)); - auto cache_manager = std::make_shared(cache_size, high_priority_pool_ratio); + std::shared_ptr cache_manager = + GetSharedCacheManager(cache_size, high_priority_pool_ratio); return std::unique_ptr(new BTreeGlobalIndexer(cache_manager, options)); } diff --git a/src/paimon/common/global_index/btree/lazy_filtered_btree_reader.cpp b/src/paimon/common/global_index/btree/lazy_filtered_btree_reader.cpp index 4ea7576b6..b2b51159a 100644 --- a/src/paimon/common/global_index/btree/lazy_filtered_btree_reader.cpp +++ b/src/paimon/common/global_index/btree/lazy_filtered_btree_reader.cpp @@ -238,16 +238,20 @@ Result> LazyFilteredBTreeReader::CreateSingle max_key_slice = MemorySlice::Wrap(index_meta->LastKey()); } - // Open input stream and create block cache - PAIMON_ASSIGN_OR_RAISE(std::shared_ptr input_stream, - file_reader_->GetInputStream(meta.file_path)); - if (read_buffer_size_) { - input_stream = std::make_shared( - input_stream, read_buffer_size_.value(), pool_.get()); - } - - auto block_cache = - std::make_shared(meta.file_path, input_stream, cache_manager_, pool_); + BlockCache::InputStreamSupplier input_stream_supplier = + [file_reader = file_reader_, file_path = meta.file_path, + read_buffer_size = read_buffer_size_, + pool = pool_]() -> Result> { + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr input_stream, + file_reader->GetInputStream(file_path)); + if (read_buffer_size) { + input_stream = std::make_shared( + input_stream, read_buffer_size.value(), pool.get()); + } + return input_stream; + }; + auto block_cache = std::make_shared( + meta.file_path, std::move(input_stream_supplier), cache_manager_, pool_); // Read footer PAIMON_ASSIGN_OR_RAISE(MemorySegment footer_segment, diff --git a/src/paimon/common/global_index/btree/lazy_filtered_btree_reader_test.cpp b/src/paimon/common/global_index/btree/lazy_filtered_btree_reader_test.cpp index a1be0108b..c6d2abcae 100644 --- a/src/paimon/common/global_index/btree/lazy_filtered_btree_reader_test.cpp +++ b/src/paimon/common/global_index/btree/lazy_filtered_btree_reader_test.cpp @@ -84,6 +84,13 @@ class FakeLazyFileReader : public GlobalIndexFileReader { std::string base_path_; }; +class FailingLazyFileReader : public GlobalIndexFileReader { + public: + Result> GetInputStream(const std::string&) const override { + return Status::Invalid("unexpected input stream open for cached btree file"); + } +}; + class LazyFilteredBTreeReaderTest : public ::testing::Test { public: void SetUp() override { @@ -377,6 +384,33 @@ TEST_F(LazyFilteredBTreeReaderTest, TestReaderCacheReuse) { CheckResult(result2, {3, 4}); } +TEST_F(LazyFilteredBTreeReaderTest, TestBlockCacheReuseAcrossIndexerInstances) { + std::map options = { + {BtreeDefs::kBtreeIndexCacheSize, "1MB"}, + {BtreeDefs::kBtreeIndexHighPriorityPoolRatio, "0.5"}}; + Literal literal_1(1); + + ASSERT_OK_AND_ASSIGN(auto first_indexer, BTreeGlobalIndexer::Create(options)); + auto first_schema = CreateArrowSchema(); + auto first_file_reader = std::make_shared(fs_, base_path_); + ASSERT_OK_AND_ASSIGN( + auto first_reader, + first_indexer->CreateReader(first_schema.get(), first_file_reader, all_metas_, pool_)); + ASSERT_OK_AND_ASSIGN(auto first_result, first_reader->VisitEqual(literal_1)); + CheckResult(first_result, {0, 1}); + first_reader.reset(); + first_indexer.reset(); + + ASSERT_OK_AND_ASSIGN(auto second_indexer, BTreeGlobalIndexer::Create(options)); + auto second_schema = CreateArrowSchema(); + auto failing_file_reader = std::make_shared(); + ASSERT_OK_AND_ASSIGN( + auto second_reader, + second_indexer->CreateReader(second_schema.get(), failing_file_reader, all_metas_, pool_)); + ASSERT_OK_AND_ASSIGN(auto second_result, second_reader->VisitEqual(literal_1)); + CheckResult(second_result, {0, 1}); +} + // --- Empty files list --- TEST_F(LazyFilteredBTreeReaderTest, TestEmptyFilesList) { diff --git a/src/paimon/common/sst/block_cache.h b/src/paimon/common/sst/block_cache.h index 8e6a86319..ea0b8ae10 100644 --- a/src/paimon/common/sst/block_cache.h +++ b/src/paimon/common/sst/block_cache.h @@ -19,6 +19,7 @@ #pragma once #include +#include #include #include @@ -31,11 +32,25 @@ namespace paimon { class PAIMON_EXPORT BlockCache { public: + using InputStreamSupplier = std::function>()>; + BlockCache(const std::string& file_path, const std::shared_ptr& in, const std::shared_ptr& cache_manager, const std::shared_ptr& pool) : pool_(pool), file_path_(file_path), in_(in), cache_manager_(cache_manager) {} + /// Creates a block cache whose pages survive this reader. The input stream is opened only on a + /// cache miss, so a later reader can reuse cached pages without reopening the immutable file. + BlockCache(const std::string& file_path, InputStreamSupplier input_stream_supplier, + const std::shared_ptr& cache_manager, + const std::shared_ptr& pool) + : pool_(pool), + cache_pool_(GetDefaultPool()), + file_path_(file_path), + input_stream_supplier_(std::move(input_stream_supplier)), + cache_manager_(cache_manager), + retain_cached_pages_(true) {} + ~BlockCache() { Close(); } @@ -44,23 +59,33 @@ class PAIMON_EXPORT BlockCache { int64_t position, int32_t length, bool is_index, std::function(const MemorySegment&)> decompress_func) { auto key = CacheKey::ForPosition(file_path_, position, length, is_index); + auto reader = [&](const std::shared_ptr&) -> Result { + PAIMON_ASSIGN_OR_RAISE(MemorySegment compress_data, ReadFrom(position, length)); + if (!decompress_func) { + return compress_data; + } + PAIMON_ASSIGN_OR_RAISE(MemorySegment decompressed_data, decompress_func(compress_data)); + if (!retain_cached_pages_ || (decompressed_data.Data() == compress_data.Data() && + decompressed_data.Size() == compress_data.Size())) { + return decompressed_data; + } + auto cached_data = + MemorySegment::AllocateHeapMemory(decompressed_data.Size(), cache_pool_.get()); + decompressed_data.CopyTo(/*offset=*/0, &cached_data, /*target_offset=*/0, + decompressed_data.Size()); + return cached_data; + }; + if (retain_cached_pages_) { + return cache_manager_->GetPage(key, reader, /*eviction_callback=*/{}); + } auto it = blocks_.find(key); if (it == blocks_.end() || it->second.GetAccessCount() == CacheManager::REFRESH_COUNT) { PAIMON_ASSIGN_OR_RAISE( MemorySegment segment, - cache_manager_->GetPage( - key, - [&](const std::shared_ptr&) -> Result { - PAIMON_ASSIGN_OR_RAISE(MemorySegment compress_data, - ReadFrom(position, length)); - if (!decompress_func) { - return compress_data; - } - return decompress_func(compress_data); - }, - [this](const std::shared_ptr& evicted_key) { - blocks_.erase(evicted_key); - })); + cache_manager_->GetPage(key, reader, + [this](const std::shared_ptr& evicted_key) { + blocks_.erase(evicted_key); + })); auto container = CacheManager::SegmentContainer(segment); const auto& result_segment = container.Access(); blocks_.insert_or_assign(key, container); @@ -81,6 +106,9 @@ class PAIMON_EXPORT BlockCache { } void Close() { + if (retain_cached_pages_) { + return; + } // Snapshot blocks_ to avoid iterator invalidation from `InvalidPage` callback. auto copied_blocks = blocks_; for (const auto& [key, _] : copied_blocks) { @@ -93,18 +121,31 @@ class PAIMON_EXPORT BlockCache { private: Result ReadFrom(int64_t offset, int32_t length) { + if (!in_) { + if (!input_stream_supplier_) { + return Status::Invalid("input stream is not available"); + } + PAIMON_ASSIGN_OR_RAISE(in_, input_stream_supplier_()); + if (!in_) { + return Status::Invalid("input stream supplier returned null"); + } + } PAIMON_RETURN_NOT_OK(in_->Seek(offset, SeekOrigin::FS_SEEK_SET)); - auto segment = MemorySegment::AllocateHeapMemory(length, pool_.get()); + MemoryPool* allocation_pool = retain_cached_pages_ ? cache_pool_.get() : pool_.get(); + auto segment = MemorySegment::AllocateHeapMemory(length, allocation_pool); PAIMON_RETURN_NOT_OK(in_->Read(segment.MutableData(), length)); return segment; } private: std::shared_ptr pool_; + std::shared_ptr cache_pool_; std::string file_path_; std::shared_ptr in_; + InputStreamSupplier input_stream_supplier_; std::shared_ptr cache_manager_; + bool retain_cached_pages_ = false; std::unordered_map, CacheManager::SegmentContainer, CacheKeyHash, CacheKeyEqual> blocks_; diff --git a/src/paimon/common/sst/block_cache_test.cpp b/src/paimon/common/sst/block_cache_test.cpp index 7417d91e1..5c4323923 100644 --- a/src/paimon/common/sst/block_cache_test.cpp +++ b/src/paimon/common/sst/block_cache_test.cpp @@ -65,6 +65,10 @@ class BlockCacheTest : public ::testing::Test { is_index); } + Result> OpenTestFile(const std::string& path) const { + return fs_->Open(path); + } + private: std::unique_ptr dir_; std::shared_ptr fs_; @@ -189,6 +193,42 @@ TEST_F(BlockCacheTest, TestClose) { ASSERT_EQ(cache_manager->DataCache()->Size(), 0); } +TEST_F(BlockCacheTest, TestRetainedPagesAreReusedByLaterReader) { + const int32_t block_size = 64; + auto file_path = dir_->Str() + "/retained.data"; + ASSERT_OK(WriteTestFile(file_path, /*num_blocks=*/1, block_size)); + + auto cache_manager = std::make_shared(block_size * 2, 0.0); + std::shared_ptr retained_pool = GetMemoryPool(); + std::weak_ptr retained_pool_ref = retained_pool; + int32_t input_stream_opens = 0; + BlockCache::InputStreamSupplier supplier = + [this, &input_stream_opens, file_path]() -> Result> { + input_stream_opens++; + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr input_stream, OpenTestFile(file_path)); + return input_stream; + }; + + { + BlockCache first_reader(file_path, supplier, cache_manager, retained_pool); + ASSERT_OK_AND_ASSIGN(auto segment, GetBlock(0, block_size, &first_reader)); + ASSERT_EQ(segment.Get(0), static_cast(0)); + ASSERT_EQ(input_stream_opens, 1); + } + ASSERT_EQ(cache_manager->DataCache()->Size(), 1); + + retained_pool.reset(); + ASSERT_TRUE(retained_pool_ref.expired()); + { + BlockCache second_reader(file_path, supplier, cache_manager, pool_); + ASSERT_OK_AND_ASSIGN(auto cached_segment, GetBlock(0, block_size, &second_reader)); + ASSERT_EQ(cached_segment.Get(0), static_cast(0)); + ASSERT_EQ(input_stream_opens, 1); + } + + cache_manager->DataCache()->InvalidateAll(); +} + /// Verifies that two BlockCache instances sharing the same CacheManager have independent blocks_ /// maps, but eviction in the shared LRU only affects the owning BlockCache's blocks_. TEST_F(BlockCacheTest, TestSharedCacheManagerEvictionIsolation) { From a55b08e404be45277f6b5e66295d38f7d64ab47d Mon Sep 17 00:00:00 2001 From: "wangyong.alen" Date: Mon, 31 Aug 2026 06:39:53 -0400 Subject: [PATCH 4/8] fix(index): retain shared cache allocator lifetime --- .../btree/btree_global_indexer.cpp | 20 ++++++++++++++----- .../btree/lazy_filtered_btree_reader_test.cpp | 16 +++++++++------ src/paimon/common/sst/block_cache_test.cpp | 4 ++-- .../append_only_file_store_scan_test.cpp | 6 ++++-- 4 files changed, 31 insertions(+), 15 deletions(-) diff --git a/src/paimon/common/global_index/btree/btree_global_indexer.cpp b/src/paimon/common/global_index/btree/btree_global_indexer.cpp index 277856eab..b5312634f 100644 --- a/src/paimon/common/global_index/btree/btree_global_indexer.cpp +++ b/src/paimon/common/global_index/btree/btree_global_indexer.cpp @@ -45,26 +45,36 @@ #include "paimon/executor.h" #include "paimon/global_index/bitmap_global_index_result.h" #include "paimon/memory/bytes.h" +#include "paimon/memory/memory_pool.h" #include "paimon/utils/roaring_bitmap64.h" namespace paimon { namespace { +struct SharedCacheManager { + SharedCacheManager(int64_t cache_size, double high_priority_pool_ratio) + : cache_pool(GetDefaultPool()), + cache_manager(std::make_shared(cache_size, high_priority_pool_ratio)) {} + + // Keep the allocator alive until after cache_manager releases all cached pages. + std::shared_ptr cache_pool; + std::shared_ptr cache_manager; +}; + std::shared_ptr GetSharedCacheManager(int64_t cache_size, double high_priority_pool_ratio) { using CacheConfig = std::pair; static std::mutex mutex; - static std::map> cache_managers; + static std::map cache_managers; std::lock_guard lock(mutex); CacheConfig config(cache_size, high_priority_pool_ratio); auto iter = cache_managers.find(config); if (iter != cache_managers.end()) { - return iter->second; + return iter->second.cache_manager; } - auto cache_manager = std::make_shared(cache_size, high_priority_pool_ratio); - cache_managers.emplace(config, cache_manager); - return cache_manager; + return cache_managers.emplace(config, SharedCacheManager(cache_size, high_priority_pool_ratio)) + .first->second.cache_manager; } } // namespace diff --git a/src/paimon/common/global_index/btree/lazy_filtered_btree_reader_test.cpp b/src/paimon/common/global_index/btree/lazy_filtered_btree_reader_test.cpp index c6d2abcae..ca48df079 100644 --- a/src/paimon/common/global_index/btree/lazy_filtered_btree_reader_test.cpp +++ b/src/paimon/common/global_index/btree/lazy_filtered_btree_reader_test.cpp @@ -390,24 +390,28 @@ TEST_F(LazyFilteredBTreeReaderTest, TestBlockCacheReuseAcrossIndexerInstances) { {BtreeDefs::kBtreeIndexHighPriorityPoolRatio, "0.5"}}; Literal literal_1(1); - ASSERT_OK_AND_ASSIGN(auto first_indexer, BTreeGlobalIndexer::Create(options)); + ASSERT_OK_AND_ASSIGN(std::unique_ptr first_indexer, + BTreeGlobalIndexer::Create(options)); auto first_schema = CreateArrowSchema(); auto first_file_reader = std::make_shared(fs_, base_path_); ASSERT_OK_AND_ASSIGN( - auto first_reader, + std::shared_ptr first_reader, first_indexer->CreateReader(first_schema.get(), first_file_reader, all_metas_, pool_)); - ASSERT_OK_AND_ASSIGN(auto first_result, first_reader->VisitEqual(literal_1)); + ASSERT_OK_AND_ASSIGN(std::shared_ptr first_result, + first_reader->VisitEqual(literal_1)); CheckResult(first_result, {0, 1}); first_reader.reset(); first_indexer.reset(); - ASSERT_OK_AND_ASSIGN(auto second_indexer, BTreeGlobalIndexer::Create(options)); + ASSERT_OK_AND_ASSIGN(std::unique_ptr second_indexer, + BTreeGlobalIndexer::Create(options)); auto second_schema = CreateArrowSchema(); auto failing_file_reader = std::make_shared(); ASSERT_OK_AND_ASSIGN( - auto second_reader, + std::shared_ptr second_reader, second_indexer->CreateReader(second_schema.get(), failing_file_reader, all_metas_, pool_)); - ASSERT_OK_AND_ASSIGN(auto second_result, second_reader->VisitEqual(literal_1)); + ASSERT_OK_AND_ASSIGN(std::shared_ptr second_result, + second_reader->VisitEqual(literal_1)); CheckResult(second_result, {0, 1}); } diff --git a/src/paimon/common/sst/block_cache_test.cpp b/src/paimon/common/sst/block_cache_test.cpp index 5c4323923..f7cae0c6b 100644 --- a/src/paimon/common/sst/block_cache_test.cpp +++ b/src/paimon/common/sst/block_cache_test.cpp @@ -211,7 +211,7 @@ TEST_F(BlockCacheTest, TestRetainedPagesAreReusedByLaterReader) { { BlockCache first_reader(file_path, supplier, cache_manager, retained_pool); - ASSERT_OK_AND_ASSIGN(auto segment, GetBlock(0, block_size, &first_reader)); + ASSERT_OK_AND_ASSIGN(MemorySegment segment, GetBlock(0, block_size, &first_reader)); ASSERT_EQ(segment.Get(0), static_cast(0)); ASSERT_EQ(input_stream_opens, 1); } @@ -221,7 +221,7 @@ TEST_F(BlockCacheTest, TestRetainedPagesAreReusedByLaterReader) { ASSERT_TRUE(retained_pool_ref.expired()); { BlockCache second_reader(file_path, supplier, cache_manager, pool_); - ASSERT_OK_AND_ASSIGN(auto cached_segment, GetBlock(0, block_size, &second_reader)); + ASSERT_OK_AND_ASSIGN(MemorySegment cached_segment, GetBlock(0, block_size, &second_reader)); ASSERT_EQ(cached_segment.Get(0), static_cast(0)); ASSERT_EQ(input_stream_opens, 1); } 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 f5865a0c2..787ac373c 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 @@ -279,7 +279,8 @@ TEST(AppendOnlyFileStoreScanTest, TestSnapshotLiveManifestCachePath) { ASSERT_OK_AND_ASSIGN(Snapshot snapshot_5, scan_first->GetSnapshotManager()->LoadSnapshot(/*snapshot_id=*/5)); scan_first->WithSnapshot(snapshot_5); - ASSERT_OK_AND_ASSIGN(auto plan_first, scan_first->CreatePlan()); + ASSERT_OK_AND_ASSIGN(std::shared_ptr plan_first, + scan_first->CreatePlan()); std::vector first_file_names = SortedFileNames(plan_first->Files()); std::shared_ptr first_metrics = scan_first->GetScanMetrics(); ASSERT_OK_AND_ASSIGN(uint64_t first_cache_enabled, @@ -299,7 +300,8 @@ TEST(AppendOnlyFileStoreScanTest, TestSnapshotLiveManifestCachePath) { // 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); - ASSERT_OK_AND_ASSIGN(auto plan_second, scan_second->CreatePlan()); + ASSERT_OK_AND_ASSIGN(std::shared_ptr plan_second, + scan_second->CreatePlan()); ASSERT_EQ(first_file_names, SortedFileNames(plan_second->Files())); std::shared_ptr second_metrics = scan_second->GetScanMetrics(); ASSERT_OK_AND_ASSIGN(uint64_t second_cache_hit, From ba430937a3169c3e2edf88552281c2395314fcc2 Mon Sep 17 00:00:00 2001 From: "wangyong.alen" Date: Mon, 31 Aug 2026 09:26:48 -0400 Subject: [PATCH 5/8] test(index): allow warm btree cache in integration test --- test/inte/primary_key_sorted_index_inte_test.cpp | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/test/inte/primary_key_sorted_index_inte_test.cpp b/test/inte/primary_key_sorted_index_inte_test.cpp index 8c86f4e04..252afee9d 100644 --- a/test/inte/primary_key_sorted_index_inte_test.cpp +++ b/test/inte/primary_key_sorted_index_inte_test.cpp @@ -352,7 +352,10 @@ TEST_P(PrimaryKeySortedIndexInteTest, MultiSourceOrdinalsBecomeBoundedFileLocalR } ASSERT_GE(source_files.size(), 2); ASSERT_EQ(200, selected_rows); - ASSERT_EQ(1, tracking_file_system->OpenedIndexPaths().size()); + // HistoricalSnapshotsMixedFilesAndDisabledIndex may already have populated the process-wide + // immutable BTree cache for this fixture. A cold scan opens the single index file; a warm scan + // opens none. Multiple source files must never cause more than one distinct index file open. + ASSERT_LE(tracking_file_system->OpenedIndexPaths().size(), 1); ASSERT_OK_AND_ASSIGN(std::vector rows, Read(/*partitioned=*/false, predicate, plan->Splits(), /*options=*/{}, From 9cd2147b6765bf373e41287ffce3e62c3f1b62f7 Mon Sep 17 00:00:00 2001 From: "wangyong.alen" Date: Mon, 31 Aug 2026 13:52:09 -0400 Subject: [PATCH 6/8] fix(scan): isolate shared cache reuse --- include/paimon/cache/cache.h | 5 +- .../io/global_index_file_reader.h | 49 +++++++++++++++ .../btree/btree_global_indexer.cpp | 35 ++++++----- .../btree/lazy_filtered_btree_reader.cpp | 5 +- .../btree/lazy_filtered_btree_reader_test.cpp | 15 ++++- src/paimon/common/io/cache/cache_key.cpp | 39 ++++++------ src/paimon/common/io/cache/cache_key.h | 6 +- src/paimon/common/io/cache/lru_cache_test.cpp | 11 +++- src/paimon/common/sst/block_cache.h | 10 +++- src/paimon/common/sst/block_cache_test.cpp | 4 +- .../global_index/global_index_file_manager.h | 2 +- .../snapshot_live_manifest_entries.cpp | 19 +++--- .../manifest/snapshot_live_manifest_entries.h | 11 +++- .../append_only_file_store_scan_test.cpp | 29 --------- src/paimon/core/operation/file_store_scan.cpp | 59 ++++++++++--------- src/paimon/core/operation/file_store_scan.h | 10 ++-- .../core/operation/file_store_scan_test.cpp | 13 ++-- .../source/primary_key_sorted_index_scan.cpp | 3 +- .../primary_key_sorted_index_inte_test.cpp | 5 +- 19 files changed, 197 insertions(+), 133 deletions(-) diff --git a/include/paimon/cache/cache.h b/include/paimon/cache/cache.h index e5e5c67fd..a8a5d0653 100644 --- a/include/paimon/cache/cache.h +++ b/include/paimon/cache/cache.h @@ -42,10 +42,11 @@ class PAIMON_EXPORT CacheKey { public: static std::shared_ptr ForPosition(const std::string& file_path, int64_t position, int32_t length, bool is_index); + static std::shared_ptr ForPosition(const std::string& cache_namespace, + const std::string& file_path, int64_t position, + int32_t length, bool is_index); static std::shared_ptr ForKind(const std::string& file_path, int64_t position, int32_t length, CacheKind kind); - static std::shared_ptr ForSnapshotLiveManifestEntries(const std::string& table_path, - const std::string& branch); static std::shared_ptr ForSnapshotLiveManifestEntries(const std::string& table_path, const std::string& branch, int32_t bucket); diff --git a/include/paimon/global_index/io/global_index_file_reader.h b/include/paimon/global_index/io/global_index_file_reader.h index ec553d3df..ad210519e 100644 --- a/include/paimon/global_index/io/global_index_file_reader.h +++ b/include/paimon/global_index/io/global_index_file_reader.h @@ -19,9 +19,15 @@ #pragma once +#include +#include +#include #include +#include #include +#include +#include "fmt/format.h" #include "paimon/result.h" #include "paimon/visibility.h" namespace paimon { @@ -29,11 +35,54 @@ class InputStream; /// Abstract interface for reading global index files from storage. class PAIMON_EXPORT GlobalIndexFileReader { public: + GlobalIndexFileReader() : cache_namespace_(NewCacheNamespace()) {} virtual ~GlobalIndexFileReader() = default; + /// Process-local backend identity used to isolate shared immutable-page caches. + const std::string& CacheNamespace() const { + return cache_namespace_; + } + /// Opens an input stream for reading the specified global index file. virtual Result> GetInputStream( const std::string& file_path) const = 0; + + protected: + explicit GlobalIndexFileReader(std::string cache_namespace) + : cache_namespace_(std::move(cache_namespace)) {} + + /// Returns a stable namespace while the same backend object is shared by reader wrappers. Dead + /// backends are removed from the registry and namespace ids are never reused. + static std::string CacheNamespaceFor(const std::shared_ptr& backend) { + using WeakBackend = std::weak_ptr; + static std::mutex mutex; + static std::map> namespaces; + + std::lock_guard lock(mutex); + for (auto iter = namespaces.begin(); iter != namespaces.end();) { + if (iter->first.expired()) { + iter = namespaces.erase(iter); + } else { + ++iter; + } + } + WeakBackend key(backend); + auto iter = namespaces.find(key); + if (iter != namespaces.end()) { + return iter->second; + } + std::string cache_namespace = NewCacheNamespace(); + namespaces.emplace(std::move(key), cache_namespace); + return cache_namespace; + } + + private: + static std::string NewCacheNamespace() { + static std::atomic next_namespace{0}; + return fmt::format("index-backend:{}", next_namespace.fetch_add(1)); + } + + std::string cache_namespace_; }; } // namespace paimon diff --git a/src/paimon/common/global_index/btree/btree_global_indexer.cpp b/src/paimon/common/global_index/btree/btree_global_indexer.cpp index b5312634f..6823231af 100644 --- a/src/paimon/common/global_index/btree/btree_global_indexer.cpp +++ b/src/paimon/common/global_index/btree/btree_global_indexer.cpp @@ -19,6 +19,7 @@ #include "paimon/common/global_index/btree/btree_global_indexer.h" #include +#include #include #include #include @@ -45,36 +46,40 @@ #include "paimon/executor.h" #include "paimon/global_index/bitmap_global_index_result.h" #include "paimon/memory/bytes.h" -#include "paimon/memory/memory_pool.h" #include "paimon/utils/roaring_bitmap64.h" namespace paimon { namespace { -struct SharedCacheManager { - SharedCacheManager(int64_t cache_size, double high_priority_pool_ratio) - : cache_pool(GetDefaultPool()), - cache_manager(std::make_shared(cache_size, high_priority_pool_ratio)) {} - - // Keep the allocator alive until after cache_manager releases all cached pages. - std::shared_ptr cache_pool; - std::shared_ptr cache_manager; -}; - std::shared_ptr GetSharedCacheManager(int64_t cache_size, double high_priority_pool_ratio) { using CacheConfig = std::pair; + struct Entry { + std::shared_ptr manager; + std::list::iterator lru_position; + }; + static constexpr size_t kMaxConfigurations = 4; static std::mutex mutex; - static std::map cache_managers; + static std::list lru_configs; + static std::map cache_managers; std::lock_guard lock(mutex); CacheConfig config(cache_size, high_priority_pool_ratio); auto iter = cache_managers.find(config); if (iter != cache_managers.end()) { - return iter->second.cache_manager; + lru_configs.splice(lru_configs.begin(), lru_configs, iter->second.lru_position); + return iter->second.manager; + } + + lru_configs.push_front(config); + auto manager = std::make_shared(cache_size, high_priority_pool_ratio); + cache_managers.emplace(config, Entry{manager, lru_configs.begin()}); + if (cache_managers.size() > kMaxConfigurations) { + CacheConfig evicted = lru_configs.back(); + lru_configs.pop_back(); + cache_managers.erase(evicted); } - return cache_managers.emplace(config, SharedCacheManager(cache_size, high_priority_pool_ratio)) - .first->second.cache_manager; + return manager; } } // namespace diff --git a/src/paimon/common/global_index/btree/lazy_filtered_btree_reader.cpp b/src/paimon/common/global_index/btree/lazy_filtered_btree_reader.cpp index b2b51159a..1282828e9 100644 --- a/src/paimon/common/global_index/btree/lazy_filtered_btree_reader.cpp +++ b/src/paimon/common/global_index/btree/lazy_filtered_btree_reader.cpp @@ -250,8 +250,9 @@ Result> LazyFilteredBTreeReader::CreateSingle } return input_stream; }; - auto block_cache = std::make_shared( - meta.file_path, std::move(input_stream_supplier), cache_manager_, pool_); + auto block_cache = + std::make_shared(file_reader_->CacheNamespace(), meta.file_path, + std::move(input_stream_supplier), cache_manager_, pool_); // Read footer PAIMON_ASSIGN_OR_RAISE(MemorySegment footer_segment, diff --git a/src/paimon/common/global_index/btree/lazy_filtered_btree_reader_test.cpp b/src/paimon/common/global_index/btree/lazy_filtered_btree_reader_test.cpp index ca48df079..9e5311ae9 100644 --- a/src/paimon/common/global_index/btree/lazy_filtered_btree_reader_test.cpp +++ b/src/paimon/common/global_index/btree/lazy_filtered_btree_reader_test.cpp @@ -72,7 +72,7 @@ class FakeLazyFileWriter : public GlobalIndexFileWriter { class FakeLazyFileReader : public GlobalIndexFileReader { public: FakeLazyFileReader(const std::shared_ptr& fs, const std::string& base_path) - : fs_(fs), base_path_(base_path) {} + : GlobalIndexFileReader(CacheNamespaceFor(fs)), fs_(fs), base_path_(base_path) {} Result> GetInputStream( const std::string& file_path) const override { @@ -86,6 +86,9 @@ class FakeLazyFileReader : public GlobalIndexFileReader { class FailingLazyFileReader : public GlobalIndexFileReader { public: + explicit FailingLazyFileReader(std::string cache_namespace) + : GlobalIndexFileReader(std::move(cache_namespace)) {} + Result> GetInputStream(const std::string&) const override { return Status::Invalid("unexpected input stream open for cached btree file"); } @@ -406,7 +409,15 @@ TEST_F(LazyFilteredBTreeReaderTest, TestBlockCacheReuseAcrossIndexerInstances) { ASSERT_OK_AND_ASSIGN(std::unique_ptr second_indexer, BTreeGlobalIndexer::Create(options)); auto second_schema = CreateArrowSchema(); - auto failing_file_reader = std::make_shared(); + auto isolated_file_reader = std::make_shared("other-backend"); + ASSERT_OK_AND_ASSIGN( + std::shared_ptr isolated_reader, + second_indexer->CreateReader(second_schema.get(), isolated_file_reader, all_metas_, pool_)); + ASSERT_NOK_WITH_MSG(isolated_reader->VisitEqual(literal_1), + "unexpected input stream open for cached btree file"); + + auto failing_file_reader = + std::make_shared(first_file_reader->CacheNamespace()); ASSERT_OK_AND_ASSIGN( std::shared_ptr second_reader, second_indexer->CreateReader(second_schema.get(), failing_file_reader, all_metas_, pool_)); diff --git a/src/paimon/common/io/cache/cache_key.cpp b/src/paimon/common/io/cache/cache_key.cpp index ae23eb303..c5bdba435 100644 --- a/src/paimon/common/io/cache/cache_key.cpp +++ b/src/paimon/common/io/cache/cache_key.cpp @@ -18,15 +18,13 @@ #include "paimon/common/io/cache/cache_key.h" -#include - namespace paimon { namespace { class SnapshotLiveManifestEntriesCacheKey : public CacheKey { public: SnapshotLiveManifestEntriesCacheKey(const std::string& table_path, const std::string& branch, - const std::optional& bucket) + int32_t bucket) : CacheKey(CacheKind::SNAPSHOT_LIVE_MANIFEST), table_path_(table_path), branch_(branch), @@ -49,11 +47,7 @@ class SnapshotLiveManifestEntriesCacheKey : public CacheKey { size_t seed = 0; seed ^= std::hash{}(table_path_) + HASH_CONSTANT + (seed << 6) + (seed >> 2); seed ^= std::hash{}(branch_) + HASH_CONSTANT + (seed << 6) + (seed >> 2); - seed ^= std::hash{}(bucket_.has_value()) + HASH_CONSTANT + (seed << 6) + (seed >> 2); - if (bucket_) { - seed ^= - std::hash{}(bucket_.value()) + HASH_CONSTANT + (seed << 6) + (seed >> 2); - } + seed ^= std::hash{}(bucket_) + HASH_CONSTANT + (seed << 6) + (seed >> 2); seed ^= std::hash{}(static_cast(GetKind())) + HASH_CONSTANT + (seed << 6) + (seed >> 2); return seed; @@ -64,29 +58,32 @@ class SnapshotLiveManifestEntriesCacheKey : public CacheKey { const std::string table_path_; const std::string branch_; - const std::optional bucket_; + const int32_t bucket_; }; } // namespace std::shared_ptr CacheKey::ForPosition(const std::string& file_path, int64_t position, int32_t length, bool is_index) { - return std::make_shared(file_path, position, length, is_index, - CacheKind::DEFAULT); + return std::make_shared(/*cache_namespace=*/"", file_path, position, length, + is_index, CacheKind::DEFAULT); +} + +std::shared_ptr CacheKey::ForPosition(const std::string& cache_namespace, + const std::string& file_path, int64_t position, + int32_t length, bool is_index) { + return std::make_shared(cache_namespace, file_path, position, length, + is_index, CacheKind::DEFAULT); } std::shared_ptr CacheKey::ForKind(const std::string& file_path, int64_t position, int32_t length, CacheKind kind) { - auto key = std::make_shared(file_path, position, length, - /*is_index=*/false, kind); + auto key = + std::make_shared(/*cache_namespace=*/"", file_path, position, length, + /*is_index=*/false, kind); return key; } -std::shared_ptr CacheKey::ForSnapshotLiveManifestEntries(const std::string& table_path, - const std::string& branch) { - return std::make_shared(table_path, branch, std::nullopt); -} - std::shared_ptr CacheKey::ForSnapshotLiveManifestEntries(const std::string& table_path, const std::string& branch, int32_t bucket) { @@ -110,12 +107,14 @@ bool PositionCacheKey::Equals(const CacheKey& other) const { if (!rhs) { return false; } - return file_path_ == rhs->file_path_ && position_ == rhs->position_ && - length_ == rhs->length_ && is_index_ == rhs->is_index_ && GetKind() == rhs->GetKind(); + return cache_namespace_ == rhs->cache_namespace_ && file_path_ == rhs->file_path_ && + position_ == rhs->position_ && length_ == rhs->length_ && is_index_ == rhs->is_index_ && + GetKind() == rhs->GetKind(); } size_t PositionCacheKey::HashCode() const { size_t seed = 0; + seed ^= std::hash{}(cache_namespace_) + HASH_CONSTANT + (seed << 6) + (seed >> 2); seed ^= std::hash{}(file_path_) + HASH_CONSTANT + (seed << 6) + (seed >> 2); seed ^= std::hash{}(position_) + HASH_CONSTANT + (seed << 6) + (seed >> 2); seed ^= std::hash{}(length_) + HASH_CONSTANT + (seed << 6) + (seed >> 2); diff --git a/src/paimon/common/io/cache/cache_key.h b/src/paimon/common/io/cache/cache_key.h index 988735d11..4c4399ea0 100644 --- a/src/paimon/common/io/cache/cache_key.h +++ b/src/paimon/common/io/cache/cache_key.h @@ -28,9 +28,10 @@ namespace paimon { class PositionCacheKey : public CacheKey { public: - PositionCacheKey(const std::string& file_path, int64_t position, int32_t length, bool is_index, - CacheKind kind) + PositionCacheKey(const std::string& cache_namespace, const std::string& file_path, + int64_t position, int32_t length, bool is_index, CacheKind kind) : CacheKey(kind), + cache_namespace_(cache_namespace), file_path_(file_path), position_(position), length_(length), @@ -45,6 +46,7 @@ class PositionCacheKey : public CacheKey { private: static constexpr uint64_t HASH_CONSTANT = 0x9e3779b97f4a7c15ULL; + const std::string cache_namespace_; const std::string file_path_; const int64_t position_; const int32_t length_; diff --git a/src/paimon/common/io/cache/lru_cache_test.cpp b/src/paimon/common/io/cache/lru_cache_test.cpp index bae436228..e8bc0fafe 100644 --- a/src/paimon/common/io/cache/lru_cache_test.cpp +++ b/src/paimon/common/io/cache/lru_cache_test.cpp @@ -391,17 +391,24 @@ TEST_F(LruCacheTest, TestForSnapshotLiveManifestEntries) { auto bucket_key = CacheKey::ForSnapshotLiveManifestEntries("table_path", "main", 1); auto hash_in_path_key = CacheKey::ForSnapshotLiveManifestEntries("table#path", "main", 0); auto hash_in_branch_key = CacheKey::ForSnapshotLiveManifestEntries("table", "path#main", 0); - auto all_buckets_key = CacheKey::ForSnapshotLiveManifestEntries("table_path", "main"); ASSERT_EQ(CacheKind::SNAPSHOT_LIVE_MANIFEST, main_key->GetKind()); ASSERT_TRUE(CacheKeyEqual()(main_key, same_key)); ASSERT_FALSE(CacheKeyEqual()(main_key, branch_key)); ASSERT_FALSE(CacheKeyEqual()(main_key, table_key)); ASSERT_FALSE(CacheKeyEqual()(main_key, bucket_key)); - ASSERT_FALSE(CacheKeyEqual()(main_key, all_buckets_key)); ASSERT_FALSE(CacheKeyEqual()(hash_in_path_key, hash_in_branch_key)); } +TEST_F(LruCacheTest, TestPositionKeyIncludesBackendNamespace) { + auto first = CacheKey::ForPosition("backend-a", "same-path", 0, 64, /*is_index=*/false); + auto same = CacheKey::ForPosition("backend-a", "same-path", 0, 64, /*is_index=*/false); + auto other = CacheKey::ForPosition("backend-b", "same-path", 0, 64, /*is_index=*/false); + + ASSERT_TRUE(CacheKeyEqual()(first, same)); + ASSERT_FALSE(CacheKeyEqual()(first, other)); +} + /// Verifies that multiple evictions happen when a single large entry is inserted. TEST_F(LruCacheTest, TestMultipleEvictions) { LruCache cache(300); diff --git a/src/paimon/common/sst/block_cache.h b/src/paimon/common/sst/block_cache.h index ea0b8ae10..8f7e6220c 100644 --- a/src/paimon/common/sst/block_cache.h +++ b/src/paimon/common/sst/block_cache.h @@ -41,11 +41,13 @@ class PAIMON_EXPORT BlockCache { /// Creates a block cache whose pages survive this reader. The input stream is opened only on a /// cache miss, so a later reader can reuse cached pages without reopening the immutable file. - BlockCache(const std::string& file_path, InputStreamSupplier input_stream_supplier, + BlockCache(const std::string& cache_namespace, const std::string& file_path, + InputStreamSupplier input_stream_supplier, const std::shared_ptr& cache_manager, const std::shared_ptr& pool) : pool_(pool), cache_pool_(GetDefaultPool()), + cache_namespace_(cache_namespace), file_path_(file_path), input_stream_supplier_(std::move(input_stream_supplier)), cache_manager_(cache_manager), @@ -58,7 +60,10 @@ class PAIMON_EXPORT BlockCache { Result GetBlock( int64_t position, int32_t length, bool is_index, std::function(const MemorySegment&)> decompress_func) { - auto key = CacheKey::ForPosition(file_path_, position, length, is_index); + auto key = + retain_cached_pages_ + ? CacheKey::ForPosition(cache_namespace_, file_path_, position, length, is_index) + : CacheKey::ForPosition(file_path_, position, length, is_index); auto reader = [&](const std::shared_ptr&) -> Result { PAIMON_ASSIGN_OR_RAISE(MemorySegment compress_data, ReadFrom(position, length)); if (!decompress_func) { @@ -140,6 +145,7 @@ class PAIMON_EXPORT BlockCache { private: std::shared_ptr pool_; std::shared_ptr cache_pool_; + std::string cache_namespace_; std::string file_path_; std::shared_ptr in_; InputStreamSupplier input_stream_supplier_; diff --git a/src/paimon/common/sst/block_cache_test.cpp b/src/paimon/common/sst/block_cache_test.cpp index f7cae0c6b..22fb0732e 100644 --- a/src/paimon/common/sst/block_cache_test.cpp +++ b/src/paimon/common/sst/block_cache_test.cpp @@ -210,7 +210,7 @@ TEST_F(BlockCacheTest, TestRetainedPagesAreReusedByLaterReader) { }; { - BlockCache first_reader(file_path, supplier, cache_manager, retained_pool); + BlockCache first_reader("test-backend", file_path, supplier, cache_manager, retained_pool); ASSERT_OK_AND_ASSIGN(MemorySegment segment, GetBlock(0, block_size, &first_reader)); ASSERT_EQ(segment.Get(0), static_cast(0)); ASSERT_EQ(input_stream_opens, 1); @@ -220,7 +220,7 @@ TEST_F(BlockCacheTest, TestRetainedPagesAreReusedByLaterReader) { retained_pool.reset(); ASSERT_TRUE(retained_pool_ref.expired()); { - BlockCache second_reader(file_path, supplier, cache_manager, pool_); + BlockCache second_reader("test-backend", file_path, supplier, cache_manager, pool_); ASSERT_OK_AND_ASSIGN(MemorySegment cached_segment, GetBlock(0, block_size, &second_reader)); ASSERT_EQ(cached_segment.Get(0), static_cast(0)); ASSERT_EQ(input_stream_opens, 1); diff --git a/src/paimon/core/global_index/global_index_file_manager.h b/src/paimon/core/global_index/global_index_file_manager.h index 3db9965b2..a2ea66f56 100644 --- a/src/paimon/core/global_index/global_index_file_manager.h +++ b/src/paimon/core/global_index/global_index_file_manager.h @@ -34,7 +34,7 @@ class GlobalIndexFileManager : public GlobalIndexFileReader, public GlobalIndexF public: GlobalIndexFileManager(const std::shared_ptr& fs, const std::shared_ptr& path_factory) - : fs_(fs), path_factory_(path_factory) {} + : GlobalIndexFileReader(CacheNamespaceFor(fs)), fs_(fs), path_factory_(path_factory) {} Result> GetInputStream( const std::string& file_path) const override { diff --git a/src/paimon/core/manifest/snapshot_live_manifest_entries.cpp b/src/paimon/core/manifest/snapshot_live_manifest_entries.cpp index 8982209a6..6d5172421 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 = 0x534d4532; // SME2 size_t NormalizeMaxSnapshots(int32_t max_snapshots) { return static_cast(std::max(0, max_snapshots)); @@ -68,15 +68,17 @@ std::optional SnapshotLiveManifestEntries::L return std::optional(); } --iter; - return Entry{iter->first, iter->second}; + return Entry{iter->first, iter->second.snapshot_generation, iter->second.entries}; } -void SnapshotLiveManifestEntries::Put(int64_t snapshot_id, std::vector&& entries) { +void SnapshotLiveManifestEntries::Put(int64_t snapshot_id, const std::string& snapshot_generation, + std::vector&& entries) { if (NormalizeMaxSnapshots(max_snapshots_) == 0) { return; } entries_by_snapshot_[snapshot_id] = - std::make_shared>(std::move(entries)); + StoredEntry{snapshot_generation, + std::make_shared>(std::move(entries))}; EvictIfNeeded(); } @@ -91,9 +93,10 @@ Result> SnapshotLiveManifestEntries::Serialize( out.WriteValue(static_cast(entries_by_snapshot_.size())); ManifestEntrySerializer serializer(pool); - for (const auto& [snapshot_id, entries] : entries_by_snapshot_) { + for (const auto& [snapshot_id, entry] : entries_by_snapshot_) { out.WriteValue(snapshot_id); - PAIMON_RETURN_NOT_OK(serializer.SerializeList(*entries, &out)); + out.WriteString(entry.snapshot_generation); + PAIMON_RETURN_NOT_OK(serializer.SerializeList(*entry.entries, &out)); } return ToBytes(out, pool); } @@ -121,9 +124,11 @@ 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(std::string snapshot_generation, in.ReadString()); 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)); + StoredEntry{std::move(snapshot_generation), + std::make_shared>(std::move(entries))}; } 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..294c0e4ea 100644 --- a/src/paimon/core/manifest/snapshot_live_manifest_entries.h +++ b/src/paimon/core/manifest/snapshot_live_manifest_entries.h @@ -24,6 +24,7 @@ #include #include #include +#include #include #include "paimon/core/manifest/manifest_entry.h" @@ -42,13 +43,15 @@ class SnapshotLiveManifestEntries { public: struct Entry { int64_t snapshot_id; + std::string snapshot_generation; 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, const std::string& snapshot_generation, + std::vector&& entries); size_t Size() const; Result> Serialize(const std::shared_ptr& pool) const; @@ -59,7 +62,11 @@ class SnapshotLiveManifestEntries { private: void EvictIfNeeded(); - std::map>> entries_by_snapshot_; + struct StoredEntry { + std::string snapshot_generation; + std::shared_ptr> entries; + }; + 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 787ac373c..82937b391 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 @@ -319,35 +319,6 @@ TEST(AppendOnlyFileStoreScanTest, TestSnapshotLiveManifestCachePath) { ASSERT_OK(second_metrics->GetHistogramStats(ScanMetrics::SNAPSHOT_CACHE_LOAD_DURATION)); } -TEST(AppendOnlyFileStoreScanTest, TestSnapshotLiveManifestCacheWithoutBucketFilter) { - 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 scan_first = BuildScan(table_path, cache); - ASSERT_OK_AND_ASSIGN(Snapshot snapshot_5, - scan_first->GetSnapshotManager()->LoadSnapshot(/*snapshot_id=*/5)); - scan_first->WithSnapshot(snapshot_5); - ASSERT_OK_AND_ASSIGN(auto plan_first, scan_first->CreatePlan()); - std::vector first_file_names = SortedFileNames(plan_first->Files()); - std::shared_ptr first_metrics = scan_first->GetScanMetrics(); - ASSERT_OK_AND_ASSIGN(uint64_t first_cache_enabled, - first_metrics->GetCounter(ScanMetrics::LAST_SNAPSHOT_CACHE_ENABLED)); - ASSERT_OK_AND_ASSIGN(uint64_t first_cache_hit, - first_metrics->GetCounter(ScanMetrics::LAST_SNAPSHOT_CACHE_HIT)); - ASSERT_EQ(first_cache_enabled, 1); - ASSERT_EQ(first_cache_hit, 0); - - auto scan_second = BuildScan(table_path, cache); - scan_second->WithSnapshot(snapshot_5); - ASSERT_OK_AND_ASSIGN(auto plan_second, scan_second->CreatePlan()); - ASSERT_EQ(first_file_names, SortedFileNames(plan_second->Files())); - std::shared_ptr second_metrics = scan_second->GetScanMetrics(); - ASSERT_OK_AND_ASSIGN(uint64_t second_cache_hit, - second_metrics->GetCounter(ScanMetrics::LAST_SNAPSHOT_CACHE_HIT)); - ASSERT_EQ(second_cache_hit, 1); -} - TEST(AppendOnlyFileStoreScanTest, TestSnapshotLiveManifestCacheRebuildOnMiss) { TimezoneGuard guard("Asia/Shanghai"); std::string table_path = paimon::test::GetDataDir() + "/orc/append_09.db/append_09/"; diff --git a/src/paimon/core/operation/file_store_scan.cpp b/src/paimon/core/operation/file_store_scan.cpp index f65430dbf..88f704bfb 100644 --- a/src/paimon/core/operation/file_store_scan.cpp +++ b/src/paimon/core/operation/file_store_scan.cpp @@ -62,6 +62,15 @@ namespace paimon { enum class FieldType; +namespace { +std::string SnapshotCacheGeneration(const Snapshot& snapshot) { + std::string generation = snapshot.BaseManifestList(); + generation.push_back('\0'); + generation.append(snapshot.DeltaManifestList()); + return generation; +} +} // namespace + Result> FileStoreScan::ReconstructPredicateWithNonCastedFields( const std::shared_ptr& predicate, const std::shared_ptr& evolution) { @@ -150,12 +159,12 @@ Result> FileStoreScan::CreatePlan() cons snapshot.has_value() && scan_mode_ == ScanMode::ALL && core_options_.GetScanManifestEntryCacheMaxSnapshots() > 0 && core_options_.GetCache() != nullptr && !table_path_.empty() && - !row_range_index_.has_value(); + !row_range_index_.has_value() && bucket_filter_.has_value(); 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, - bucket_filter_, &manifest_entries, + bucket_filter_.value(), &manifest_entries, &snapshot_cache_hit)); lazy_decode_scanned_rows = manifest_entries.size(); std::vector filtered_entries; @@ -324,13 +333,12 @@ Status FileStoreScan::ReadManifestEntries(const std::vector& m return ReadAndNoMergeFileEntries(manifest_metas, manifest_entries); } -// Cache merged live manifest entries before applying scan filters. Each cache value keeps a -// bounded number of snapshot results for the same table/branch and optional bucket. Exact snapshot -// hits can be returned directly; cache misses rebuild the target snapshot from its data manifests. +// Cache merged live manifest entries for one bucket before applying scan filters. Each cache value +// keeps a bounded number of snapshot results for the same table/branch/bucket. Exact snapshot hits +// can be returned directly; cache misses rebuild the target snapshot bucket from its manifests. Status FileStoreScan::ReadManifestEntriesWithCache( const Snapshot& snapshot, const std::vector& all_manifest_metas, - const std::optional& bucket, std::vector* manifest_entries, - bool* cache_hit) const { + int32_t bucket, std::vector* manifest_entries, bool* cache_hit) const { Duration cache_load_duration; PAIMON_ASSIGN_OR_RAISE(SnapshotLiveManifestEntries cached_entries, LoadSnapshotLiveManifestEntries(bucket)); @@ -340,30 +348,25 @@ Status FileStoreScan::ReadManifestEntriesWithCache( static_cast(cache_load_duration_ms)); std::optional cached = cached_entries.LatestBeforeOrEqual(snapshot.Id()); - if (cached && cached->snapshot_id == snapshot.Id()) { + const std::string snapshot_generation = SnapshotCacheGeneration(snapshot); + if (cached && cached->snapshot_id == snapshot.Id() && + cached->snapshot_generation == snapshot_generation) { *cache_hit = true; *manifest_entries = *cached->entries; return Status::OK(); } *cache_hit = false; - if (bucket) { - std::vector bucket_manifest_metas; - for (const auto& meta : all_manifest_metas) { - if (MayContainBucket(meta, bucket.value())) { - bucket_manifest_metas.push_back(meta); - } + std::vector bucket_manifest_metas; + for (const auto& meta : all_manifest_metas) { + if (MayContainBucket(meta, bucket)) { + bucket_manifest_metas.push_back(meta); } - PAIMON_RETURN_NOT_OK( - ReadAndMergeBucketFileEntries(bucket_manifest_metas, bucket.value(), manifest_entries)); - } else { - std::vector unmerged_entries; - PAIMON_RETURN_NOT_OK( - ReadFileEntries(all_manifest_metas, &unmerged_entries, /*apply_scan_filter=*/false)); - PAIMON_RETURN_NOT_OK(MergeLiveEntries(unmerged_entries, manifest_entries)); } + 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(), snapshot_generation, 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(); @@ -373,15 +376,13 @@ Status FileStoreScan::ReadManifestEntriesWithCache( return Status::OK(); } -std::shared_ptr FileStoreScan::SnapshotLiveManifestEntriesCacheKey( - const std::optional& bucket) const { - const std::string branch = BranchManager::NormalizeBranch(core_options_.GetBranch()); - return bucket ? CacheKey::ForSnapshotLiveManifestEntries(table_path_, branch, bucket.value()) - : CacheKey::ForSnapshotLiveManifestEntries(table_path_, branch); +std::shared_ptr FileStoreScan::SnapshotLiveManifestEntriesCacheKey(int32_t bucket) const { + return CacheKey::ForSnapshotLiveManifestEntries( + table_path_, BranchManager::NormalizeBranch(core_options_.GetBranch()), bucket); } Result FileStoreScan::LoadSnapshotLiveManifestEntries( - const std::optional& bucket) const { + int32_t bucket) const { auto supplier = [](const std::shared_ptr&) -> Result> { return std::shared_ptr(); }; @@ -401,7 +402,7 @@ Result FileStoreScan::LoadSnapshotLiveManifestEntri } Status FileStoreScan::StoreSnapshotLiveManifestEntries( - const std::optional& bucket, const SnapshotLiveManifestEntries& entries) const { + int32_t bucket, const SnapshotLiveManifestEntries& entries) const { Result> bytes_result = entries.Serialize(pool_); if (!bytes_result.ok()) { return Status::OK(); diff --git a/src/paimon/core/operation/file_store_scan.h b/src/paimon/core/operation/file_store_scan.h index 962637341..f7f770ca5 100644 --- a/src/paimon/core/operation/file_store_scan.h +++ b/src/paimon/core/operation/file_store_scan.h @@ -261,14 +261,12 @@ class FileStoreScan { Status ReadManifestEntriesWithCache(const Snapshot& snapshot, const std::vector& all_manifest_metas, - const std::optional& bucket, + int32_t bucket, std::vector* manifest_entries, bool* cache_hit) const; - std::shared_ptr SnapshotLiveManifestEntriesCacheKey( - const std::optional& bucket) const; - Result LoadSnapshotLiveManifestEntries( - const std::optional& bucket) const; - Status StoreSnapshotLiveManifestEntries(const std::optional& bucket, + std::shared_ptr SnapshotLiveManifestEntriesCacheKey(int32_t bucket) const; + Result LoadSnapshotLiveManifestEntries(int32_t bucket) const; + Status StoreSnapshotLiveManifestEntries(int32_t bucket, const SnapshotLiveManifestEntries& entries) const; Status ReadAndMergeBucketFileEntries(const std::vector& manifest_metas, diff --git a/src/paimon/core/operation/file_store_scan_test.cpp b/src/paimon/core/operation/file_store_scan_test.cpp index c4f0289e7..b92ef810e 100644 --- a/src/paimon/core/operation/file_store_scan_test.cpp +++ b/src/paimon/core/operation/file_store_scan_test.cpp @@ -203,11 +203,12 @@ TEST_F(FileStoreScanTest, TestSnapshotLiveManifestEntries) { 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)); + entries.Put(/*snapshot_id=*/1, /*snapshot_generation=*/"generation-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->snapshot_generation, "generation-1"); ASSERT_EQ(hit->entries->size(), 1); ASSERT_EQ((*hit->entries)[0].FileName(), "file-1"); auto latest_before_2 = entries.LatestBeforeOrEqual(/*snapshot_id=*/2); @@ -225,13 +226,13 @@ TEST_F(FileStoreScanTest, TestSnapshotLiveManifestEntries) { /*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)); + entries.Put(/*snapshot_id=*/3, /*snapshot_generation=*/"generation-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, {}); + entries.Put(/*snapshot_id=*/5, /*snapshot_generation=*/"generation-5", {}); ASSERT_EQ(entries.Size(), 2); ASSERT_FALSE(entries.LatestBeforeOrEqual(/*snapshot_id=*/1)); ASSERT_TRUE(entries.LatestBeforeOrEqual(/*snapshot_id=*/3)); @@ -251,8 +252,9 @@ 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, /*snapshot_generation=*/"generation-1", + std::move(manifest_entries)); + entries.Put(/*snapshot_id=*/3, /*snapshot_generation=*/"generation-3", {}); ASSERT_OK_AND_ASSIGN(auto bytes, entries.Serialize(GetDefaultPool())); ASSERT_OK_AND_ASSIGN(auto deserialized, @@ -262,6 +264,7 @@ TEST_F(FileStoreScanTest, TestSnapshotLiveManifestEntriesSerialization) { auto hit = deserialized.LatestBeforeOrEqual(/*snapshot_id=*/2); ASSERT_TRUE(hit); ASSERT_EQ(hit->snapshot_id, 1); + ASSERT_EQ(hit->snapshot_generation, "generation-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); diff --git a/src/paimon/core/table/source/primary_key_sorted_index_scan.cpp b/src/paimon/core/table/source/primary_key_sorted_index_scan.cpp index a523b920d..f40382052 100644 --- a/src/paimon/core/table/source/primary_key_sorted_index_scan.cpp +++ b/src/paimon/core/table/source/primary_key_sorted_index_scan.cpp @@ -510,7 +510,8 @@ namespace { class FsGlobalIndexFileReader : public GlobalIndexFileReader { public: explicit FsGlobalIndexFileReader(std::shared_ptr file_system) - : file_system_(std::move(file_system)) {} + : GlobalIndexFileReader(CacheNamespaceFor(file_system)), + file_system_(std::move(file_system)) {} Result> GetInputStream( const std::string& file_path) const override { diff --git a/test/inte/primary_key_sorted_index_inte_test.cpp b/test/inte/primary_key_sorted_index_inte_test.cpp index 252afee9d..8c86f4e04 100644 --- a/test/inte/primary_key_sorted_index_inte_test.cpp +++ b/test/inte/primary_key_sorted_index_inte_test.cpp @@ -352,10 +352,7 @@ TEST_P(PrimaryKeySortedIndexInteTest, MultiSourceOrdinalsBecomeBoundedFileLocalR } ASSERT_GE(source_files.size(), 2); ASSERT_EQ(200, selected_rows); - // HistoricalSnapshotsMixedFilesAndDisabledIndex may already have populated the process-wide - // immutable BTree cache for this fixture. A cold scan opens the single index file; a warm scan - // opens none. Multiple source files must never cause more than one distinct index file open. - ASSERT_LE(tracking_file_system->OpenedIndexPaths().size(), 1); + ASSERT_EQ(1, tracking_file_system->OpenedIndexPaths().size()); ASSERT_OK_AND_ASSIGN(std::vector rows, Read(/*partitioned=*/false, predicate, plan->Splits(), /*options=*/{}, From b35e9c700dda9aab607fb10871f5adb7fb71319c Mon Sep 17 00:00:00 2001 From: "wangyong.alen" Date: Mon, 31 Aug 2026 13:58:40 -0400 Subject: [PATCH 7/8] fix(index): keep backend isolation internal --- .../io/global_index_file_reader.h | 49 ------------------- .../btree/lazy_filtered_btree_reader.cpp | 7 +-- .../btree/lazy_filtered_btree_reader_test.cpp | 22 +++++++-- .../global_index/cache_namespace_provider.h | 46 +++++++++++++++++ src/paimon/common/sst/block_cache.h | 10 +++- src/paimon/common/sst/block_cache_test.cpp | 6 ++- .../global_index/global_index_file_manager.h | 12 ++++- .../source/primary_key_sorted_index_scan.cpp | 10 ++-- 8 files changed, 96 insertions(+), 66 deletions(-) create mode 100644 src/paimon/common/global_index/cache_namespace_provider.h diff --git a/include/paimon/global_index/io/global_index_file_reader.h b/include/paimon/global_index/io/global_index_file_reader.h index ad210519e..ec553d3df 100644 --- a/include/paimon/global_index/io/global_index_file_reader.h +++ b/include/paimon/global_index/io/global_index_file_reader.h @@ -19,15 +19,9 @@ #pragma once -#include -#include -#include #include -#include #include -#include -#include "fmt/format.h" #include "paimon/result.h" #include "paimon/visibility.h" namespace paimon { @@ -35,54 +29,11 @@ class InputStream; /// Abstract interface for reading global index files from storage. class PAIMON_EXPORT GlobalIndexFileReader { public: - GlobalIndexFileReader() : cache_namespace_(NewCacheNamespace()) {} virtual ~GlobalIndexFileReader() = default; - /// Process-local backend identity used to isolate shared immutable-page caches. - const std::string& CacheNamespace() const { - return cache_namespace_; - } - /// Opens an input stream for reading the specified global index file. virtual Result> GetInputStream( const std::string& file_path) const = 0; - - protected: - explicit GlobalIndexFileReader(std::string cache_namespace) - : cache_namespace_(std::move(cache_namespace)) {} - - /// Returns a stable namespace while the same backend object is shared by reader wrappers. Dead - /// backends are removed from the registry and namespace ids are never reused. - static std::string CacheNamespaceFor(const std::shared_ptr& backend) { - using WeakBackend = std::weak_ptr; - static std::mutex mutex; - static std::map> namespaces; - - std::lock_guard lock(mutex); - for (auto iter = namespaces.begin(); iter != namespaces.end();) { - if (iter->first.expired()) { - iter = namespaces.erase(iter); - } else { - ++iter; - } - } - WeakBackend key(backend); - auto iter = namespaces.find(key); - if (iter != namespaces.end()) { - return iter->second; - } - std::string cache_namespace = NewCacheNamespace(); - namespaces.emplace(std::move(key), cache_namespace); - return cache_namespace; - } - - private: - static std::string NewCacheNamespace() { - static std::atomic next_namespace{0}; - return fmt::format("index-backend:{}", next_namespace.fetch_add(1)); - } - - std::string cache_namespace_; }; } // namespace paimon diff --git a/src/paimon/common/global_index/btree/lazy_filtered_btree_reader.cpp b/src/paimon/common/global_index/btree/lazy_filtered_btree_reader.cpp index 1282828e9..ff032d285 100644 --- a/src/paimon/common/global_index/btree/lazy_filtered_btree_reader.cpp +++ b/src/paimon/common/global_index/btree/lazy_filtered_btree_reader.cpp @@ -27,6 +27,7 @@ #include "paimon/common/global_index/btree/btree_global_index_reader.h" #include "paimon/common/global_index/btree/btree_index_meta.h" #include "paimon/common/global_index/btree/key_serializer.h" +#include "paimon/common/global_index/cache_namespace_provider.h" #include "paimon/common/global_index/union_global_index_reader.h" #include "paimon/common/memory/memory_slice.h" #include "paimon/common/memory/memory_slice_input.h" @@ -250,9 +251,9 @@ Result> LazyFilteredBTreeReader::CreateSingle } return input_stream; }; - auto block_cache = - std::make_shared(file_reader_->CacheNamespace(), meta.file_path, - std::move(input_stream_supplier), cache_manager_, pool_); + auto block_cache = std::make_shared( + GetGlobalIndexCacheNamespace(file_reader_), file_reader_, meta.file_path, + std::move(input_stream_supplier), cache_manager_, pool_); // Read footer PAIMON_ASSIGN_OR_RAISE(MemorySegment footer_segment, diff --git a/src/paimon/common/global_index/btree/lazy_filtered_btree_reader_test.cpp b/src/paimon/common/global_index/btree/lazy_filtered_btree_reader_test.cpp index 9e5311ae9..a57420b94 100644 --- a/src/paimon/common/global_index/btree/lazy_filtered_btree_reader_test.cpp +++ b/src/paimon/common/global_index/btree/lazy_filtered_btree_reader_test.cpp @@ -28,6 +28,7 @@ #include "gtest/gtest.h" #include "paimon/common/global_index/btree/btree_global_index_writer.h" #include "paimon/common/global_index/btree/btree_global_indexer.h" +#include "paimon/common/global_index/cache_namespace_provider.h" #include "paimon/executor.h" #include "paimon/fs/file_system.h" #include "paimon/global_index/bitmap_global_index_result.h" @@ -69,10 +70,14 @@ class FakeLazyFileWriter : public GlobalIndexFileWriter { mutable int64_t file_counter_ = 0; }; -class FakeLazyFileReader : public GlobalIndexFileReader { +class FakeLazyFileReader : public GlobalIndexFileReader, public CacheNamespaceProvider { public: FakeLazyFileReader(const std::shared_ptr& fs, const std::string& base_path) - : GlobalIndexFileReader(CacheNamespaceFor(fs)), fs_(fs), base_path_(base_path) {} + : fs_(fs), base_path_(base_path) {} + + std::string CacheNamespace() const override { + return fmt::format("filesystem:{}", fmt::ptr(fs_.get())); + } Result> GetInputStream( const std::string& file_path) const override { @@ -84,14 +89,21 @@ class FakeLazyFileReader : public GlobalIndexFileReader { std::string base_path_; }; -class FailingLazyFileReader : public GlobalIndexFileReader { +class FailingLazyFileReader : public GlobalIndexFileReader, public CacheNamespaceProvider { public: explicit FailingLazyFileReader(std::string cache_namespace) - : GlobalIndexFileReader(std::move(cache_namespace)) {} + : cache_namespace_(std::move(cache_namespace)) {} + + std::string CacheNamespace() const override { + return cache_namespace_; + } Result> GetInputStream(const std::string&) const override { return Status::Invalid("unexpected input stream open for cached btree file"); } + + private: + std::string cache_namespace_; }; class LazyFilteredBTreeReaderTest : public ::testing::Test { @@ -417,7 +429,7 @@ TEST_F(LazyFilteredBTreeReaderTest, TestBlockCacheReuseAcrossIndexerInstances) { "unexpected input stream open for cached btree file"); auto failing_file_reader = - std::make_shared(first_file_reader->CacheNamespace()); + std::make_shared(GetGlobalIndexCacheNamespace(first_file_reader)); ASSERT_OK_AND_ASSIGN( std::shared_ptr second_reader, second_indexer->CreateReader(second_schema.get(), failing_file_reader, all_metas_, pool_)); diff --git a/src/paimon/common/global_index/cache_namespace_provider.h b/src/paimon/common/global_index/cache_namespace_provider.h new file mode 100644 index 000000000..4508e9d52 --- /dev/null +++ b/src/paimon/common/global_index/cache_namespace_provider.h @@ -0,0 +1,46 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#pragma once + +#include +#include + +#include "fmt/format.h" +#include "paimon/global_index/io/global_index_file_reader.h" + +namespace paimon { + +/// Internal extension for reader wrappers that can identify a shared storage backend. +class CacheNamespaceProvider { + public: + virtual ~CacheNamespaceProvider() = default; + virtual std::string CacheNamespace() const = 0; +}; + +inline std::string GetGlobalIndexCacheNamespace( + const std::shared_ptr& file_reader) { + const auto* provider = dynamic_cast(file_reader.get()); + if (provider) { + return provider->CacheNamespace(); + } + return fmt::format("index-reader:{}", fmt::ptr(file_reader.get())); +} + +} // namespace paimon diff --git a/src/paimon/common/sst/block_cache.h b/src/paimon/common/sst/block_cache.h index 8f7e6220c..ab49cf6f8 100644 --- a/src/paimon/common/sst/block_cache.h +++ b/src/paimon/common/sst/block_cache.h @@ -41,13 +41,15 @@ class PAIMON_EXPORT BlockCache { /// Creates a block cache whose pages survive this reader. The input stream is opened only on a /// cache miss, so a later reader can reuse cached pages without reopening the immutable file. - BlockCache(const std::string& cache_namespace, const std::string& file_path, + BlockCache(const std::string& cache_namespace, + const std::shared_ptr& cache_namespace_owner, const std::string& file_path, InputStreamSupplier input_stream_supplier, const std::shared_ptr& cache_manager, const std::shared_ptr& pool) : pool_(pool), cache_pool_(GetDefaultPool()), cache_namespace_(cache_namespace), + cache_namespace_owner_(cache_namespace_owner), file_path_(file_path), input_stream_supplier_(std::move(input_stream_supplier)), cache_manager_(cache_manager), @@ -81,7 +83,10 @@ class PAIMON_EXPORT BlockCache { return cached_data; }; if (retain_cached_pages_) { - return cache_manager_->GetPage(key, reader, /*eviction_callback=*/{}); + return cache_manager_->GetPage( + key, reader, [owner = cache_namespace_owner_](const std::shared_ptr&) { + (void)owner; + }); } auto it = blocks_.find(key); if (it == blocks_.end() || it->second.GetAccessCount() == CacheManager::REFRESH_COUNT) { @@ -146,6 +151,7 @@ class PAIMON_EXPORT BlockCache { std::shared_ptr pool_; std::shared_ptr cache_pool_; std::string cache_namespace_; + std::shared_ptr cache_namespace_owner_; std::string file_path_; std::shared_ptr in_; InputStreamSupplier input_stream_supplier_; diff --git a/src/paimon/common/sst/block_cache_test.cpp b/src/paimon/common/sst/block_cache_test.cpp index 22fb0732e..1ad8fa0ae 100644 --- a/src/paimon/common/sst/block_cache_test.cpp +++ b/src/paimon/common/sst/block_cache_test.cpp @@ -210,7 +210,8 @@ TEST_F(BlockCacheTest, TestRetainedPagesAreReusedByLaterReader) { }; { - BlockCache first_reader("test-backend", file_path, supplier, cache_manager, retained_pool); + BlockCache first_reader("test-backend", /*cache_namespace_owner=*/nullptr, file_path, + supplier, cache_manager, retained_pool); ASSERT_OK_AND_ASSIGN(MemorySegment segment, GetBlock(0, block_size, &first_reader)); ASSERT_EQ(segment.Get(0), static_cast(0)); ASSERT_EQ(input_stream_opens, 1); @@ -220,7 +221,8 @@ TEST_F(BlockCacheTest, TestRetainedPagesAreReusedByLaterReader) { retained_pool.reset(); ASSERT_TRUE(retained_pool_ref.expired()); { - BlockCache second_reader("test-backend", file_path, supplier, cache_manager, pool_); + BlockCache second_reader("test-backend", /*cache_namespace_owner=*/nullptr, file_path, + supplier, cache_manager, pool_); ASSERT_OK_AND_ASSIGN(MemorySegment cached_segment, GetBlock(0, block_size, &second_reader)); ASSERT_EQ(cached_segment.Get(0), static_cast(0)); ASSERT_EQ(input_stream_opens, 1); diff --git a/src/paimon/core/global_index/global_index_file_manager.h b/src/paimon/core/global_index/global_index_file_manager.h index a2ea66f56..65d0b78ee 100644 --- a/src/paimon/core/global_index/global_index_file_manager.h +++ b/src/paimon/core/global_index/global_index_file_manager.h @@ -22,6 +22,8 @@ #include #include +#include "fmt/format.h" +#include "paimon/common/global_index/cache_namespace_provider.h" #include "paimon/common/utils/uuid.h" #include "paimon/core/index/index_path_factory.h" #include "paimon/fs/file_system.h" @@ -30,11 +32,17 @@ namespace paimon { /// Helper class for managing global index files. -class GlobalIndexFileManager : public GlobalIndexFileReader, public GlobalIndexFileWriter { +class GlobalIndexFileManager : public GlobalIndexFileReader, + public GlobalIndexFileWriter, + public CacheNamespaceProvider { public: GlobalIndexFileManager(const std::shared_ptr& fs, const std::shared_ptr& path_factory) - : GlobalIndexFileReader(CacheNamespaceFor(fs)), fs_(fs), path_factory_(path_factory) {} + : fs_(fs), path_factory_(path_factory) {} + + std::string CacheNamespace() const override { + return fmt::format("filesystem:{}", fmt::ptr(fs_.get())); + } Result> GetInputStream( const std::string& file_path) const override { diff --git a/src/paimon/core/table/source/primary_key_sorted_index_scan.cpp b/src/paimon/core/table/source/primary_key_sorted_index_scan.cpp index f40382052..df6ef273a 100644 --- a/src/paimon/core/table/source/primary_key_sorted_index_scan.cpp +++ b/src/paimon/core/table/source/primary_key_sorted_index_scan.cpp @@ -30,6 +30,7 @@ #include "arrow/c/bridge.h" #include "arrow/c/helpers.h" #include "fmt/format.h" +#include "paimon/common/global_index/cache_namespace_provider.h" #include "paimon/common/types/data_field.h" #include "paimon/common/utils/arrow/status_utils.h" #include "paimon/common/utils/scope_guard.h" @@ -507,11 +508,14 @@ Result PrimaryKeySortedIndexScan::Eval } namespace { -class FsGlobalIndexFileReader : public GlobalIndexFileReader { +class FsGlobalIndexFileReader : public GlobalIndexFileReader, public CacheNamespaceProvider { public: explicit FsGlobalIndexFileReader(std::shared_ptr file_system) - : GlobalIndexFileReader(CacheNamespaceFor(file_system)), - file_system_(std::move(file_system)) {} + : file_system_(std::move(file_system)) {} + + std::string CacheNamespace() const override { + return fmt::format("filesystem:{}", fmt::ptr(file_system_.get())); + } Result> GetInputStream( const std::string& file_path) const override { From 959dd42ef0ea7af2140bd62340f8b9bc365b45d7 Mon Sep 17 00:00:00 2001 From: "wangyong.alen" Date: Mon, 31 Aug 2026 14:07:17 -0400 Subject: [PATCH 8/8] test(index): use fresh schema for cache isolation --- .../global_index/btree/lazy_filtered_btree_reader_test.cpp | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/paimon/common/global_index/btree/lazy_filtered_btree_reader_test.cpp b/src/paimon/common/global_index/btree/lazy_filtered_btree_reader_test.cpp index a57420b94..5850f81d4 100644 --- a/src/paimon/common/global_index/btree/lazy_filtered_btree_reader_test.cpp +++ b/src/paimon/common/global_index/btree/lazy_filtered_btree_reader_test.cpp @@ -428,11 +428,12 @@ TEST_F(LazyFilteredBTreeReaderTest, TestBlockCacheReuseAcrossIndexerInstances) { ASSERT_NOK_WITH_MSG(isolated_reader->VisitEqual(literal_1), "unexpected input stream open for cached btree file"); + auto third_schema = CreateArrowSchema(); auto failing_file_reader = std::make_shared(GetGlobalIndexCacheNamespace(first_file_reader)); ASSERT_OK_AND_ASSIGN( std::shared_ptr second_reader, - second_indexer->CreateReader(second_schema.get(), failing_file_reader, all_metas_, pool_)); + second_indexer->CreateReader(third_schema.get(), failing_file_reader, all_metas_, pool_)); ASSERT_OK_AND_ASSIGN(std::shared_ptr second_result, second_reader->VisitEqual(literal_1)); CheckResult(second_result, {0, 1});