github-actions[bot] commented on code in PR #68141: URL: https://github.com/apache/doris/pull/68141#discussion_r4039910272
########## fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVCacheManager.java: ########## @@ -0,0 +1,225 @@ +// 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 org.apache.doris.common.DdlException; + +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) { Review Comment: [P2] Fail loudly on an impossible null publication. Both production callers already guarantee this value is non-null: refresh guards each sibling before `put()`, and lazy `createRewriteCache()` returns an `MTMVCache` or throws. Silently returning here can hide a broken publication path and leave an absent or stale cache without exposing the invariant violation, contrary to this repository's rule to report/crash on unexpected states rather than continue defensively. Require non-null (or let Caffeine reject it) and remove the null-no-op test. ########## fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVCacheManager.java: ########## @@ -0,0 +1,225 @@ +// 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 org.apache.doris.common.DdlException; + +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) { + if (cache == null) { + return; + } + 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(); + } + + public Snapshot snapshot() { + CacheStats s = caches.stats(); Review Comment: [P2] Snapshot one cache instance for each proc read. `snapshot()` dereferences the volatile `caches` field once for `stats()` and again for `estimatedSize()`, so `updateConfig()` can swap between them and pair the retired cache's counters with the replacement cache's size. `hotEntries()` has the same split in its fallback: it can observe no expiry policy on the old cache, then iterate the new expiry-enabled cache while reporting unordered rows with `IdleMs=-1`. This is distinct from the existing policy-reset/no-expiry threads because one response combines two generations. Capture `Cache<Key, MTMVCache> current = caches` once and derive the whole result from `current`, with a coordinated swap test. ########## fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVCacheManager.java: ########## @@ -0,0 +1,225 @@ +// 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 org.apache.doris.common.DdlException; + +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) { + if (cache == null) { + return; + } + 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(); + } + + public Snapshot snapshot() { + CacheStats s = caches.stats(); + return new Snapshot(caches.estimatedSize(), s.hitCount(), s.missCount(), + s.evictionCount(), s.loadFailureCount(), s.hitRate()); Review Comment: [P2] Track failures from the code that actually builds MTMV plans. This is a manual Caffeine cache: every access is `getIfPresent()` or `put()`, while lazy and refresh `createRewriteCache()` calls run outside the cache. A lazy exception propagates before any manager call, and refresh catches it and invalidates, so Caffeine never executes a load and `loadFailureCount()` remains zero even if every plan build fails. The new proc row is therefore misleading. Either maintain an explicit build-failure counter incremented on both construction paths, or remove/rename this statistic, and add a failure-path proc test. ########## fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVCacheManager.java: ########## @@ -0,0 +1,225 @@ +// 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 org.apache.doris.common.DdlException; + +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) { + if (cache == null) { + return; + } + 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(); + } + + public Snapshot snapshot() { + CacheStats s = caches.stats(); + return new Snapshot(caches.estimatedSize(), s.hitCount(), s.missCount(), + s.evictionCount(), s.loadFailureCount(), 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(); + } + return caches.policy().expireAfterAccess() + .map(exp -> exp.youngest(limit).keySet().stream() + .map(k -> new HotEntry(k.mtmvId, k.guarded, + exp.ageOf(k, TimeUnit.MILLISECONDS).orElse(-1L))) + .collect(Collectors.toList())) + .orElseGet(() -> caches.asMap().keySet().stream() + .limit(limit) + .map(k -> new HotEntry(k.mtmvId, k.guarded, -1L)) + .collect(Collectors.toList())); + } + + public void updateConfig() { + Cache<Key, MTMVCache> fresh = build(Config.mtmv_cache_manage_num, Config.expire_mtmv_cache_in_fe_second); + synchronized (swapLock) { + fresh.putAll(caches.asMap()); + fresh.cleanUp(); + caches = fresh; + } + } + + public static synchronized void reloadConfig() { + Env env = Env.getCurrentEnv(); + if (env == null) { + return; + } + MTMVCacheManager manager = env.getMtmvCacheManager(); + if (manager == null) { + return; + } + manager.updateConfig(); + } + + private static Cache<Key, MTMVCache> build(int maxSize, long expireAfterAccessSeconds) { + Caffeine<Object, Object> builder = Caffeine.newBuilder().softValues().recordStats() + .maximumSize(Math.max(maxSize, 0)); Review Comment: [P2] Skip eager refresh planning when zero disables retention. This `maximumSize(0)` policy guarantees that every put is discarded, but `MTMV.addTaskResult()` still calls `createRewriteCache()` twice on each successful live refresh; each call enters `MTMVCache.from()` with `needCost=true` and runs a full Nereids planning/costing pass before these values are thrown away. This is distinct from the existing zero-value storage thread: the bound is now correct, but the producer still does redundant work. Expose an `isEnabled()`/capacity check so refresh keeps its generation/invalidation transition without building either cache in zero mode, and cover that path with a refresh test. ########## fe/fe-core/src/main/java/org/apache/doris/catalog/MTMV.java: ########## @@ -497,51 +502,43 @@ public Set<TableNameInfo> getQueryRewriteConsistencyRelaxedTables() { */ public MTMVCache getOrGenerateCache(ConnectContext connectionContext) throws org.apache.doris.nereids.exceptions.AnalysisException { - // store two MTMVCaches: one is a cache where SessionVariables differ from those at creation time, - // and the MTMV plan includes a guardexpr; - // the other is a cache where SessionVariables are the same as at creation time, and the MTMV plan - // does not include a guardexpr; - // This way, when sessionVariables are the same, rewriting is possible; - // When sessionVariables are different, there are two cases: - // 1. If a guardexpr is present, rewriting is not possible; - // 2. If no guardexpr is present, rewriting is possible. - // Determine if current session variables match MV creation session variables Map<String, String> currentSessionVars = connectionContext.getSessionVariable().getAffectQueryResultInPlanVariables(); boolean sessionVarsMatch = SessionVarGuardRewriter.checkSessionVariablesMatch( currentSessionVars, this.sessionVariables); + boolean guarded = !sessionVarsMatch; + MTMVCacheManager manager = Env.getCurrentEnv().getMtmvCacheManager(); while (true) { long cacheGeneration; - // Select appropriate cache based on session variable match + MTMVCache cached; readMvLock(); try { - MTMVCache cache = getCache(sessionVarsMatch); - if (cache != null) { - return cache; - } + cached = manager.getIfPresent(this.id, guarded); cacheGeneration = rewriteCacheGeneration; } finally { readMvUnlock(); } - - // Generate cache if not exists - // Concurrent situations may result in duplicate cache generation, - // but we tolerate this in order to prevent nested use of readLock and write MvLock for the table - MTMVCache mtmvCache = createRewriteCache(connectionContext, false, !sessionVarsMatch); - writeMvLock(); + if (cached != null) { + return cached; + } + MTMVCache generated = createRewriteCache(connectionContext, false, guarded); Review Comment: [P2] Reuse one generated plan throughout each zero-retention planning operation. With this PR's supported `maximumSize(0)`, every put is discarded, but current consumers immediately ask for the same plan again: `LogicalOlapScan.computeDataTrait()` invokes four trait methods and each calls `getOrGenerateCache()` both directly and through `constructReplaceMap()`, producing up to eight equivalent builds for one scan. Separately, async-MV initialization already obtains a costed cache, but `AsyncMaterializationContext` drops it and `getPlanStatistics()` performs another full costed build after every successful rewrite. This is distinct from the eager refresh waste and the zero-storage-policy thread. Pin the cache for scan trait derivation and retain its statistics in the async context; add zero-mode build-count tests for both paths. ########## fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVCacheManager.java: ########## @@ -0,0 +1,225 @@ +// 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 org.apache.doris.common.DdlException; + +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) { + if (cache == null) { + return; + } + 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(); + } + + public Snapshot snapshot() { + CacheStats s = caches.stats(); + return new Snapshot(caches.estimatedSize(), s.hitCount(), s.missCount(), + s.evictionCount(), s.loadFailureCount(), 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(); + } + return caches.policy().expireAfterAccess() + .map(exp -> exp.youngest(limit).keySet().stream() + .map(k -> new HotEntry(k.mtmvId, k.guarded, + exp.ageOf(k, TimeUnit.MILLISECONDS).orElse(-1L))) + .collect(Collectors.toList())) + .orElseGet(() -> caches.asMap().keySet().stream() + .limit(limit) + .map(k -> new HotEntry(k.mtmvId, k.guarded, -1L)) + .collect(Collectors.toList())); + } + + public void updateConfig() { + Cache<Key, MTMVCache> fresh = build(Config.mtmv_cache_manage_num, Config.expire_mtmv_cache_in_fe_second); + synchronized (swapLock) { + fresh.putAll(caches.asMap()); + fresh.cleanUp(); + caches = fresh; + } + } + + public static synchronized void reloadConfig() { + Env env = Env.getCurrentEnv(); + if (env == null) { + return; + } + MTMVCacheManager manager = env.getMtmvCacheManager(); + if (manager == null) { + return; + } + manager.updateConfig(); + } + + private static Cache<Key, MTMVCache> build(int maxSize, long expireAfterAccessSeconds) { + Caffeine<Object, Object> builder = Caffeine.newBuilder().softValues().recordStats() + .maximumSize(Math.max(maxSize, 0)); + if (expireAfterAccessSeconds > 0) { + builder.expireAfterAccess(Duration.ofSeconds(expireAfterAccessSeconds)); + } + return builder.build(); + } + + /** Reject negative maximums so the cache can never lose its entry bound. */ + @VisibleForTesting + static void checkMaxSize(String confVal) throws DdlException { + int value; + try { + value = Integer.parseInt(confVal.trim()); + } catch (NumberFormatException e) { + throw new DdlException("mtmv_cache_manage_num requires an integer, but got: " + confVal); + } + if (value < 0) { + throw new DdlException("mtmv_cache_manage_num must not be negative, 0 disables the cache"); + } + } + + // NOTE: referenced by Config.mtmv_cache_manage_num.callbackClassString and + // Config.expire_mtmv_cache_in_fe_second.callbackClassString. + public static class UpdateConfig extends DefaultConfHandler { + @Override + public void handle(Field field, String confVal) throws Exception { + if ("mtmv_cache_manage_num".equals(field.getName())) { Review Comment: [P2] Validate the maximum before the default callback publishes it. `ConfigBase.setMutableConfig()` invokes the annotation's default handler first, so by the time this string callback calls `checkMaxSize()`, `Config.mtmv_cache_manage_num` is already negative. Throwing leaves a failed `ADMIN SET` visible; a later valid expiry update reloads with that poisoned value and `build()` clamps it to zero, silently disabling the cache. Startup `setFields()` also bypasses string callbacks, so a negative `fe.conf` value is never rejected. This is distinct from the existing value-policy thread: the attempted rejection itself is not atomic or applied at startup. Please validate in the Config-side setter/startup path and test `ConfigBase.setMutableConfig()` rather than only `checkMaxSize()` directly. ########## regression-test/suites/mtmv_p0/test_mtmv_cache_proc.groovy: ########## @@ -0,0 +1,90 @@ +// 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. + +suite("test_mtmv_cache_proc", "mtmv") { + def dbName = "regression_test_mtmv_p0" + def tableName = "t_test_mtmv_cache_proc_user" Review Comment: [P2] Bring this suite into the mandatory regression-test conventions. It creates one ordinary table, so Testing Standard 4 requires hardcoding `t_test_mtmv_cache_proc_user` in the SQL instead of using `def tableName`. The `/mtmv_cache` directory has a fixed ordered `stat`/`hot` result, so Standard 6 requires an ordered `qt` case with auto-generated output rather than assertion-only validation. Finally, the EOF drops violate Standard 3 because the suite already cleans before creation and must preserve final state for debugging. These are separate from the existing vacuous-hot-assertion thread. Please hardcode the table name, add the deterministic qt/output, and remove the trailing drops. -- 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]
