refactor(scan): simplify snapshot cache eviction

Author: gripleaf

docs/source/api/scan.rst

@@ -81,14 +81,17 @@ progress, executors and scan metrics remain independent. ``SetTableSchema()`` ke
 behavior: on main it bypasses the shared schema cache; on other branches it is ignored.
 
 Successfully loaded snapshots are cached by their full file paths, with an LRU limit of 20 entries
-per resource object. Entries expire after 10 minutes without access or 30 minutes since their last
-cache write, whichever comes first. Repeated reads do not extend the write deadline. Expiration is
-checked during cache operations, without a background refresh thread. Snapshot caching is enabled
-by these resources; ordinary snapshot managers have no cache unless one is explicitly supplied.
+per resource object. The whole snapshot cache is replaced on the first access after it reaches
+30 minutes of age. Reads and writes do not extend this deadline; recently added entries are
+discarded along with older ones. There is no per-entry TTL or background refresh thread.
+Snapshot caching is enabled by these resources; ordinary snapshot managers have no cache unless
+one is explicitly supplied.
 Latest and earliest snapshot discovery and existence checks still use the catalog or file system
 as appropriate. Read or parse failures are not cached.
 
-Snapshot deletion through a manager invalidates the corresponding entry in its injected cache.
+Snapshot deletion through a manager replaces its entire injected snapshot cache before and after
+the deletion attempt. Explicit invalidation also discards the whole cache. In-flight loads can
+finish against the old cache but cannot populate the replacement used by subsequent callers.
 Historical commit lookup, timestamp searches, and retained-snapshot publication checks read files
 directly so cached metadata cannot hide missing or replaced files. Other clients' deletions do not
 immediately invalidate this process's cache. A cached snapshot can remain available after its

include/paimon/table/source/table_scan_resources.h

@@ -36,14 +36,15 @@ class TableScanResourcesAccess;
 /// the latest schema ID and chooses its own snapshot. Schema versions and schema-derived resources
 /// are retained for this object's lifetime without an entry limit. Statistics evolutions are also
 /// retained without an entry limit; each caches up to 64 dense-field mappings. Snapshots use a
-/// separate 20-entry LRU cache, expiring after 10 minutes without access or 30 minutes since the
-/// last cache write. Eviction releases cache references; active scans retain the metadata they
-/// use. Snapshot and dense-field cache limits count entries, not bytes. Latest snapshot discovery
-/// and existence checks still query the file system. A cached snapshot does not pin its metadata or
-/// data files. The caller must recreate this object after fast-forward, dropping and recreating a
-/// table or branch, or changing the file system's access configuration. Fast-forward can replace
-/// schema and snapshot contents under existing IDs; there is no immediate cross-client
-/// invalidation. Use the new resources for subsequent scans.
+/// separate 20-entry LRU cache. On access, the whole snapshot cache is replaced once it is
+/// 30 minutes old; reads and writes do not extend that deadline. Eviction releases cache
+/// references; active scans retain the metadata they use. Snapshot and dense-field cache limits
+/// count entries, not bytes. Latest snapshot discovery and existence checks still query the file
+/// system. A cached snapshot does not pin its metadata or data files. The caller must recreate this
+/// object after fast-forward, dropping and recreating a table or branch, or changing the file
+/// system's access configuration. Fast-forward can replace schema and snapshot contents under
+/// existing IDs; there is no immediate cross-client invalidation. Use the new resources for
+/// subsequent scans.
 ///
 /// Concurrent scans may share these resources. The supplied file system and metadata memory pool
 /// must support concurrent use. Filters, executors, scan progress and scan metrics are not shared.

src/paimon/common/utils/generic_lru_cache.h

@@ -38,10 +38,9 @@ namespace paimon {
 /// A generic LRU cache with support for weight-based eviction, time-based expiration,
 /// and removal callbacks.
 ///
-/// Uses std::list + unordered_map for O(1) lookup and LRU updates:
+/// Uses std::list + unordered_map for O(1) get/put/evict:
 /// - list stores entries in LRU order (most recently used at front)
 /// - map stores key -> list::iterator for O(1) lookup
-/// When write expiration is enabled, insertion and replacement also scan for expired entries.
 ///
 /// @tparam K Key type
 /// @tparam V Value type
@@ -57,13 +56,12 @@ class GenericLruCache {
     enum class RemovalCause {
         EXPLICIT,  // Removed by Invalidate() or InvalidateAll()
         SIZE,      // Evicted because total weight exceeded max_weight
-        EXPIRED,   // Evicted because the entry expired
+        EXPIRED,   // Evicted because the entry expired (expireAfterAccess)
         REPLACED   // Replaced by a new value for the same key via Put()
     };
 
     using WeighFunc = std::function<int64_t(const K&, const V&)>;
     using RemovalCallback = std::function<void(const K&, const V&, RemovalCause)>;
-    using Clock = std::function<std::chrono::steady_clock::time_point()>;
 
     /// Configuration options for the cache.
     struct Options {
@@ -80,17 +78,9 @@ class GenericLruCache {
         /// Callback invoked when an entry is removed (evicted, invalidated, or replaced).
         /// If nullptr, no callback is invoked.
         RemovalCallback removal_callback = nullptr;
-
-        /// Time in milliseconds after insertion or Put() before an entry expires.
-        /// Reads do not extend this deadline. A value < 0 disables expiration.
-        int64_t expire_after_write_ms = -1;
     };
 
-    explicit GenericLruCache(Options options)
-        : GenericLruCache(std::move(options), std::chrono::steady_clock::now) {}
-
-    GenericLruCache(Options options, Clock clock)
-        : options_(std::move(options)), clock_(std::move(clock)) {}
+    explicit GenericLruCache(Options options) : options_(std::move(options)) {}
 
     /// Look up a key in the cache. On hit, promotes the entry to the front (most recently
     /// used) and updates its access time. Returns std::nullopt on miss or if the entry
@@ -102,17 +92,13 @@ class GenericLruCache {
 
     /// Look up a key. On miss, load via the supplier, insert into cache, and return.
     /// If the supplier returns an error, the error is propagated and nothing is cached.
-    /// Invalidation during a load prevents its result from being cached, but the result is
-    /// still returned to the caller whose load was already in progress.
     Result<V> Get(const K& key, std::function<Result<V>(const K&)> supplier) {
-        uint64_t generation;
         {
             std::unique_lock<std::shared_mutex> lock(mutex_);
             auto cached = FindPromoteOrExpire(key);
             if (cached.has_value()) {
                 return std::move(cached.value());
             }
-            generation = generation_;
         }
 
         // Cache miss: load via supplier outside the lock
@@ -123,9 +109,6 @@ class GenericLruCache {
         }
 
         std::unique_lock<std::shared_mutex> lock(mutex_);
-        if (generation != generation_) {
-            return value;
-        }
         // Double-check: another thread may have inserted while we were loading
         auto cached = FindPromoteOrExpire(key);
         if (cached.has_value()) {
@@ -152,7 +135,6 @@ class GenericLruCache {
         auto it = lru_map_.find(key);
         if (it != lru_map_.end()) {
             if (ValuesEqual(it->second->value, value)) {
-                it->second->last_write_time = clock_();
                 Promote(it->second);
                 return Status::OK();
             }
@@ -168,7 +150,6 @@ class GenericLruCache {
     /// Remove a specific entry. Invokes the EXPLICIT removal callback if the key exists.
     void Invalidate(const K& key) {
         std::unique_lock<std::shared_mutex> lock(mutex_);
-        ++generation_;
         auto it = lru_map_.find(key);
         if (it != lru_map_.end()) {
             RemoveEntry(it->second, RemovalCause::EXPLICIT);
@@ -178,7 +159,6 @@ class GenericLruCache {
     /// Remove all entries. Each entry's EXPLICIT removal callback is invoked.
     void InvalidateAll() {
         std::unique_lock<std::shared_mutex> lock(mutex_);
-        ++generation_;
         while (!lru_list_.empty()) {
             RemoveEntry(std::prev(lru_list_.end()), RemovalCause::EXPLICIT);
         }
@@ -208,7 +188,6 @@ class GenericLruCache {
         V value;
         int64_t weight;
         std::chrono::steady_clock::time_point last_access_time;
-        std::chrono::steady_clock::time_point last_write_time;
     };
 
     using EntryList = std::list<CacheEntry>;
@@ -223,7 +202,7 @@ class GenericLruCache {
             return std::nullopt;
         }
         auto list_it = it->second;
-        if (IsExpired(*list_it, clock_())) {
+        if (IsExpired(list_it->last_access_time)) {
             RemoveEntry(list_it, RemovalCause::EXPIRED);
             return std::nullopt;
         }
@@ -233,14 +212,14 @@ class GenericLruCache {
 
     /// Move an entry to the front of the LRU list and update its access time.
     void Promote(typename EntryList::iterator list_it) {
-        list_it->last_access_time = clock_();
+        list_it->last_access_time = std::chrono::steady_clock::now();
         lru_list_.splice(lru_list_.begin(), lru_list_, list_it);
     }
 
     /// Insert a new entry at the front of the LRU list.
     void InsertEntry(const K& key, V value, int64_t weight) {
-        auto now = clock_();
-        lru_list_.emplace_front(CacheEntry{key, std::move(value), weight, now, now});
+        lru_list_.emplace_front(
+            CacheEntry{key, std::move(value), weight, std::chrono::steady_clock::now()});
         lru_map_[key] = lru_list_.begin();
         current_weight_ += weight;
     }
@@ -271,8 +250,7 @@ class GenericLruCache {
         V old_value = std::move(list_it->value);
         list_it->value = std::move(new_value);
         list_it->weight = new_weight;
-        list_it->last_access_time = clock_();
-        list_it->last_write_time = list_it->last_access_time;
+        list_it->last_access_time = std::chrono::steady_clock::now();
         current_weight_ += new_weight;
         lru_list_.splice(lru_list_.begin(), lru_list_, list_it);
 
@@ -289,33 +267,25 @@ class GenericLruCache {
         InvokeCallback(key, value, cause);
     }
 
-    /// Evict expired entries, then evict by weight if still over capacity.
+    /// Evict expired entries from the tail, then evict by weight if still over capacity.
     void EvictIfNeeded() {
         EvictExpired();
         while (current_weight_ > options_.max_weight && !lru_list_.empty()) {
             RemoveEntry(std::prev(lru_list_.end()), RemovalCause::SIZE);
         }
     }
 
-    /// Write deadlines are not ordered by LRU position, so write expiration requires a scan.
-    /// With only access expiration, the oldest access time is at the tail.
+    /// Evict expired entries from the tail of the LRU list.
+    /// Since the tail has the oldest access time, we can stop as soon as we find
+    /// a non-expired entry.
     void EvictExpired() {
-        auto now = clock_();
-        if (options_.expire_after_write_ms >= 0) {
-            for (auto it = lru_list_.begin(); it != lru_list_.end();) {
-                auto current = it++;
-                if (IsExpired(*current, now)) {
-                    RemoveEntry(current, RemovalCause::EXPIRED);
-                }
-            }
-            return;
-        }
         if (options_.expire_after_access_ms < 0) {
             return;
         }
+        auto now = std::chrono::steady_clock::now();
         while (!lru_list_.empty()) {
             auto it = std::prev(lru_list_.end());
-            if (!IsExpired(*it, now)) {
+            if (!IsExpired(it->last_access_time, now)) {
                 break;
             }
             RemoveEntry(it, RemovalCause::EXPIRED);
@@ -330,16 +300,16 @@ class GenericLruCache {
         return 1;
     }
 
-    bool IsExpired(const CacheEntry& entry,
-                   const std::chrono::steady_clock::time_point& now) const {
-        auto access_elapsed =
-            std::chrono::duration_cast<std::chrono::milliseconds>(now - entry.last_access_time);
-        auto write_elapsed =
-            std::chrono::duration_cast<std::chrono::milliseconds>(now - entry.last_write_time);
-        return (options_.expire_after_access_ms >= 0 &&
-                access_elapsed.count() >= options_.expire_after_access_ms) ||
-               (options_.expire_after_write_ms >= 0 &&
-                write_elapsed.count() >= options_.expire_after_write_ms);
+    /// Check if an entry has expired based on its last access time.
+    bool IsExpired(
+        const std::chrono::steady_clock::time_point& last_access_time,
+        const std::chrono::steady_clock::time_point& now = std::chrono::steady_clock::now()) const {
+        if (options_.expire_after_access_ms < 0) {
+            return false;
+        }
+        auto elapsed =
+            std::chrono::duration_cast<std::chrono::milliseconds>(now - last_access_time);
+        return elapsed.count() >= options_.expire_after_access_ms;
     }
 
     /// Invoke the removal callback if one is configured.
@@ -350,9 +320,6 @@ class GenericLruCache {
     }
 
     Options options_;
-    Clock clock_;
-    // A single generation avoids retaining per-key state for loads that have not completed.
-    uint64_t generation_ = 0;
     int64_t current_weight_ = 0;
     EntryList lru_list_;
     EntryMap lru_map_;

src/paimon/common/utils/generic_lru_cache_test.cpp

@@ -68,80 +68,6 @@ TEST_F(GenericLruCacheTest, ConstructorAndDefaults) {
 
 // ==================== GetIfPresent ====================
 
-TEST_F(GenericLruCacheTest, WriteDeadlineSurvivesRepeatedAccess) {
-    std::chrono::steady_clock::time_point now{};
-    IntIntCache::Options options;
-    options.expire_after_access_ms = 10;
-    options.expire_after_write_ms = 30;
-    IntIntCache cache(options, [&now]() { return now; });
-    ASSERT_OK(cache.Put(1, 10));
-    for (int32_t i = 0; i < 5; ++i) {
-        now += std::chrono::milliseconds(5);
-        ASSERT_EQ(cache.GetIfPresent(1), 10);
-    }
-    now += std::chrono::milliseconds(5);
-    ASSERT_FALSE(cache.GetIfPresent(1));
-    ASSERT_OK_AND_ASSIGN(int32_t reloaded,
-                         cache.Get(1, [](const int32_t&) -> Result<int32_t> { return 20; }));
-    ASSERT_EQ(reloaded, 20);
-    now += std::chrono::milliseconds(10);
-    ASSERT_FALSE(cache.GetIfPresent(1));
-}
-
-TEST_F(GenericLruCacheTest, PutRefreshesWriteDeadlineForEqualAndDifferentValues) {
-    for (int32_t replacement : {10, 20}) {
-        std::chrono::steady_clock::time_point now{};
-        IntIntCache::Options options;
-        options.expire_after_write_ms = 30;
-        IntIntCache cache(options, [&now]() { return now; });
-        ASSERT_OK(cache.Put(1, 10));
-        now += std::chrono::milliseconds(20);
-        ASSERT_OK(cache.Put(1, replacement));
-        now += std::chrono::milliseconds(20);
-        ASSERT_EQ(cache.GetIfPresent(1), replacement);
-        now += std::chrono::milliseconds(10);
-        ASSERT_FALSE(cache.GetIfPresent(1));
-    }
-}
-
-TEST_F(GenericLruCacheTest, WriteExpiredEntryIsEvictedEvenWhenMostRecentlyUsed) {
-    std::chrono::steady_clock::time_point now{};
-    IntIntCache::Options options;
-    options.max_weight = 2;
-    options.expire_after_write_ms = 30;
-    IntIntCache cache(options, [&now]() { return now; });
-    ASSERT_OK(cache.Put(1, 10));
-    now += std::chrono::milliseconds(20);
-    ASSERT_OK(cache.Put(2, 20));
-    ASSERT_EQ(cache.GetIfPresent(1), 10);
-    now += std::chrono::milliseconds(10);
-    ASSERT_OK(cache.Put(3, 30));
-    ASSERT_FALSE(cache.GetIfPresent(1));
-    ASSERT_EQ(cache.GetIfPresent(2), 20);
-    ASSERT_EQ(cache.GetIfPresent(3), 30);
-}
-
-TEST_F(GenericLruCacheTest, InvalidationDuringLoadPreventsRefill) {
-    for (bool invalidate_all : {false, true}) {
-        IntIntCache cache(IntIntCache::Options{});
-        ASSERT_OK_AND_ASSIGN(int32_t loaded,
-                             cache.Get(1, [&](const int32_t& key) -> Result<int32_t> {
-                                 if (invalidate_all) {
-                                     cache.InvalidateAll();
-                                 } else {
-                                     cache.Invalidate(key);
-                                 }
-                                 return 10;
-                             }));
-        ASSERT_EQ(loaded, 10);
-        ASSERT_FALSE(cache.GetIfPresent(1));
-        ASSERT_OK_AND_ASSIGN(int32_t reloaded,
-                             cache.Get(1, [](const int32_t&) -> Result<int32_t> { return 20; }));
-        ASSERT_EQ(reloaded, 20);
-        ASSERT_EQ(cache.GetIfPresent(1), 20);
-    }
-}
-
 TEST_F(GenericLruCacheTest, GetIfPresentMissAndHit) {
     StringIntCache::Options options;
     StringIntCache cache(options);

src/paimon/core/operation/expire_snapshots_test.cpp

@@ -230,8 +230,7 @@ TEST_F(ExpireSnapshotsTest, TestInvalidInput) {
 }
 
 TEST_F(ExpireSnapshotsTest, CachedRetainedSnapshotDoesNotProvePublication) {
-    auto cache = std::make_shared<SnapshotManager::SnapshotCache>(
-        SnapshotManager::SnapshotCache::Options{/*max_weight=*/20});
+    auto cache = std::make_shared<SnapshotManager::SnapshotCache>();
     auto manager = std::make_shared<SnapshotManager>(fs_, test_data_path_, "main", cache);
     ASSERT_OK(fs_->Mkdirs(manager->SnapshotDirectory()));
     for (int64_t id : {1, 2, 3}) {

src/paimon/core/table/source/table_scan_resources.cpp

@@ -28,18 +28,6 @@
 #include "paimon/core/utils/field_mapping.h"
 
 namespace paimon {
-namespace {
-
-std::shared_ptr<SnapshotManager::SnapshotCache> CreateSnapshotCache() {
-    SnapshotManager::SnapshotCache::Options options;
-    options.max_weight = 20;
-    options.expire_after_access_ms = 10 * 60 * 1000;
-    options.expire_after_write_ms = 30 * 60 * 1000;
-    return std::make_shared<SnapshotManager::SnapshotCache>(options);
-}
-
-}  // namespace
-
 Result<std::shared_ptr<TableScanResources>> TableScanResources::Create(
     const std::string& table_path, const std::shared_ptr<FileSystem>& file_system,
     const std::string& branch, const std::shared_ptr<MemoryPool>& memory_pool) {
@@ -72,8 +60,8 @@ TableScanResources::Impl::Impl(const std::string& path, const std::string& branc
       file_system_(file_system),
       memory_pool_(memory_pool),
       schema_manager_(std::make_shared<SchemaManager>(file_system, path, branch)),
-      snapshot_manager_(
-          std::make_shared<SnapshotManager>(file_system, path, branch, CreateSnapshotCache())) {}
+      snapshot_manager_(std::make_shared<SnapshotManager>(
+          file_system, path, branch, std::make_shared<SnapshotManager::SnapshotCache>())) {}
 
 Status TableScanResources::Impl::Validate(
     const std::string& scan_path, const std::string& scan_branch,