This is an automated email from the ASF dual-hosted git repository.
morrySnow pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 1c28250c836 [opt](mtmv) manage MTMVCache with a global LRU manager
(#68141)
1c28250c836 is described below
commit 1c28250c8365169dd8cf088c60d1842472adef50
Author: xy720 <[email protected]>
AuthorDate: Wed Sep 23 11:38:19 2026 +0800
[opt](mtmv) manage MTMVCache with a global LRU manager (#68141)
### What problem does this PR solve?
Problem Summary:
Before this patch each MTMV kept two MTMVCache (cacheWithGuard /
cacheWithoutGuard) that held Nereids plan trees.
The lifetime of MTMVCache is same as the MTMV object, when these
MTMVCaches are at large scale (e.g. thousands of mtmv were created and
refreshed, but they were not dropped afterwards) may ocupy lots of FE
heap space.
A heap-histogram check showing 72540 MTMVCache instances held ~30GB in
FE heap on the affected cluster.
This pr replace the origin MTMVCaches with Caffeine cache keyed by
(mtmvId, guarded), using maximumSize + expireAfterAccess so cold entries
are evicted under memory pressure instead of accumulating forever.
Initial config, all mutable via ADMIN SET FRONTEND CONFIG:
```
mtmv_cache_manage_num=3000,
expire_mtmv_cache_in_fe_second=86400,
mtmv_cache_hot_show_num=500.
```
Also support observability:
```
SHOW PROC '/mtmv_cache/stat' returns:
size / hitCount / missCount / evictionCount / loadFailureCount / hitRate
SHOW PROC '/mtmv_cache/hot' returns MtmvId / DbName / MvName / Guarded /
IdleMs
ordered by most-recently-accessed for MVs
```
This pr refers to the NereidsSqlCacheManager introduced by #33262
---
.../main/java/org/apache/doris/common/Config.java | 36 +++
.../java/org/apache/doris/common/ConfigTest.java | 26 ++
.../src/main/java/org/apache/doris/DorisFE.java | 1 +
.../main/java/org/apache/doris/catalog/Env.java | 8 +
.../main/java/org/apache/doris/catalog/MTMV.java | 114 +++++----
.../doris/common/proc/MTMVCacheHotProcNode.java | 110 +++++++++
.../apache/doris/common/proc/MTMVCacheProcDir.java | 61 +++++
.../doris/common/proc/MTMVCacheStatProcNode.java | 49 ++++
.../org/apache/doris/common/proc/ProcService.java | 1 +
.../org/apache/doris/mtmv/MTMVCacheManager.java | 210 ++++++++++++++++
.../org/apache/doris/nereids/StatementContext.java | 13 +
.../apache/doris/mtmv/MTMVCacheManagerTest.java | 268 +++++++++++++++++++++
.../test/java/org/apache/doris/mtmv/MTMVTest.java | 233 +++++++++++++++++-
.../data/mtmv_p0/test_mtmv_cache_proc.out | 5 +
.../suites/mtmv_p0/test_mtmv_cache_proc.groovy | 83 +++++++
15 files changed, 1165 insertions(+), 53 deletions(-)
diff --git a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
index 40a1b8dfcfb..1704a3fa252 100644
--- a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
+++ b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
@@ -2305,6 +2305,42 @@ public class Config extends ConfigBase {
+ "pruning.")
public static int cache_partition_meta_table_manage_num = 100;
+ @ConfField(
+ mutable = true,
+ callback = NonNegativeMtmvCacheNumConfHandler.class,
+ callbackClassString =
"org.apache.doris.mtmv.MTMVCacheManager$UpdateConfig",
+ description = "Max mtmv plan cache entries kept by
MTMVCacheManager. 0 disables the cache, "
+ + "negative values are rejected. Default 3000.")
+ public static int mtmv_cache_manage_num = 3000;
+
+ public static class NonNegativeMtmvCacheNumConfHandler implements
ConfHandler {
+ @Override
+ public void handle(Field field, String value) throws Exception {
+ int parsed = Integer.parseInt(value.trim());
+ if (parsed < 0) {
+ throw new ConfigException(field.getName() + " must not be
negative, 0 disables the cache");
+ }
+ field.setInt(null, parsed);
+ }
+ }
+
+ public static void validateMtmvCacheConfig() throws ConfigException {
+ if (mtmv_cache_manage_num < 0) {
+ throw new ConfigException("mtmv_cache_manage_num must not be
negative, 0 disables the cache");
+ }
+ }
+
+ @ConfField(
+ mutable = true,
+ callbackClassString =
"org.apache.doris.mtmv.MTMVCacheManager$UpdateConfig",
+ description = "Idle expiration in seconds for entries in
MTMVCacheManager. Default 86400.")
+ public static long expire_mtmv_cache_in_fe_second = 86400;
+
+ @ConfField(
+ mutable = true,
+ description = "Row cap for SHOW PROC '/mtmv_cache/hot'. Default
500.")
+ public static int mtmv_cache_hot_show_num = 500;
+
/**
* HBO plan stats. cache number which can be reused for the next query.
*/
diff --git a/fe/fe-common/src/test/java/org/apache/doris/common/ConfigTest.java
b/fe/fe-common/src/test/java/org/apache/doris/common/ConfigTest.java
index e18b3bd4355..c6eee5bef5d 100644
--- a/fe/fe-common/src/test/java/org/apache/doris/common/ConfigTest.java
+++ b/fe/fe-common/src/test/java/org/apache/doris/common/ConfigTest.java
@@ -186,6 +186,32 @@ public class ConfigTest {
}
}
+ @Test
+ public void testMtmvCacheManageNumRejectsNegative() throws Exception {
+ int original = Config.mtmv_cache_manage_num;
+ try {
+ Config.mtmv_cache_manage_num = 100;
+ // ADMIN SET FRONTEND CONFIG runs the annotation callback before
the cache-reload handler,
+ // so a negative maximum must be refused there and leave the field
untouched.
+ ConfigException negative =
Assertions.assertThrows(ConfigException.class,
+ () -> ConfigBase.setMutableConfig("mtmv_cache_manage_num",
"-1"));
+ Assertions.assertTrue(negative.getMessage().contains("must not be
negative"));
+ Assertions.assertEquals(100, Config.mtmv_cache_manage_num);
+
+ // 0 is the documented way to disable the cache.
+ new Config.NonNegativeMtmvCacheNumConfHandler()
+ .handle(ConfigBase.getField("mtmv_cache_manage_num"), " 0
");
+ Assertions.assertEquals(0, Config.mtmv_cache_manage_num);
+ Assertions.assertDoesNotThrow(Config::validateMtmvCacheConfig);
+
+ // fe.conf assigns the field without running any callback, so
startup validates it too.
+ Config.mtmv_cache_manage_num = -1;
+ Assertions.assertThrows(ConfigException.class,
Config::validateMtmvCacheConfig);
+ } finally {
+ Config.mtmv_cache_manage_num = original;
+ }
+ }
+
@Test
public void testValidateWebSqlStartupConfig() throws ConfigException {
int originalIdleTimeout = Config.web_sql_session_idle_timeout_seconds;
diff --git a/fe/fe-core/src/main/java/org/apache/doris/DorisFE.java
b/fe/fe-core/src/main/java/org/apache/doris/DorisFE.java
index 56d90909b13..f341a486ff9 100755
--- a/fe/fe-core/src/main/java/org/apache/doris/DorisFE.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/DorisFE.java
@@ -151,6 +151,7 @@ public class DorisFE {
// Because the path of custom config file is defined in fe.conf
config.initCustom(Config.custom_config_dir + "/fe_custom.conf");
Config.validateWebSqlConfig();
+ Config.validateMtmvCacheConfig();
// inverted_index_storage_format's runtime callback is not invoked
while parsing
// fe.conf/fe_custom.conf, so validate the loaded value here after
both files are loaded
// and merged, to reject a "V1" left over in the config files at
startup.
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java
b/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java
index 5f20e4feb39..ab594e7487d 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java
@@ -155,6 +155,7 @@ import org.apache.doris.meta.MetaContext;
import org.apache.doris.metric.MetricRepo;
import org.apache.doris.mtmv.BaseTableInfo;
import org.apache.doris.mtmv.MTMVAlterOpType;
+import org.apache.doris.mtmv.MTMVCacheManager;
import org.apache.doris.mtmv.MTMVPartitionExprFactory;
import org.apache.doris.mtmv.MTMVPartitionInfo;
import org.apache.doris.mtmv.MTMVPartitionInfo.MTMVPartitionType;
@@ -589,6 +590,8 @@ public class Env {
private final NereidsSortedPartitionsCacheManager
sortedPartitionsCacheManager;
+ private final MTMVCacheManager mtmvCacheManager;
+
private final SplitSourceManager splitSourceManager;
private final GlobalExternalTransactionInfoMgr
globalExternalTransactionInfoMgr;
@@ -887,6 +890,7 @@ public class Env {
this.dnsCache = new DNSCache();
this.sqlCacheManager = new NereidsSqlCacheManager();
this.sortedPartitionsCacheManager = new
NereidsSortedPartitionsCacheManager();
+ this.mtmvCacheManager = new MTMVCacheManager();
this.splitSourceManager = new SplitSourceManager();
this.globalExternalTransactionInfoMgr = new
GlobalExternalTransactionInfoMgr();
this.tokenManager = new TokenManager();
@@ -7667,6 +7671,10 @@ public class Env {
return sqlCacheManager;
}
+ public MTMVCacheManager getMtmvCacheManager() {
+ return mtmvCacheManager;
+ }
+
public NereidsSortedPartitionsCacheManager
getSortedPartitionsCacheManager() {
return sortedPartitionsCacheManager;
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/MTMV.java
b/fe/fe-core/src/main/java/org/apache/doris/catalog/MTMV.java
index ec97efb324f..586d9341aa6 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/MTMV.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/MTMV.java
@@ -35,6 +35,7 @@ import org.apache.doris.mtmv.BaseTableInfo;
import org.apache.doris.mtmv.EnvInfo;
import org.apache.doris.mtmv.MTMVAlterOpType;
import org.apache.doris.mtmv.MTMVCache;
+import org.apache.doris.mtmv.MTMVCacheManager;
import org.apache.doris.mtmv.MTMVJobInfo;
import org.apache.doris.mtmv.MTMVJobManager;
import org.apache.doris.mtmv.MTMVPartitionExpander;
@@ -56,6 +57,7 @@ import org.apache.doris.mtmv.MTMVStatus;
import org.apache.doris.mtmv.MTMVUtil;
import org.apache.doris.mtmv.ivm.IvmInfo;
import org.apache.doris.mtmv.ivm.IvmUtil;
+import org.apache.doris.nereids.StatementContext;
import org.apache.doris.nereids.rules.analysis.SessionVarGuardRewriter;
import
org.apache.doris.nereids.trees.plans.commands.info.RefreshMTMVInfo.RefreshMode;
import org.apache.doris.persist.AlterMTMV;
@@ -125,11 +127,6 @@ public class MTMV extends OlapTable {
*/
@SerializedName("pst")
private Map<String, MTMVPartitionState> partitionStates;
- // Should update after every fresh, not persist
- // Cache with SessionVarGuardExpr: used when query session variables
differ from MV creation variables
- private MTMVCache cacheWithGuard;
- // Cache without SessionVarGuardExpr: used when query session variables
match MV creation variables
- private MTMVCache cacheWithoutGuard;
// Increased every time rewrite cache is invalidated to prevent publishing
stale in-flight cache builds.
private transient long rewriteCacheGeneration;
private long schemaChangeVersion;
@@ -283,8 +280,8 @@ public class MTMV extends OlapTable {
}
try {
// The replay thread may not have initialized the catalog yet
to avoid getting stuck due
- // to connection issues such as S3, so it is directly set to
null
- if (!isReplay) {
+ // to connection issues such as S3, so it is directly set to
null.
+ if (!isReplay &&
Env.getCurrentEnv().getMtmvCacheManager().isEnabled()) {
ConnectContext currentContext = ConnectContext.get();
// shouldn't do this while holding mvWriteLock
// TODO: these two cache compute share something same, can
be simplified in future
@@ -329,12 +326,19 @@ public class MTMV extends OlapTable {
}
ivmInfo.clearBaselineRebuild();
}
+ // The refresh publishes a new plan, so every cache built
before this commit is stale.
+ // Bump before publishing so an in-flight build cannot pass
its generation check later.
+ boolean publishCache = needUpdateCache && cacheGeneration ==
rewriteCacheGeneration && !isDropped;
+ rewriteCacheGeneration++;
if (needUpdateCache) {
- if (cacheGeneration == rewriteCacheGeneration) {
- // Initialize cacheWithGuard, cacheWithoutGuard will
be lazily generated when needed
- this.cacheWithGuard = mtmvCacheWithGuard;
- // Clear the other cache to ensure consistency
- this.cacheWithoutGuard = mtmvCacheWithoutGuard;
+ MTMVCacheManager manager =
Env.getCurrentEnv().getMtmvCacheManager();
+ if (publishCache && mtmvCacheWithGuard != null) {
+ manager.put(this.id, true, mtmvCacheWithGuard);
+ } else {
+ manager.invalidate(this.id);
+ }
+ if (publishCache && mtmvCacheWithoutGuard != null) {
+ manager.put(this.id, false, mtmvCacheWithoutGuard);
}
}
} else {
@@ -545,51 +549,56 @@ public class MTMV extends OlapTable {
*/
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();
+ StatementContext statementContext =
connectionContext.getStatementContext();
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.isEnabled() ? manager.getIfPresent(this.id,
guarded) : null;
+ if (cached == null && statementContext != null) {
+ cached = statementContext.getQueryLocalMtmvCache(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);
+ readMvLock();
try {
- MTMVCache cache = getCache(sessionVarsMatch);
- if (cache != null) {
- return cache;
- }
if (cacheGeneration != rewriteCacheGeneration) {
+ // Someone invalidated between our snapshot and now; drop
the stale build and retry.
continue;
}
- setCache(sessionVarsMatch, mtmvCache);
- return mtmvCache;
+ if (manager.isEnabled()) {
+ MTMVCache existing = manager.getIfPresent(this.id,
guarded);
+ if (existing != null) {
+ return existing;
+ }
+ if (!isDropped) {
+ manager.put(this.id, guarded, generated);
+ }
+ } else if (statementContext != null && !isDropped) {
+ // Global cache is disabled (maximumSize=0); keep one copy
for this statement only.
+ MTMVCache existing =
statementContext.getQueryLocalMtmvCache(this.id, guarded);
+ if (existing != null) {
+ return existing;
+ }
+ statementContext.putQueryLocalMtmvCache(this.id, guarded,
generated);
+ }
+ return generated;
} finally {
- writeMvUnlock();
+ readMvUnlock();
}
}
}
@@ -1049,8 +1058,7 @@ public class MTMV extends OlapTable {
writeMvLock();
try {
rewriteCacheGeneration++;
- cacheWithGuard = null;
- cacheWithoutGuard = null;
+ Env.getCurrentEnv().getMtmvCacheManager().invalidate(this.id);
} finally {
writeMvUnlock();
}
@@ -1215,18 +1223,6 @@ public class MTMV extends OlapTable {
this.mvRwLock.writeLock().unlock();
}
- private MTMVCache getCache(boolean sessionVarsMatch) {
- return sessionVarsMatch ? cacheWithoutGuard : cacheWithGuard;
- }
-
- private void setCache(boolean sessionVarsMatch, MTMVCache cache) {
- if (sessionVarsMatch) {
- this.cacheWithoutGuard = cache;
- } else {
- this.cacheWithGuard = cache;
- }
- }
-
// toString() is not easy to find where to call the method
public String toInfoString() {
final StringBuilder sb = new StringBuilder("MTMV{");
@@ -1304,6 +1300,20 @@ public class MTMV extends OlapTable {
compatiblePctSnapshot(partitionSnapshots);
}
+ @Override
+ public void markDropped() {
+ super.markDropped();
+ // A refresh or query building a cache outside the MV lock must not
+ // be able to republish it after the drop.
+ writeMvLock();
+ try {
+ rewriteCacheGeneration++;
+ Env.getCurrentEnv().getMtmvCacheManager().invalidate(this.id);
+ } finally {
+ writeMvUnlock();
+ }
+ }
+
private void compatiblePctSnapshot(Map<String,
MTMVRefreshPartitionSnapshot> partitionSnapshots) {
BaseTableInfo relatedTableInfo = mvPartitionInfo.getRelatedTableInfo();
if (relatedTableInfo == null) {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/common/proc/MTMVCacheHotProcNode.java
b/fe/fe-core/src/main/java/org/apache/doris/common/proc/MTMVCacheHotProcNode.java
new file mode 100644
index 00000000000..45f0e32f7f5
--- /dev/null
+++
b/fe/fe-core/src/main/java/org/apache/doris/common/proc/MTMVCacheHotProcNode.java
@@ -0,0 +1,110 @@
+// 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.Database;
+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;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+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);
+ List<HotEntry> entries =
manager.hotEntries(Config.mtmv_cache_hot_show_num);
+ if (entries.isEmpty()) {
+ return result;
+ }
+ Set<Long> wanted = new HashSet<>();
+ for (HotEntry e : entries) {
+ wanted.add(e.mtmvId);
+ }
+ Map<Long, Table> idToTable = resolveTables(wanted);
+ for (HotEntry entry : entries) {
+ String dbName = UNKNOWN_DB;
+ String mvName = DROPPED_MV;
+ Table table = idToTable.get(entry.mtmvId);
+ if (table != null) {
+ mvName = table.getName();
+ String qualified = table.getQualifiedDbName();
+ if (qualified != null && !qualified.isEmpty()) {
+ dbName = qualified;
+ }
+ }
+ result.addRow(Lists.newArrayList(
+ String.valueOf(entry.mtmvId),
+ dbName,
+ mvName,
+ entry.guarded ? "Yes" : "No",
+ String.valueOf(entry.idleMs)));
+ }
+ return result;
+ }
+
+ private static Map<Long, Table> resolveTables(Set<Long> ids) {
+ Map<Long, Table> out = new HashMap<>();
+ if (Env.getCurrentEnv() == null) {
+ return out;
+ }
+ InternalCatalog catalog = Env.getCurrentInternalCatalog();
+ if (catalog == null) {
+ return out;
+ }
+ for (Database db : catalog.getDbs()) {
+ for (Long id : ids) {
+ if (out.containsKey(id)) {
+ continue;
+ }
+ Table t = db.getTableNullable(id);
+ if (t != null) {
+ out.put(id, t);
+ }
+ }
+ if (out.size() == ids.size()) {
+ break;
+ }
+ }
+ return out;
+ }
+}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/common/proc/MTMVCacheProcDir.java
b/fe/fe-core/src/main/java/org/apache/doris/common/proc/MTMVCacheProcDir.java
new file mode 100644
index 00000000000..c1cc208e88f
--- /dev/null
+++
b/fe/fe-core/src/main/java/org/apache/doris/common/proc/MTMVCacheProcDir.java
@@ -0,0 +1,61 @@
+// 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.common.AnalysisException;
+import org.apache.doris.mtmv.MTMVCacheManager;
+
+import com.google.common.base.Strings;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Lists;
+
+/** Two-level proc dir for '/mtmv_cache': "stat" and "hot" child nodes. */
+public class MTMVCacheProcDir implements ProcDirInterface {
+ public static final ImmutableList<String> TITLE_NAMES = new
ImmutableList.Builder<String>()
+ .add("Name").add("Info").build();
+
+ @Override
+ public ProcResult fetchResult() throws AnalysisException {
+ BaseProcResult result = new BaseProcResult();
+ result.setNames(TITLE_NAMES);
+ result.addRow(Lists.newArrayList("stat", "Global cache stats"));
+ result.addRow(Lists.newArrayList("hot", "Top hot mtmv cache entries"));
+ return result;
+ }
+
+ @Override
+ public boolean register(String name, ProcNodeInterface node) {
+ return false;
+ }
+
+ @Override
+ public ProcNodeInterface lookup(String name) throws AnalysisException {
+ if (Strings.isNullOrEmpty(name)) {
+ throw new AnalysisException("mtmv_cache child name is empty");
+ }
+ MTMVCacheManager manager = Env.getCurrentEnv().getMtmvCacheManager();
+ if (name.equalsIgnoreCase("stat")) {
+ return new MTMVCacheStatProcNode(manager);
+ }
+ if (name.equalsIgnoreCase("hot")) {
+ return new MTMVCacheHotProcNode(manager);
+ }
+ throw new AnalysisException("unknown mtmv_cache child: " + name);
+ }
+}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/common/proc/MTMVCacheStatProcNode.java
b/fe/fe-core/src/main/java/org/apache/doris/common/proc/MTMVCacheStatProcNode.java
new file mode 100644
index 00000000000..ed735ab71e8
--- /dev/null
+++
b/fe/fe-core/src/main/java/org/apache/doris/common/proc/MTMVCacheStatProcNode.java
@@ -0,0 +1,49 @@
+// 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.common.AnalysisException;
+import org.apache.doris.mtmv.MTMVCacheManager;
+import org.apache.doris.mtmv.MTMVCacheManager.Snapshot;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Lists;
+
+public class MTMVCacheStatProcNode implements ProcNodeInterface {
+ public static final ImmutableList<String> TITLE_NAMES = new
ImmutableList.Builder<String>()
+ .add("Name").add("Value").build();
+
+ private final MTMVCacheManager manager;
+
+ public MTMVCacheStatProcNode(MTMVCacheManager manager) {
+ this.manager = manager;
+ }
+
+ @Override
+ public ProcResult fetchResult() throws AnalysisException {
+ BaseProcResult result = new BaseProcResult();
+ result.setNames(TITLE_NAMES);
+ Snapshot s = manager.snapshot();
+ result.addRow(Lists.newArrayList("size", String.valueOf(s.size)));
+ result.addRow(Lists.newArrayList("hitCount",
String.valueOf(s.hitCount)));
+ result.addRow(Lists.newArrayList("missCount",
String.valueOf(s.missCount)));
+ result.addRow(Lists.newArrayList("evictionCount",
String.valueOf(s.evictionCount)));
+ result.addRow(Lists.newArrayList("hitRate", String.format("%.4f",
s.hitRate)));
+ return result;
+ }
+}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/common/proc/ProcService.java
b/fe/fe-core/src/main/java/org/apache/doris/common/proc/ProcService.java
index a1f54901bde..63b7c0d96e7 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/common/proc/ProcService.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/common/proc/ProcService.java
@@ -59,6 +59,7 @@ public final class ProcService {
root.register("bdbje", new BDBJEProcDir());
root.register("diagnose", new DiagnoseProcDir());
root.register("binlog", new BinlogProcDir());
+ root.register("mtmv_cache", new MTMVCacheProcDir());
}
// 通过指定的路径获得对应的PROC Node
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVCacheManager.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVCacheManager.java
new file mode 100644
index 00000000000..a6e35938318
--- /dev/null
+++ b/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVCacheManager.java
@@ -0,0 +1,210 @@
+// 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(stream -> stream
+ .limit(limit)
+ .map(entry -> {
+ Key k = entry.getKey();
+ long expireMs =
exp.getExpiresAfter(TimeUnit.MILLISECONDS);
+ long idleMs = Math.max(expireMs -
entry.expiresAfter().toMillis(), 0L);
+ return new HotEntry(k.mtmvId, k.guarded, idleMs);
+ })
+ .collect(Collectors.toList())))
+ .orElseGet(() -> current.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;
+ }
+ env.getMtmvCacheManager().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();
+ }
+
+ // 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 {
+ super.handle(field, confVal);
+ MTMVCacheManager.reloadConfig();
+ }
+ }
+
+ /** Stable composite key so it is immune to BaseTableInfo hashCode drift.
*/
+ public static final class Key {
+ public final long mtmvId;
+ public final boolean guarded;
+
+ public Key(long mtmvId, boolean guarded) {
+ this.mtmvId = mtmvId;
+ this.guarded = guarded;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof Key)) {
+ return false;
+ }
+ Key that = (Key) o;
+ return mtmvId == that.mtmvId && guarded == that.guarded;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(mtmvId, guarded);
+ }
+ }
+
+ public static final class HotEntry {
+ public final long mtmvId;
+ public final boolean guarded;
+ public final long idleMs;
+
+ public HotEntry(long mtmvId, boolean guarded, long idleMs) {
+ this.mtmvId = mtmvId;
+ this.guarded = guarded;
+ this.idleMs = idleMs;
+ }
+ }
+
+ public static final class Snapshot {
+ public final long size;
+ public final long hitCount;
+ public final long missCount;
+ public final long evictionCount;
+ public final double hitRate;
+
+ public Snapshot(long size, long hitCount, long missCount, long
evictionCount, double hitRate) {
+ this.size = size;
+ this.hitCount = hitCount;
+ this.missCount = missCount;
+ this.evictionCount = evictionCount;
+ this.hitRate = hitRate;
+ }
+ }
+
+ @VisibleForTesting
+ public Cache<Key, MTMVCache> getCachesForTest() {
+ return caches;
+ }
+}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/StatementContext.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/StatementContext.java
index 4903a00147a..64308e66baf 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/StatementContext.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/StatementContext.java
@@ -40,6 +40,7 @@ import org.apache.doris.datasource.mvcc.MvccTable;
import org.apache.doris.datasource.mvcc.MvccTableInfo;
import org.apache.doris.foundation.format.FormatOptions;
import org.apache.doris.mtmv.BaseTableInfo;
+import org.apache.doris.mtmv.MTMVCache;
import org.apache.doris.mtmv.ivm.IvmRewriteContext;
import org.apache.doris.nereids.analyzer.UnboundRelation;
import org.apache.doris.nereids.cost.CostWeight;
@@ -330,6 +331,10 @@ public class StatementContext implements Closeable {
// Record mtmv and valid partitions map because this is time-consuming
behavior
private final Map<BaseTableInfo, Collection<Partition>>
mvCanRewritePartitionsMap = new HashMap<>();
+ // When the Env-wide MTMVCacheManager is disabled
(mtmv_cache_manage_num=0), reuse rewrite plans
+ // in the same statement so multiple rewrite paths do not rebuild the same
MV plan.
+ private final Map<Pair<Long, Boolean>, MTMVCache> queryLocalMtmvCaches =
new HashMap<>();
+
/// for dictionary sink.
private List<Backend> usedBackendsDistributing; // report used backends
after done distribute planning.
private long dictionaryUsedSrcVersion; // base table data version used in
this refreshing.
@@ -1605,6 +1610,14 @@ public class StatementContext implements Closeable {
this.materializationRewrittenSuccessSet.add(materializationQualifier);
}
+ public MTMVCache getQueryLocalMtmvCache(long mtmvId, boolean guarded) {
+ return queryLocalMtmvCaches.get(Pair.of(mtmvId, guarded));
+ }
+
+ public void putQueryLocalMtmvCache(long mtmvId, boolean guarded, MTMVCache
cache) {
+ queryLocalMtmvCaches.put(Pair.of(mtmvId, guarded), cache);
+ }
+
public Multimap<List<String>, Pair<RelationId, Set<String>>>
getTableUsedPartitionNameMap() {
return tableUsedPartitionNameMap;
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVCacheManagerTest.java
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVCacheManagerTest.java
new file mode 100644
index 00000000000..6b94fa7c5da
--- /dev/null
+++ b/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVCacheManagerTest.java
@@ -0,0 +1,268 @@
+// 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.lang.ref.Reference;
+import java.util.ArrayList;
+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);
+ Reference.reachabilityFence(c1);
+ }
+
+ @Test
+ public void testHotEntriesHonorsLimit() {
+ MTMVCacheManager manager = new MTMVCacheManager();
+ List<MTMVCache> values = new ArrayList<>();
+ for (int i = 0; i < 5; i++) {
+ MTMVCache c = Mockito.mock(MTMVCache.class);
+ values.add(c);
+ 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);
+ }
+ Reference.reachabilityFence(values);
+ }
+
+ @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();
+ List<MTMVCache> values = new ArrayList<>();
+ for (int i = 0; i < 10; i++) {
+ MTMVCache c = Mockito.mock(MTMVCache.class);
+ values.add(c);
+ manager.put(i, true, c);
+ }
+ manager.getCachesForTest().cleanUp();
+ Assertions.assertEquals(10L, manager.size());
+
+ Config.mtmv_cache_manage_num = 2;
+ manager.updateConfig();
+ Assertions.assertEquals(2L, manager.size());
+
+ Config.mtmv_cache_manage_num = 0;
+ manager.updateConfig();
+ Assertions.assertEquals(0L, manager.size());
+ MTMVCache discarded = Mockito.mock(MTMVCache.class);
+ manager.put(99L, true, discarded);
+ manager.getCachesForTest().cleanUp();
+ Assertions.assertEquals(0L, manager.size());
+ Reference.reachabilityFence(values);
+ Reference.reachabilityFence(discarded);
+ } finally {
+ Config.mtmv_cache_manage_num = originalMaxSize;
+ }
+ }
+
+ @Test
+ public void testKeyEqualityAndHash() {
+ Key k1 = new Key(42L, true);
+ Key k2 = new Key(42L, true);
+ Key k3 = new Key(42L, false);
+ Assertions.assertEquals(k1, k2);
+ Assertions.assertEquals(k1.hashCode(), k2.hashCode());
+ Assertions.assertNotEquals(k1, k3);
+ }
+}
diff --git a/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVTest.java
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVTest.java
index 98aeb274eb9..5e7ea127b46 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVTest.java
@@ -33,6 +33,7 @@ import org.apache.doris.catalog.ScalarType;
import org.apache.doris.catalog.SinglePartitionInfo;
import org.apache.doris.catalog.info.TableNameInfo;
import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.Config;
import org.apache.doris.common.jmockit.Deencapsulation;
import org.apache.doris.common.util.PropertyAnalyzer;
import org.apache.doris.job.common.IntervalUnit;
@@ -43,11 +44,14 @@ import
org.apache.doris.mtmv.MTMVRefreshEnum.MTMVRefreshState;
import org.apache.doris.mtmv.MTMVRefreshEnum.MTMVState;
import org.apache.doris.mtmv.MTMVRefreshEnum.RefreshMethod;
import org.apache.doris.mtmv.MTMVRefreshEnum.RefreshTrigger;
+import org.apache.doris.nereids.StatementContext;
import org.apache.doris.persist.AlterMTMV;
import org.apache.doris.persist.EditLog;
import org.apache.doris.persist.EditLog.EditLogItem;
import org.apache.doris.persist.OperationType;
import org.apache.doris.persist.gson.GsonUtils;
+import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.qe.SessionVariable;
import org.apache.doris.thrift.TStorageType;
import com.google.common.collect.Lists;
@@ -433,6 +437,230 @@ public class MTMVTest {
Mockito.verify(editLogItem).await();
}
+ @Test
+ public void testRefreshPublishAdvancesCacheGeneration() {
+ MTMVCacheManager manager = new MTMVCacheManager();
+ HookedMTMV mtmv = buildHookedMTMV();
+ MTMVCache refreshedGuarded = Mockito.mock(MTMVCache.class);
+ MTMVCache refreshedUnguarded = Mockito.mock(MTMVCache.class);
+ mtmv.refreshGuardedCache = refreshedGuarded;
+ mtmv.refreshUnguardedCache = refreshedUnguarded;
+ long generationBefore = Deencapsulation.getField(mtmv,
"rewriteCacheGeneration");
+
+ Env env = mockEnv(manager);
+ try (MockedStatic<Env> mockedEnv = Mockito.mockStatic(Env.class)) {
+ mockedEnv.when(Env::getCurrentEnv).thenReturn(env);
+
Assertions.assertTrue(mtmv.addTaskResult(buildSuccessTaskResult(mtmv), false));
+ }
+
+ long generationAfter = Deencapsulation.getField(mtmv,
"rewriteCacheGeneration");
+ Assertions.assertEquals(generationBefore + 1, generationAfter);
+ Assertions.assertSame(refreshedGuarded,
manager.getIfPresent(mtmv.getId(), true));
+ Assertions.assertSame(refreshedUnguarded,
manager.getIfPresent(mtmv.getId(), false));
+ }
+
+ @Test
+ public void testRefreshSkipsPlanBuildWhenCacheDisabled() {
+ int originalMaxSize = Config.mtmv_cache_manage_num;
+ try {
+ Config.mtmv_cache_manage_num = 0;
+ MTMVCacheManager manager = new MTMVCacheManager();
+ HookedMTMV mtmv = buildHookedMTMV();
+ mtmv.refreshGuardedCache = Mockito.mock(MTMVCache.class);
+ mtmv.refreshUnguardedCache = Mockito.mock(MTMVCache.class);
+ long generationBefore = Deencapsulation.getField(mtmv,
"rewriteCacheGeneration");
+
+ Env env = mockEnv(manager);
+ try (MockedStatic<Env> mockedEnv = Mockito.mockStatic(Env.class)) {
+ mockedEnv.when(Env::getCurrentEnv).thenReturn(env);
+
Assertions.assertTrue(mtmv.addTaskResult(buildSuccessTaskResult(mtmv), false));
+ }
+
+ // The generation/invalidation transition still happens, but
neither plan was built.
+ long generationAfter = Deencapsulation.getField(mtmv,
"rewriteCacheGeneration");
+ Assertions.assertEquals(generationBefore + 1, generationAfter);
+ Assertions.assertEquals(0, mtmv.refreshBuildCount);
+ Assertions.assertNull(manager.getIfPresent(mtmv.getId(), true));
+ Assertions.assertNull(manager.getIfPresent(mtmv.getId(), false));
+ } finally {
+ Config.mtmv_cache_manage_num = originalMaxSize;
+ }
+ }
+
+ @Test
+ public void testDisabledCacheReusesPlanWithinSameStatement() throws
Exception {
+ int originalMaxSize = Config.mtmv_cache_manage_num;
+ try {
+ Config.mtmv_cache_manage_num = 0;
+ MTMVCacheManager manager = new MTMVCacheManager();
+ Assertions.assertFalse(manager.isEnabled());
+
+ HookedMTMV mtmv = buildHookedMTMV();
+ MTMVCache plan = Mockito.mock(MTMVCache.class);
+ mtmv.lazyCaches.add(plan);
+
+ ConnectContext context = mockConnectContext();
+ StatementContext statementContext = new StatementContext();
+
Mockito.when(context.getStatementContext()).thenReturn(statementContext);
+
+ Env env = mockEnv(manager);
+ try (MockedStatic<Env> mockedEnv = Mockito.mockStatic(Env.class)) {
+ mockedEnv.when(Env::getCurrentEnv).thenReturn(env);
+ MTMVCache first = mtmv.getOrGenerateCache(context);
+ MTMVCache second = mtmv.getOrGenerateCache(context);
+ Assertions.assertSame(plan, first);
+ Assertions.assertSame(first, second);
+ }
+
+ Assertions.assertEquals(1, mtmv.lazyBuildCount);
+ Assertions.assertNull(manager.getIfPresent(mtmv.getId(), false));
+ Assertions.assertSame(plan,
statementContext.getQueryLocalMtmvCache(mtmv.getId(), false));
+ } finally {
+ Config.mtmv_cache_manage_num = originalMaxSize;
+ }
+ }
+
+ @Test
+ public void testPausedBuilderCannotRepublishPreRefreshPlan() {
+ MTMVCacheManager manager = new MTMVCacheManager();
+ HookedMTMV mtmv = buildHookedMTMV();
+ MTMVCache prePublishPlan = Mockito.mock(MTMVCache.class);
+ MTMVCache rebuiltPlan = Mockito.mock(MTMVCache.class);
+ MTMVCache refreshedUnguarded = Mockito.mock(MTMVCache.class);
+ mtmv.lazyCaches.add(prePublishPlan);
+ mtmv.lazyCaches.add(rebuiltPlan);
+ mtmv.refreshGuardedCache = Mockito.mock(MTMVCache.class);
+ mtmv.refreshUnguardedCache = refreshedUnguarded;
+
+ Env env = mockEnv(manager);
+ try (MockedStatic<Env> mockedEnv = Mockito.mockStatic(Env.class)) {
+ mockedEnv.when(Env::getCurrentEnv).thenReturn(env);
+ // The builder snapshotted the generation and is now paused
outside the MV lock: the refresh
+ // publishes its pair and the fresh entry is then evicted before
the builder resumes.
+ mtmv.duringLazyBuild = () -> {
+
Assertions.assertTrue(mtmv.addTaskResult(buildSuccessTaskResult(mtmv), false));
+ Assertions.assertSame(refreshedUnguarded,
manager.getIfPresent(mtmv.getId(), false));
+ manager.invalidate(mtmv.getId());
+ };
+ MTMVCache published =
mtmv.getOrGenerateCache(mockConnectContext());
+
+ Assertions.assertSame(rebuiltPlan, published);
+ Assertions.assertSame(rebuiltPlan,
manager.getIfPresent(mtmv.getId(), false));
+ Assertions.assertNotSame(prePublishPlan,
manager.getIfPresent(mtmv.getId(), false));
+ }
+ }
+
+ @Test
+ public void testTaskCompletionDoesNotPublishForDroppedMv() {
+ MTMVCacheManager manager = new MTMVCacheManager();
+ HookedMTMV mtmv = buildHookedMTMV();
+ mtmv.refreshGuardedCache = Mockito.mock(MTMVCache.class);
+ mtmv.refreshUnguardedCache = Mockito.mock(MTMVCache.class);
+
+ Env env = mockEnv(manager);
+ try (MockedStatic<Env> mockedEnv = Mockito.mockStatic(Env.class)) {
+ mockedEnv.when(Env::getCurrentEnv).thenReturn(env);
+ manager.put(mtmv.getId(), true, Mockito.mock(MTMVCache.class));
+ // The task builds its caches outside the MV lock; the drop lands
in that window.
+ mtmv.duringRefreshBuild = mtmv::markDropped;
+
Assertions.assertTrue(mtmv.addTaskResult(buildSuccessTaskResult(mtmv), false));
+
+ Assertions.assertTrue(mtmv.isDropped);
+ Assertions.assertNull(manager.getIfPresent(mtmv.getId(), true));
+ Assertions.assertNull(manager.getIfPresent(mtmv.getId(), false));
+ }
+ }
+
+ @Test
+ public void testDropStopsPausedBuilderFromPublishing() {
+ MTMVCacheManager manager = new MTMVCacheManager();
+ HookedMTMV mtmv = buildHookedMTMV();
+ MTMVCache builtPlan = Mockito.mock(MTMVCache.class);
+ mtmv.lazyCaches.add(builtPlan);
+
+ Env env = mockEnv(manager);
+ try (MockedStatic<Env> mockedEnv = Mockito.mockStatic(Env.class)) {
+ mockedEnv.when(Env::getCurrentEnv).thenReturn(env);
+ manager.put(mtmv.getId(), true, Mockito.mock(MTMVCache.class));
+ mtmv.duringLazyBuild = mtmv::markDropped;
+ MTMVCache generated =
mtmv.getOrGenerateCache(mockConnectContext());
+
+ Assertions.assertSame(builtPlan, generated);
+ Assertions.assertNull(manager.getIfPresent(mtmv.getId(), true));
+ Assertions.assertNull(manager.getIfPresent(mtmv.getId(), false));
+ }
+ }
+
+ private HookedMTMV buildHookedMTMV() {
+ HookedMTMV mtmv = configureMTMV(new HookedMTMV());
+ mtmv.getIvmInfo();
+ return mtmv;
+ }
+
+ private Env mockEnv(MTMVCacheManager manager) {
+ Env env = Mockito.mock(Env.class);
+ EditLog editLog = Mockito.mock(EditLog.class);
+ Mockito.when(env.getEditLog()).thenReturn(editLog);
+
Mockito.when(env.getMtmvService()).thenReturn(Mockito.mock(MTMVService.class));
+ Mockito.when(env.getMtmvCacheManager()).thenReturn(manager);
+ Mockito.when(editLog.submitEdit(Mockito.anyShort(), Mockito.any()))
+ .thenReturn(Mockito.mock(EditLogItem.class));
+ return env;
+ }
+
+ private ConnectContext mockConnectContext() {
+ ConnectContext context = Mockito.mock(ConnectContext.class);
+ SessionVariable sessionVariable = Mockito.mock(SessionVariable.class);
+ Mockito.when(context.getSessionVariable()).thenReturn(sessionVariable);
+
Mockito.when(sessionVariable.getAffectQueryResultInPlanVariables()).thenReturn(Map.of());
+ return context;
+ }
+
+ private AlterMTMV buildSuccessTaskResult(MTMV mtmv) {
+ MTMVRelation relation = mtmv.getRelation();
+ MTMVTask task = new MTMVTask(mtmv, relation, null);
+ task.setStatus(TaskStatus.SUCCESS);
+ AlterMTMV alterMTMV = new AlterMTMV(new TableNameInfo("db1", "mv1"),
MTMVAlterOpType.ADD_TASK);
+ alterMTMV.setTask(task);
+ alterMTMV.setRelation(relation);
+ alterMTMV.setPartitionSnapshots(Map.of());
+ return alterMTMV;
+ }
+
+ /**
+ * Runs a hook inside the lock-free cache build so a refresh or a drop can
be interleaved with an
+ * in-flight build deterministically, without threads.
+ */
+ private static class HookedMTMV extends MTMV {
+ private final List<MTMVCache> lazyCaches = Lists.newArrayList();
+ private Runnable duringRefreshBuild;
+ private Runnable duringLazyBuild;
+ private MTMVCache refreshGuardedCache;
+ private MTMVCache refreshUnguardedCache;
+ private int lazyBuildCount;
+ private int refreshBuildCount;
+
+ @Override
+ protected MTMVCache createRewriteCache(ConnectContext currentContext,
boolean needLock,
+ boolean addSessionVarGuard) {
+ // needLock is true only on the refresh path, false on the lazy
query path.
+ Runnable hook = needLock ? duringRefreshBuild : duringLazyBuild;
+ if (needLock) {
+ duringRefreshBuild = null;
+ } else {
+ duringLazyBuild = null;
+ }
+ if (hook != null) {
+ hook.run();
+ }
+ if (needLock) {
+ refreshBuildCount++;
+ return addSessionVarGuard ? refreshGuardedCache :
refreshUnguardedCache;
+ }
+ return lazyCaches.get(Math.min(lazyBuildCount++, lazyCaches.size()
- 1));
+ }
+ }
+
private void replayAlterMvProperties(MTMV mtmv, Map<String, String>
properties) {
AlterMTMV alterMTMV = new AlterMTMV(
new TableNameInfo("db", "mv"), MTMVAlterOpType.ALTER_PROPERTY);
@@ -454,7 +682,10 @@ public class MTMVTest {
}
private MTMV buildSerializableMTMV() {
- MTMV mtmv = new MTMV();
+ return configureMTMV(new MTMV());
+ }
+
+ private <T extends MTMV> T configureMTMV(T mtmv) {
mtmv.setId(1L);
mtmv.setQualifiedDbName("db1");
mtmv.setRefreshInfo(buildMTMVRefreshInfo(mtmv));
diff --git a/regression-test/data/mtmv_p0/test_mtmv_cache_proc.out
b/regression-test/data/mtmv_p0/test_mtmv_cache_proc.out
new file mode 100644
index 00000000000..ef0bfb7ea65
--- /dev/null
+++ b/regression-test/data/mtmv_p0/test_mtmv_cache_proc.out
@@ -0,0 +1,5 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !mtmv_cache_dir --
+hot Top hot mtmv cache entries
+stat Global cache stats
+
diff --git a/regression-test/suites/mtmv_p0/test_mtmv_cache_proc.groovy
b/regression-test/suites/mtmv_p0/test_mtmv_cache_proc.groovy
new file mode 100644
index 00000000000..ea64ea65e57
--- /dev/null
+++ b/regression-test/suites/mtmv_p0/test_mtmv_cache_proc.groovy
@@ -0,0 +1,83 @@
+// 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,nonConcurrent") {
+ def dbName = "regression_test_mtmv_p0"
+ def mvName = "mtmv_cache_proc_mv"
+
+ sql """drop materialized view if exists ${mvName}"""
+ sql """drop table if exists t_test_mtmv_cache_proc_user"""
+
+ sql """
+ CREATE TABLE IF NOT EXISTS t_test_mtmv_cache_proc_user (
+ event_day DATE,
+ id BIGINT,
+ username VARCHAR(20)
+ )
+ DISTRIBUTED BY HASH(id) BUCKETS 2
+ PROPERTIES ('replication_num' = '1');
+ """
+
+ // The '/mtmv_cache' directory listing is a fixed pair of children.
+ order_qt_mtmv_cache_dir """SHOW PROC '/mtmv_cache'"""
+
+ // SHOW PROC '/mtmv_cache/stat' returns the same KV rows regardless of
cache contents; the
+ // values themselves are runtime-dependent, so only the key set can be
checked here.
+ def statRows = sql """SHOW PROC '/mtmv_cache/stat'"""
+ def statKeys = statRows.collect { it[0] }
+ ["size", "hitCount", "missCount", "evictionCount", "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 t_test_mtmv_cache_proc_user;
+ """
+ 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 t_test_mtmv_cache_proc_user"""
+
+ // hot proc: 5 columns; our MV MUST appear with its real DbName/MvName.
+ def hotRows = sql """SHOW PROC '/mtmv_cache/hot'"""
+ assertTrue(!hotRows.isEmpty(),
+ "hot cache should contain at least one entry after the MV was
queried")
+ assertEquals(5, hotRows[0].size())
+ def mvRow = hotRows.find { it[2] == mvName }
+ assertNotNull(mvRow, "MV ${mvName} should be visible in /mtmv_cache/hot
after query")
+ assertEquals(dbName, mvRow[1])
+ assertTrue(mvRow[3] == "Yes" || mvRow[3] == "No")
+ assertTrue((mvRow[4] as Long) >= 0L, "IdleMs must be non-negative")
+
+ // mtmv_cache_hot_show_num caps the row count.
+ def originalCap = sql """ADMIN SHOW FRONTEND CONFIG LIKE
'mtmv_cache_hot_show_num'"""
+ def originalCapVal = originalCap.isEmpty() ? "500" : originalCap[0][1]
+ try {
+ sql """ADMIN SET FRONTEND CONFIG ('mtmv_cache_hot_show_num' = '1')"""
+ def capped = sql """SHOW PROC '/mtmv_cache/hot'"""
+ assertEquals(1, capped.size(),
+ "hot row count should be exactly 1 after capping to 1, got
${capped.size()}")
+ } finally {
+ sql """ADMIN SET FRONTEND CONFIG ('mtmv_cache_hot_show_num' =
'${originalCapVal}')"""
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
