github-actions[bot] commented on code in PR #68141: URL: https://github.com/apache/doris/pull/68141#discussion_r4038418577
########## 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" + def mvName = "mtmv_cache_proc_mv" + + sql """drop materialized view if exists ${mvName}""" + sql """drop table if exists ${tableName}""" + + sql """ + CREATE TABLE IF NOT EXISTS ${tableName} ( + event_day DATE, + id BIGINT, + username VARCHAR(20) + ) + DISTRIBUTED BY HASH(id) BUCKETS 2 + PROPERTIES ('replication_num' = '1'); + """ + + // SHOW PROC '/mtmv_cache' should list two children: stat and hot. + def dirRows = sql """SHOW PROC '/mtmv_cache'""" + assertEquals(2, dirRows.size()) + def dirNames = dirRows.collect { it[0] } + assertTrue(dirNames.contains("stat")) + assertTrue(dirNames.contains("hot")) + + // SHOW PROC '/mtmv_cache/stat' returns 6 KV rows. + def statRows = sql """SHOW PROC '/mtmv_cache/stat'""" + def statKeys = statRows.collect { it[0] } + ["size", "hitCount", "missCount", "evictionCount", "loadFailureCount", "hitRate"].each { + assertTrue(statKeys.contains(it), "stat missing key: ${it}") + } + + // Create an MV and trigger cache fill via a rewrite-eligible query. + sql """ + CREATE MATERIALIZED VIEW ${mvName} + BUILD DEFERRED REFRESH COMPLETE ON MANUAL + DISTRIBUTED BY RANDOM BUCKETS 2 + PROPERTIES ('replication_num' = '1') + AS + SELECT event_day, id, username FROM ${tableName}; + """ + def jobName = getJobName(dbName, mvName) + sql """REFRESH MATERIALIZED VIEW ${mvName} AUTO""" + waitingMTMVTaskFinished(jobName) + // Query the base table so nereids checks the MV — fills the cache. + sql """SELECT event_day, id, username FROM ${tableName}""" + + // hot proc: 5 columns, our MV should appear with its real DbName/MvName. + def hotRows = sql """SHOW PROC '/mtmv_cache/hot'""" + if (!hotRows.isEmpty()) { Review Comment: [P2] Make the hot-cache assertion non-vacuous. If cache publication is broken and this proc always returns zero rows, this entire block is skipped; if name resolution drops only this MV, the inner block is skipped; and the later `size() <= 1` check still accepts zero. The regression therefore passes without exercising the feature it claims to cover. Please assert that `hotRows` is non-empty and that `mvRow` is non-null before checking its fields, and assert the expected capped row rather than only an upper bound. ########## fe/fe-core/src/main/java/org/apache/doris/catalog/MTMV.java: ########## @@ -497,51 +498,40 @@ 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) { + MTMVCache cached = manager.getIfPresent(this.id, guarded); Review Comment: [P1] Order cache hits with the generation invalidation. This lookup no longer takes `readMvLock()`, so after `invalidateRewriteCache()` has acquired the write lock and incremented `rewriteCacheGeneration`, a planner can still read and return the old Caffeine entry before the writer reaches `manager.invalidate()`. The removed implementation checked the per-MTMV cache under the read lock, so it could not cross this writer critical section. For ADD/DROP CONSTRAINT invalidations that can expose a plan carrying obsolete uniqueness/FK traits. Please perform the hit lookup under the MV read lock (or atomically validate the generation with it) and add a paused-invalidation concurrency test. ########## fe/fe-core/src/main/java/org/apache/doris/common/proc/MTMVCacheHotProcNode.java: ########## @@ -0,0 +1,74 @@ +// 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.common.proc; + +import org.apache.doris.catalog.Env; +import org.apache.doris.catalog.Table; +import org.apache.doris.common.AnalysisException; +import org.apache.doris.common.Config; +import org.apache.doris.datasource.InternalCatalog; +import org.apache.doris.mtmv.MTMVCacheManager; +import org.apache.doris.mtmv.MTMVCacheManager.HotEntry; + +import com.google.common.collect.ImmutableList; +import com.google.common.collect.Lists; + +public class MTMVCacheHotProcNode implements ProcNodeInterface { + public static final ImmutableList<String> TITLE_NAMES = new ImmutableList.Builder<String>() + .add("MtmvId").add("DbName").add("MvName").add("Guarded").add("IdleMs").build(); + + private static final String UNKNOWN_DB = "<unknown>"; + private static final String DROPPED_MV = "<dropped>"; + + private final MTMVCacheManager manager; + + public MTMVCacheHotProcNode(MTMVCacheManager manager) { + this.manager = manager; + } + + @Override + public ProcResult fetchResult() throws AnalysisException { + BaseProcResult result = new BaseProcResult(); + result.setNames(TITLE_NAMES); + if (manager == null) { + return result; + } + InternalCatalog catalog = Env.getCurrentEnv() == null ? null : Env.getCurrentInternalCatalog(); + for (HotEntry entry : manager.hotEntries(Config.mtmv_cache_hot_show_num)) { + String dbName = UNKNOWN_DB; + String mvName = DROPPED_MV; + if (catalog != null) { + Table table = catalog.getTableByTableId(entry.mtmvId); Review Comment: [P2] Avoid scanning every database for every hot entry. `InternalCatalog.getTableByTableId()` linearly walks all databases, this call is inside the up-to-500-row loop, and `ShowProcCommand` fetches the node once for rows and again for metadata. On a many-database FE one diagnostic request therefore performs `2 * hotRows * databaseCount` synchronous probes, and the mutable cap can be raised further. Please retain enough catalog/database identity for direct lookup, or build one table-ID/name snapshot per fetch instead of restarting the catalog scan for every row. ########## fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVCacheManager.java: ########## @@ -0,0 +1,193 @@ +// 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 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; + } + caches.put(new Key(mtmvId, guarded), cache); + } + + public void invalidate(long mtmvId) { + caches.invalidate(new Key(mtmvId, true)); + caches.invalidate(new Key(mtmvId, false)); + } + + public void invalidateAll() { + 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. */ + 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 static synchronized void updateConfig() { + Env env = Env.getCurrentEnv(); + if (env == null) { + return; + } + MTMVCacheManager manager = env.getMtmvCacheManager(); + if (manager == null) { + return; + } + Cache<Key, MTMVCache> fresh = build(Config.mtmv_cache_manage_num, Config.expire_mtmv_cache_in_fe_second); + fresh.putAll(manager.caches.asMap()); Review Comment: [P1] Preserve mutations and policy state across dynamic updates. `updateConfig()` copies the old cache and only later publishes `fresh`, while `put()`/`invalidate()` dereference `caches` without the same lock. For example, the copy can capture an MTMV plan, ADD/DROP CONSTRAINT can advance `rewriteCacheGeneration` and invalidate only the old instance, and line 106 can then republish the stale plan; a concurrent refresh/lazy `put` can likewise be lost in the retired instance. Independently, `putAll` records every retained entry as a fresh write, so even a size-only update resets expiry age/recency/statistics and rejuvenates nearly expired entries. Please mutate the live Caffeine policies in place where supported; serialize any unavoidable rebuild with every mutation and preserve its policy metadata. Add latch and fake-ticker tests for both orderings. ########## fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVCacheManager.java: ########## @@ -0,0 +1,193 @@ +// 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 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; + } + caches.put(new Key(mtmvId, guarded), cache); + } + + public void invalidate(long mtmvId) { + caches.invalidate(new Key(mtmvId, true)); + caches.invalidate(new Key(mtmvId, false)); + } + + public void invalidateAll() { + 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. */ + 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() Review Comment: [P2] Preserve the hot-view contract when expiration is disabled. This fallback is reachable because the mutable expiration config accepts `0`/negative values and `build()` then omits `expireAfterAccess`. In that state `asMap().keySet()` has no access-recency order and every row gets `IdleMs=-1`, even though this method and `/mtmv_cache/hot` unconditionally advertise most-recently-accessed ordering and idle age. Please either reject non-positive expiration values or retain independent access-time metadata, and cover the no-expiry configuration in a test. -- 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]
