feat(cache): cache typed objects and reuse vector index readers

docs/source/user_guide/manifest_cache.rst

@@ -102,7 +102,8 @@ concurrent cache misses.
 
 Cache implementations should use ``CacheValue::GetMemoryUsage()`` for admission
 and eviction accounting. For manifest IPC, this reports the retained Arrow
-buffer capacity, including unused space from growth. ``GetSegment().Size()``
-continues to describe the valid IPC byte length. The built-in ``LruCache`` uses
+buffer capacity, including unused space from growth.
+``CacheValue::Get<const MemorySegment>(value)->Size()`` describes the valid IPC
+byte length. The built-in ``LruCache`` uses
 the memory usage value; existing cache values without an explicit allocation
 size continue to charge their segment length.

include/paimon/cache/cache.h

@@ -18,11 +18,13 @@
 
 #pragma once
 
+#include <algorithm>
 #include <cstddef>
 #include <cstdint>
 #include <functional>
 #include <memory>
 #include <string>
+#include <utility>
 
 #include "paimon/memory/memory_segment.h"
 #include "paimon/result.h"
@@ -36,6 +38,7 @@ enum class CacheKind {
     MANIFEST,
     DATA_FILE_FOOTER,
     SNAPSHOT_LIVE_MANIFEST,
+    INDEX_READER,
 };
 
 class PAIMON_EXPORT CacheKey {
@@ -44,6 +47,8 @@ class PAIMON_EXPORT CacheKey {
                                                  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> ForIndexReader(const std::string& uri,
+                                                    const std::string& column_name);
 
  public:
     virtual ~CacheKey() = default;
@@ -88,28 +93,84 @@ class PAIMON_EXPORT Cache {
 
 class PAIMON_EXPORT CacheValue {
  public:
-    /// Charge the segment's logical size when no separate allocation size is supplied.
-    CacheValue(const MemorySegment& segment, CacheCallback callback);
+    virtual ~CacheValue();
 
-    /// @param memory_usage Retained allocation size in bytes, including unused capacity.
-    ///                     The charge is at least the segment's logical size.
-    CacheValue(const MemorySegment& segment, CacheCallback callback, int64_t memory_usage);
+    /// Returns the typed payload, retaining its owner; null for a null value or type mismatch.
+    template <typename T>
+    static std::shared_ptr<T> Get(const std::shared_ptr<CacheValue>& value);
 
-    ~CacheValue();
-
-    const MemorySegment& GetSegment() const;
-
-    /// Bytes to charge against the cache budget. May exceed GetSegment().Size().
+    /// Bytes charged against the cache budget.
     int64_t GetMemoryUsage() const;
 
     void OnEvict(const std::shared_ptr<CacheKey>& key) const;
 
-    bool operator==(const CacheValue& other) const;
+    virtual bool operator==(const CacheValue& other) const = 0;
+
+ protected:
+    CacheValue(CacheCallback callback, int64_t memory_usage);
 
  private:
-    MemorySegment segment_;
     CacheCallback callback_;
     int64_t memory_usage_;
 };
 
+template <typename T>
+class PAIMON_EXPORT TypedCacheValue : public CacheValue {
+ protected:
+    using CacheValue::CacheValue;
+
+ private:
+    friend class CacheValue;
+    virtual T* GetValuePointer() = 0;
+};
+
+template <typename T>
+std::shared_ptr<T> CacheValue::Get(const std::shared_ptr<CacheValue>& value) {
+    auto* typed_value = dynamic_cast<TypedCacheValue<T>*>(value.get());
+    if (typed_value) {
+        if (T* payload = typed_value->GetValuePointer()) {
+            return std::shared_ptr<T>(value, payload);
+        }
+    }
+    return nullptr;
+}
+
+class PAIMON_EXPORT MemorySegmentCacheValue final : public TypedCacheValue<const MemorySegment> {
+ public:
+    MemorySegmentCacheValue(const MemorySegment& segment, CacheCallback callback);
+
+    /// Charges retained allocation size, at least the segment's logical size.
+    MemorySegmentCacheValue(const MemorySegment& segment, CacheCallback callback,
+                            int64_t memory_usage);
+
+    bool operator==(const CacheValue& other) const override;
+
+ private:
+    const MemorySegment* GetValuePointer() override {
+        return &segment_;
+    }
+
+    MemorySegment segment_;
+};
+
+template <typename T>
+class PAIMON_EXPORT ObjectCacheValue final : public TypedCacheValue<T> {
+ public:
+    ObjectCacheValue(std::shared_ptr<T> value, CacheCallback callback, int64_t memory_usage)
+        : TypedCacheValue<T>(std::move(callback), std::max<int64_t>(1, memory_usage)),
+          value_(std::move(value)) {}
+
+    bool operator==(const CacheValue& other) const override {
+        const auto* rhs = dynamic_cast<const ObjectCacheValue<T>*>(&other);
+        return rhs && value_ == rhs->value_;
+    }
+
+ private:
+    T* GetValuePointer() override {
+        return value_.get();
+    }
+
+    std::shared_ptr<T> value_;
+};
+
 }  // namespace paimon

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

@@ -24,20 +24,11 @@
 
 namespace paimon {
 
-CacheValue::CacheValue(const MemorySegment& segment, CacheCallback callback)
-    : CacheValue(segment, std::move(callback), segment.Size()) {}
-
-CacheValue::CacheValue(const MemorySegment& segment, CacheCallback callback, int64_t memory_usage)
-    : segment_(segment),
-      callback_(std::move(callback)),
-      memory_usage_(std::max<int64_t>(segment.Size(), memory_usage)) {}
+CacheValue::CacheValue(CacheCallback callback, int64_t memory_usage)
+    : callback_(std::move(callback)), memory_usage_(memory_usage) {}
 
 CacheValue::~CacheValue() = default;
 
-const MemorySegment& CacheValue::GetSegment() const {
-    return segment_;
-}
-
 int64_t CacheValue::GetMemoryUsage() const {
     return memory_usage_;
 }
@@ -48,11 +39,19 @@ void CacheValue::OnEvict(const std::shared_ptr<CacheKey>& key) const {
     }
 }
 
-bool CacheValue::operator==(const CacheValue& other) const {
-    if (this == &other) {
-        return true;
-    }
-    return segment_ == other.segment_;
+MemorySegmentCacheValue::MemorySegmentCacheValue(const MemorySegment& segment,
+                                                 CacheCallback callback)
+    : MemorySegmentCacheValue(segment, std::move(callback), segment.Size()) {}
+
+MemorySegmentCacheValue::MemorySegmentCacheValue(const MemorySegment& segment,
+                                                 CacheCallback callback, int64_t memory_usage)
+    : TypedCacheValue<const MemorySegment>(std::move(callback),
+                                           std::max<int64_t>(segment.Size(), memory_usage)),
+      segment_(segment) {}
+
+bool MemorySegmentCacheValue::operator==(const CacheValue& other) const {
+    const auto* rhs = dynamic_cast<const MemorySegmentCacheValue*>(&other);
+    return rhs && segment_ == rhs->segment_;
 }
 
 }  // namespace paimon

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

@@ -18,6 +18,8 @@
 
 #include "paimon/common/io/cache/cache_key.h"
 
+#include "fmt/format.h"
+
 namespace paimon {
 
 std::shared_ptr<CacheKey> CacheKey::ForPosition(const std::string& file_path, int64_t position,
@@ -33,6 +35,13 @@ std::shared_ptr<CacheKey> CacheKey::ForKind(const std::string& file_path, int64_
     return key;
 }
 
+std::shared_ptr<CacheKey> CacheKey::ForIndexReader(const std::string& uri,
+                                                   const std::string& column_name) {
+    std::string identity =
+        fmt::format("index-reader:{}:{}:{}:{}", uri.size(), uri, column_name.size(), column_name);
+    return ForKind(identity, 0, 0, CacheKind::INDEX_READER);
+}
+
 std::shared_ptr<CacheKey> SnapshotLiveManifestEntriesCacheKey::ForExplicit(
     const std::string& table_path, const std::string& branch, int32_t bucket) {
     return std::shared_ptr<CacheKey>(new SnapshotLiveManifestEntriesCacheKey(

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

@@ -28,10 +28,14 @@ Result<MemorySegment> CacheManager::GetPage(
     auto supplier =
         [&](const std::shared_ptr<CacheKey>& key) -> Result<std::shared_ptr<CacheValue>> {
         PAIMON_ASSIGN_OR_RAISE(MemorySegment segment, reader(key));
-        return std::make_shared<CacheValue>(segment, std::move(eviction_callback));
+        return std::make_shared<MemorySegmentCacheValue>(segment, std::move(eviction_callback));
     };
     PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<CacheValue> cache_value, cache->Get(key, supplier));
-    return cache_value->GetSegment();
+    auto segment = CacheValue::Get<const MemorySegment>(cache_value);
+    if (!segment) {
+        return Status::Invalid("Page cache returned an invalid value type");
+    }
+    return *segment;
 }
 
 void CacheManager::InvalidPage(const std::shared_ptr<CacheKey>& key) {