perf(scan): reuse index planning resources across queries

include/paimon/cache/cache.h

@@ -42,6 +42,9 @@ class PAIMON_EXPORT CacheKey {
  public:
     static std::shared_ptr<CacheKey> ForPosition(const std::string& file_path, int64_t position,
                                                  int32_t length, bool is_index);
+    static std::shared_ptr<CacheKey> ForPosition(const std::string& cache_namespace,
+                                                 const std::string& file_path, int64_t position,
+                                                 int32_t length, bool is_index);
     static std::shared_ptr<CacheKey> ForKind(const std::string& file_path, int64_t position,
                                              int32_t length, CacheKind kind);
     static std::shared_ptr<CacheKey> ForSnapshotLiveManifestEntries(const std::string& table_path,

src/paimon/common/global_index/btree/btree_global_indexer.cpp

@@ -19,8 +19,12 @@
 #include "paimon/common/global_index/btree/btree_global_indexer.h"
 
 #include <climits>
+#include <list>
+#include <map>
 #include <memory>
+#include <mutex>
 #include <string>
+#include <utility>
 
 #include "arrow/c/bridge.h"
 #include "fmt/format.h"
@@ -45,6 +49,41 @@
 #include "paimon/utils/roaring_bitmap64.h"
 
 namespace paimon {
+namespace {
+
+std::shared_ptr<CacheManager> GetSharedCacheManager(int64_t cache_size,
+                                                    double high_priority_pool_ratio) {
+    using CacheConfig = std::pair<int64_t, double>;
+    struct Entry {
+        std::shared_ptr<CacheManager> manager;
+        std::list<CacheConfig>::iterator lru_position;
+    };
+    static constexpr size_t kMaxConfigurations = 4;
+    static std::mutex mutex;
+    static std::list<CacheConfig> lru_configs;
+    static std::map<CacheConfig, Entry> cache_managers;
+
+    std::lock_guard<std::mutex> lock(mutex);
+    CacheConfig config(cache_size, high_priority_pool_ratio);
+    auto iter = cache_managers.find(config);
+    if (iter != cache_managers.end()) {
+        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<CacheManager>(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 manager;
+}
+
+}  // namespace
+
 Result<std::unique_ptr<BTreeGlobalIndexer>> BTreeGlobalIndexer::Create(
     const std::map<std::string, std::string>& options) {
     // parse cache options
@@ -57,7 +96,8 @@ Result<std::unique_ptr<BTreeGlobalIndexer>> BTreeGlobalIndexer::Create(
         double high_priority_pool_ratio,
         OptionsUtils::GetValueFromMap<double>(options, BtreeDefs::kBtreeIndexHighPriorityPoolRatio,
                                               BtreeDefs::kDefaultBtreeIndexHighPriorityPoolRatio));
-    auto cache_manager = std::make_shared<CacheManager>(cache_size, high_priority_pool_ratio);
+    std::shared_ptr<CacheManager> cache_manager =
+        GetSharedCacheManager(cache_size, high_priority_pool_ratio);
     return std::unique_ptr<BTreeGlobalIndexer>(new BTreeGlobalIndexer(cache_manager, options));
 }
 

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"
@@ -238,16 +239,21 @@ Result<std::shared_ptr<GlobalIndexReader>> LazyFilteredBTreeReader::CreateSingle
         max_key_slice = MemorySlice::Wrap(index_meta->LastKey());
     }
 
-    // Open input stream and create block cache
-    PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<InputStream> input_stream,
-                           file_reader_->GetInputStream(meta.file_path));
-    if (read_buffer_size_) {
-        input_stream = std::make_shared<BufferedInputStream>(
-            input_stream, read_buffer_size_.value(), pool_.get());
-    }
-
-    auto block_cache =
-        std::make_shared<BlockCache>(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<std::shared_ptr<InputStream>> {
+        PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<InputStream> input_stream,
+                               file_reader->GetInputStream(file_path));
+        if (read_buffer_size) {
+            input_stream = std::make_shared<BufferedInputStream>(
+                input_stream, read_buffer_size.value(), pool.get());
+        }
+        return input_stream;
+    };
+    auto block_cache = std::make_shared<BlockCache>(
+        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,

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,11 +70,15 @@ 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<FileSystem>& fs, const std::string& base_path)
         : fs_(fs), base_path_(base_path) {}
 
+    std::string CacheNamespace() const override {
+        return fmt::format("filesystem:{}", fmt::ptr(fs_.get()));
+    }
+
     Result<std::unique_ptr<InputStream>> GetInputStream(
         const std::string& file_path) const override {
         return fs_->Open(file_path);
@@ -84,6 +89,23 @@ class FakeLazyFileReader : public GlobalIndexFileReader {
     std::string base_path_;
 };
 
+class FailingLazyFileReader : public GlobalIndexFileReader, public CacheNamespaceProvider {
+ public:
+    explicit FailingLazyFileReader(std::string cache_namespace)
+        : cache_namespace_(std::move(cache_namespace)) {}
+
+    std::string CacheNamespace() const override {
+        return cache_namespace_;
+    }
+
+    Result<std::unique_ptr<InputStream>> 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 {
  public:
     void SetUp() override {
@@ -377,6 +399,46 @@ TEST_F(LazyFilteredBTreeReaderTest, TestReaderCacheReuse) {
     CheckResult(result2, {3, 4});
 }
 
+TEST_F(LazyFilteredBTreeReaderTest, TestBlockCacheReuseAcrossIndexerInstances) {
+    std::map<std::string, std::string> options = {
+        {BtreeDefs::kBtreeIndexCacheSize, "1MB"},
+        {BtreeDefs::kBtreeIndexHighPriorityPoolRatio, "0.5"}};
+    Literal literal_1(1);
+
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<BTreeGlobalIndexer> first_indexer,
+                         BTreeGlobalIndexer::Create(options));
+    auto first_schema = CreateArrowSchema();
+    auto first_file_reader = std::make_shared<FakeLazyFileReader>(fs_, base_path_);
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<GlobalIndexReader> first_reader,
+        first_indexer->CreateReader(first_schema.get(), first_file_reader, all_metas_, pool_));
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<GlobalIndexResult> first_result,
+                         first_reader->VisitEqual(literal_1));
+    CheckResult(first_result, {0, 1});
+    first_reader.reset();
+    first_indexer.reset();
+
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<BTreeGlobalIndexer> second_indexer,
+                         BTreeGlobalIndexer::Create(options));
+    auto second_schema = CreateArrowSchema();
+    auto isolated_file_reader = std::make_shared<FailingLazyFileReader>("other-backend");
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<GlobalIndexReader> 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 third_schema = CreateArrowSchema();
+    auto failing_file_reader =
+        std::make_shared<FailingLazyFileReader>(GetGlobalIndexCacheNamespace(first_file_reader));
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<GlobalIndexReader> second_reader,
+        second_indexer->CreateReader(third_schema.get(), failing_file_reader, all_metas_, pool_));
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<GlobalIndexResult> second_result,
+                         second_reader->VisitEqual(literal_1));
+    CheckResult(second_result, {0, 1});
+}
+
 // --- Empty files list ---
 
 TEST_F(LazyFilteredBTreeReaderTest, TestEmptyFilesList) {

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 <memory>
+#include <string>
+
+#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<GlobalIndexFileReader>& file_reader) {
+    const auto* provider = dynamic_cast<const CacheNamespaceProvider*>(file_reader.get());
+    if (provider) {
+        return provider->CacheNamespace();
+    }
+    return fmt::format("index-reader:{}", fmt::ptr(file_reader.get()));
+}
+
+}  // namespace paimon

src/paimon/common/io/cache/cache_key.cpp

@@ -65,14 +65,22 @@ class SnapshotLiveManifestEntriesCacheKey : public CacheKey {
 
 std::shared_ptr<CacheKey> CacheKey::ForPosition(const std::string& file_path, int64_t position,
                                                 int32_t length, bool is_index) {
-    return std::make_shared<PositionCacheKey>(file_path, position, length, is_index,
-                                              CacheKind::DEFAULT);
+    return std::make_shared<PositionCacheKey>(/*cache_namespace=*/"", file_path, position, length,
+                                              is_index, CacheKind::DEFAULT);
+}
+
+std::shared_ptr<CacheKey> 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<PositionCacheKey>(cache_namespace, file_path, position, length,
+                                              is_index, CacheKind::DEFAULT);
 }
 
 std::shared_ptr<CacheKey> CacheKey::ForKind(const std::string& file_path, int64_t position,
                                             int32_t length, CacheKind kind) {
-    auto key = std::make_shared<PositionCacheKey>(file_path, position, length,
-                                                  /*is_index=*/false, kind);
+    auto key =
+        std::make_shared<PositionCacheKey>(/*cache_namespace=*/"", file_path, position, length,
+                                           /*is_index=*/false, kind);
     return key;
 }
 
@@ -99,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<std::string>{}(cache_namespace_) + HASH_CONSTANT + (seed << 6) + (seed >> 2);
     seed ^= std::hash<std::string>{}(file_path_) + HASH_CONSTANT + (seed << 6) + (seed >> 2);
     seed ^= std::hash<int64_t>{}(position_) + HASH_CONSTANT + (seed << 6) + (seed >> 2);
     seed ^= std::hash<int32_t>{}(length_) + HASH_CONSTANT + (seed << 6) + (seed >> 2);

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_;

src/paimon/common/io/cache/lru_cache_test.cpp

@@ -400,6 +400,15 @@ TEST_F(LruCacheTest, TestForSnapshotLiveManifestEntries) {
     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);