github-actions[bot] commented on code in PR #68141: URL: https://github.com/apache/doris/pull/68141#discussion_r4040773432
########## fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVCacheManager.java: ########## @@ -0,0 +1,209 @@ +// 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. + +package org.apache.doris.mtmv; + +import org.apache.doris.catalog.Env; +import org.apache.doris.common.Config; +import org.apache.doris.common.ConfigBase.DefaultConfHandler; + +import com.github.benmanes.caffeine.cache.Cache; +import com.github.benmanes.caffeine.cache.Caffeine; +import com.github.benmanes.caffeine.cache.stats.CacheStats; +import com.google.common.annotations.VisibleForTesting; + +import java.lang.reflect.Field; +import java.time.Duration; +import java.util.Collections; +import java.util.List; +import java.util.Objects; +import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; + +/** + * FE-local cache manager for materialized view cache. + */ +public class MTMVCacheManager { + + private final Object swapLock = new Object(); + private volatile Cache<Key, MTMVCache> caches; + + public MTMVCacheManager() { + caches = build(Config.mtmv_cache_manage_num, Config.expire_mtmv_cache_in_fe_second); + } + + public MTMVCache getIfPresent(long mtmvId, boolean guarded) { + return caches.getIfPresent(new Key(mtmvId, guarded)); + } + + public void put(long mtmvId, boolean guarded, MTMVCache cache) { + Objects.requireNonNull(cache, "mtmv cache to publish must not be null"); + synchronized (swapLock) { + caches.put(new Key(mtmvId, guarded), cache); + } + } + + public void invalidate(long mtmvId) { + synchronized (swapLock) { + caches.invalidate(new Key(mtmvId, true)); + caches.invalidate(new Key(mtmvId, false)); + } + } + + public void invalidateAll() { + synchronized (swapLock) { + caches.invalidateAll(); + } + } + + public long size() { + return caches.estimatedSize(); + } + + /** False when the live maximum is 0, i.e. every put would be discarded immediately. */ + public boolean isEnabled() { + return caches.policy().eviction().map(eviction -> eviction.getMaximum() > 0).orElse(true); + } + + public Snapshot snapshot() { + Cache<Key, MTMVCache> current = caches; + CacheStats s = current.stats(); + return new Snapshot(current.estimatedSize(), s.hitCount(), s.missCount(), + s.evictionCount(), s.hitRate()); + } + + /** + * Snapshot for SHOW PROC '/mtmv_cache/hot'. Ordered by most-recently-accessed first when + * expireAfterAccess is enabled; falls back to iteration order with idleMs=-1 otherwise. + */ + public List<HotEntry> hotEntries(int limit) { + if (limit <= 0) { + return Collections.emptyList(); + } + Cache<Key, MTMVCache> current = caches; + return current.policy().expireAfterAccess() + .map(exp -> exp.youngest(limit).keySet().stream() Review Comment: [P2] Project lightweight hot-entry metadata without retaining the values. In pinned Caffeine 3.2.3, `youngest(int)` materializes a `LinkedHashMap<K,V>` snapshot (the limiter inserts `entry.getValue()`), so this read temporarily makes up to `min(limit, live entries)` full `MTMVCache` plan/`StructInfo` graphs strongly reachable even though the cache uses `softValues()`; the default limit is 500 and it can be raised. The detached key snapshot then queries live `ageOf`, so concurrent invalidation or eviction can emit `IdleMs=-1` while expiration is enabled. Use the `youngest(stream -> ...)` overload with a short metadata-only projection and resolve names afterward; either omit an entry whose live age disappeared, or upgrade before deriving age from `CacheEntry` timestamps because [upstream commit `ca49c08`](https://github.com/ben-manes/caffeine/commit/ca49c08b5dcfad011ba254699f64208eb1e1d2d3) fixes their remaining-duration calculation in 3.2.4. ########## fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVCacheManagerTest.java: ########## @@ -0,0 +1,256 @@ +// 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. + +package org.apache.doris.mtmv; + +import org.apache.doris.common.Config; +import org.apache.doris.common.jmockit.Deencapsulation; +import org.apache.doris.mtmv.MTMVCacheManager.HotEntry; +import org.apache.doris.mtmv.MTMVCacheManager.Key; +import org.apache.doris.mtmv.MTMVCacheManager.Snapshot; + +import com.github.benmanes.caffeine.cache.Cache; +import com.github.benmanes.caffeine.cache.Policy; +import com.github.benmanes.caffeine.cache.stats.CacheStats; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.mockito.Mockito; + +import java.util.Collections; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.ConcurrentHashMap; + +public class MTMVCacheManagerTest { + + @Test + public void testPutGetInvalidate() { + MTMVCacheManager manager = new MTMVCacheManager(); + MTMVCache cacheGuarded = Mockito.mock(MTMVCache.class); + MTMVCache cacheUnguarded = Mockito.mock(MTMVCache.class); + manager.put(1L, true, cacheGuarded); + manager.put(1L, false, cacheUnguarded); + Assertions.assertSame(cacheGuarded, manager.getIfPresent(1L, true)); + Assertions.assertSame(cacheUnguarded, manager.getIfPresent(1L, false)); + Assertions.assertEquals(2L, manager.size()); + + manager.invalidate(1L); + Assertions.assertNull(manager.getIfPresent(1L, true)); + Assertions.assertNull(manager.getIfPresent(1L, false)); + Assertions.assertEquals(0L, manager.size()); + } + + @Test + public void testPutRejectsNull() { + MTMVCacheManager manager = new MTMVCacheManager(); + Assertions.assertThrows(NullPointerException.class, () -> manager.put(1L, true, null)); + } + + @Test + public void testDifferentMtmvsAreIndependent() { + MTMVCacheManager manager = new MTMVCacheManager(); + MTMVCache c1 = Mockito.mock(MTMVCache.class); + MTMVCache c2 = Mockito.mock(MTMVCache.class); + manager.put(1L, true, c1); + manager.put(2L, true, c2); + manager.invalidate(1L); + Assertions.assertNull(manager.getIfPresent(1L, true)); + Assertions.assertSame(c2, manager.getIfPresent(2L, true)); + } + + @Test + public void testSnapshotReportsHitAndMiss() { + MTMVCacheManager manager = new MTMVCacheManager(); + MTMVCache c1 = Mockito.mock(MTMVCache.class); + manager.put(1L, true, c1); + manager.getIfPresent(1L, true); + manager.getIfPresent(1L, false); + Snapshot snap = manager.snapshot(); + Assertions.assertEquals(1L, snap.size); + Assertions.assertTrue(snap.hitCount >= 1); + Assertions.assertTrue(snap.missCount >= 1); + } + + @Test + public void testHotEntriesHonorsLimit() { + MTMVCacheManager manager = new MTMVCacheManager(); + MTMVCache c = Mockito.mock(MTMVCache.class); + for (int i = 0; i < 5; i++) { + manager.put(i, true, c); + } + List<HotEntry> hot = manager.hotEntries(3); + Assertions.assertEquals(3, hot.size()); + for (HotEntry e : hot) { + Assertions.assertTrue(e.idleMs >= 0, + "idleMs should be >= 0 when expireAfterAccess is set, got " + e.idleMs); + } + } + + @Test + public void testHotEntriesEmptyForZeroOrNegativeLimit() { + MTMVCacheManager manager = new MTMVCacheManager(); + MTMVCache c = Mockito.mock(MTMVCache.class); + manager.put(1L, true, c); + Assertions.assertTrue(manager.hotEntries(0).isEmpty()); + Assertions.assertTrue(manager.hotEntries(-1).isEmpty()); + } + + @Test + public void testInvalidateAll() { + MTMVCacheManager manager = new MTMVCacheManager(); + MTMVCache c = Mockito.mock(MTMVCache.class); + manager.put(1L, true, c); + manager.put(2L, false, c); + manager.invalidateAll(); + Assertions.assertEquals(0L, manager.size()); + } + + // updateConfig() can swap the field between the two reads. + @Test + public void testSnapshotReadsOneCacheInstance() { + int originalMaxSize = Config.mtmv_cache_manage_num; + try { + MTMVCacheManager manager = new MTMVCacheManager(); + Cache<Key, MTMVCache> original = mockCache(); + Mockito.when(original.estimatedSize()).thenReturn(7L); + Mockito.when(original.asMap()).thenReturn(new ConcurrentHashMap<>()); + Mockito.when(original.stats()).thenAnswer(invocation -> { + // The swap lands while snapshot() is between its reads. + Config.mtmv_cache_manage_num = 0; + manager.updateConfig(); + return CacheStats.of(3L, 1L, 0L, 0L, 0L, 2L, 0L); + }); + Deencapsulation.setField(manager, "caches", original); + + Snapshot snap = manager.snapshot(); + + Assertions.assertEquals(7L, snap.size); + Assertions.assertEquals(3L, snap.hitCount); + Assertions.assertEquals(1L, snap.missCount); + Assertions.assertEquals(2L, snap.evictionCount); + } finally { + Config.mtmv_cache_manage_num = originalMaxSize; + } + } + + @Test + public void testHotEntriesReadsOneCacheInstance() { + int originalMaxSize = Config.mtmv_cache_manage_num; + try { + MTMVCacheManager manager = new MTMVCacheManager(); + Cache<Key, MTMVCache> original = mockCache(); + Policy<Key, MTMVCache> policy = mockPolicy(); + Mockito.when(policy.expireAfterAccess()).thenReturn(Optional.empty()); + Mockito.when(original.asMap()).thenReturn( + new ConcurrentHashMap<>(Collections.singletonMap(new Key(1L, true), + Mockito.mock(MTMVCache.class)))); + Mockito.when(original.policy()).thenAnswer(invocation -> { + // Swapping to a disabled cache would leave the fresh instance empty. + Config.mtmv_cache_manage_num = 0; + manager.updateConfig(); + return policy; + }); + Deencapsulation.setField(manager, "caches", original); + + List<HotEntry> hot = manager.hotEntries(10); + + Assertions.assertEquals(1, hot.size()); + Assertions.assertEquals(1L, hot.get(0).mtmvId); + } finally { + Config.mtmv_cache_manage_num = originalMaxSize; + } + } + + @Test + public void testIsEnabledFollowsLiveMaxSize() { + int originalMaxSize = Config.mtmv_cache_manage_num; + try { + Config.mtmv_cache_manage_num = 10; + MTMVCacheManager manager = new MTMVCacheManager(); + Assertions.assertTrue(manager.isEnabled()); + + Config.mtmv_cache_manage_num = 0; + manager.updateConfig(); + Assertions.assertFalse(manager.isEnabled()); + } finally { + Config.mtmv_cache_manage_num = originalMaxSize; + } + } + + @SuppressWarnings("unchecked") + private static Cache<Key, MTMVCache> mockCache() { + return Mockito.mock(Cache.class); + } + + @SuppressWarnings("unchecked") + private static Policy<Key, MTMVCache> mockPolicy() { + return Mockito.mock(Policy.class); + } + + @Test + public void testZeroMaxSizeDisablesCacheInsteadOfUnbounding() { + int originalMaxSize = Config.mtmv_cache_manage_num; + try { + Config.mtmv_cache_manage_num = 0; + MTMVCacheManager manager = new MTMVCacheManager(); + for (int i = 0; i < 5; i++) { + manager.put(i, true, Mockito.mock(MTMVCache.class)); + } + manager.getCachesForTest().cleanUp(); + Assertions.assertEquals(0L, manager.size()); + } finally { + Config.mtmv_cache_manage_num = originalMaxSize; + } + } + + @Test + public void testUpdateConfigShrinksToNewMaxSize() { + int originalMaxSize = Config.mtmv_cache_manage_num; + try { + Config.mtmv_cache_manage_num = 10; + MTMVCacheManager manager = new MTMVCacheManager(); + for (int i = 0; i < 10; i++) { + manager.put(i, true, Mockito.mock(MTMVCache.class)); Review Comment: [P2] Keep these values observably reachable while testing the size-policy shrink. The production manager always uses `softValues()`, and every mock created here loses its last test-owned strong reference as soon as `put()` returns. A memory-pressure GC before `cleanUp()` can therefore make `size() == 10` fail; if collection happens later, `size() <= 2` can pass without proving that the new maximum evicted anything. Retain the mocks in a list, assert the exact post-shrink size of 2, and call `Reference.reachabilityFence(values)` after the assertions. Apply the same liveness pattern to other new tests that require an entry after their local's last use, such as `testSnapshotReportsHitAndMiss` and `testHotEntriesHonorsLimit`; lexical scope alone is not a reachability guarantee. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
