This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new ea2df20d2fc [fix](mtmv) Eliminate redundant filters in nested MV
candidate defini… (#69381)
ea2df20d2fc is described below
commit ea2df20d2fce98342c447961363bbd84c4038278
Author: foxtail463 <[email protected]>
AuthorDate: Thu Oct 29 09:46:21 2026 +0800
[fix](mtmv) Eliminate redundant filters in nested MV candidate defini…
(#69381)
Cherry-picked from #63639
---------
Co-authored-by: yangtao555 <[email protected]>
---
.../main/java/org/apache/doris/catalog/MTMV.java | 128 +++--
.../main/java/org/apache/doris/mtmv/MTMVCache.java | 36 +-
.../java/org/apache/doris/mtmv/MTMVPlanUtil.java | 17 +-
.../org/apache/doris/nereids/StatementContext.java | 11 +
.../nereids/rules/analysis/ExpressionAnalyzer.java | 15 +-
.../exploration/mv/MaterializationContext.java | 7 +-
.../mv/MaterializedViewPredicateCollector.java | 124 ++++
.../nereids/rules/exploration/mv/Predicates.java | 16 +
.../nereids/rules/exploration/mv/StructInfo.java | 41 ++
.../nereids/rules/rewrite/PullUpPredicates.java | 2 +-
.../doris/mtmv/MTMVOutputPredicatesTest.java | 223 ++++++++
.../org/apache/doris/nereids/mv/MTMVCacheTest.java | 147 +++++
.../nereids/mv/RelationImpliedPredicateTest.java | 627 +++++++++++++++++++++
.../relation_implied_predicate_nested_mtmv.out | 8 +
.../relation_implied_predicate_nested_mtmv.groovy | 120 ++++
15 files changed, 1459 insertions(+), 63 deletions(-)
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 9933481475b..24fe0adb30f 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
@@ -92,6 +92,8 @@ public class MTMV extends OlapTable {
private MTMVCache cacheWithGuard;
// Cache without SessionVarGuardExpr: used when query session variables
match MV creation variables
private MTMVCache cacheWithoutGuard;
+ // Increased on invalidation to reject caches built against an earlier
definition or refresh.
+ private transient long rewriteCacheGeneration;
private long schemaChangeVersion;
@SerializedName(value = "sv")
private Map<String, String> sessionVariables;
@@ -189,6 +191,7 @@ public class MTMV extends OlapTable {
// only can update state, refresh state will be change by add task
this.schemaChangeVersion++;
this.refreshSnapshot = new MTMVRefreshSnapshot();
+ invalidateRewriteCache();
return this.status.updateStateAndDetail(newStatus);
} finally {
writeMvUnlock();
@@ -196,15 +199,7 @@ public class MTMV extends OlapTable {
}
public void processBaseViewChange(String schemaChangeDetail) {
- writeMvLock();
- try {
- this.schemaChangeVersion++;
- this.status.setState(MTMVState.SCHEMA_CHANGE);
- this.status.setSchemaChangeDetail(schemaChangeDetail);
- this.refreshSnapshot = new MTMVRefreshSnapshot();
- } finally {
- writeMvUnlock();
- }
+ alterStatus(new MTMVStatus(MTMVState.SCHEMA_CHANGE,
schemaChangeDetail));
}
public boolean addTaskResult(MTMVTask task, MTMVRelation relation,
@@ -212,9 +207,16 @@ public class MTMV extends OlapTable {
MTMVCache mtmvCacheWithGuard = null;
MTMVCache mtmvCacheWithoutGuard = null;
boolean needUpdateCache = false;
+ long cacheGeneration = -1;
if (task.getStatus() == TaskStatus.SUCCESS && !Env.isCheckpointThread()
&& !Config.enable_check_compatibility_mode) {
needUpdateCache = true;
+ readMvLock();
+ try {
+ cacheGeneration = rewriteCacheGeneration;
+ } finally {
+ readMvUnlock();
+ }
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
@@ -222,12 +224,8 @@ public class MTMV extends OlapTable {
ConnectContext currentContext = ConnectContext.get();
// shouldn't do this while holding mvWriteLock
// TODO: these two cache compute share something same, can
be simplified in future
- mtmvCacheWithGuard = MTMVCache.from(this.getQuerySql(),
- MTMVPlanUtil.createMTMVContext(this,
MTMVPlanUtil.DISABLE_RULES_WHEN_GENERATE_MTMV_CACHE),
- true, true, currentContext, true);
- mtmvCacheWithoutGuard = MTMVCache.from(this.getQuerySql(),
- MTMVPlanUtil.createMTMVContext(this,
MTMVPlanUtil.DISABLE_RULES_WHEN_GENERATE_MTMV_CACHE),
- true, true, currentContext, false);
+ mtmvCacheWithGuard = createRewriteCache(currentContext,
true, true);
+ mtmvCacheWithoutGuard = createRewriteCache(currentContext,
true, false);
}
} catch (Throwable e) {
mtmvCacheWithGuard = null;
@@ -245,23 +243,31 @@ public class MTMV extends OlapTable {
name, task.getTaskId(),
task.getMtmvSchemaChangeVersion(), this.schemaChangeVersion);
return false;
}
+ boolean publishCache = false;
if (task.getStatus() == TaskStatus.SUCCESS) {
+ publishCache = needUpdateCache && cacheGeneration ==
rewriteCacheGeneration;
+ // A successful refresh changes the snapshot coverage and
supersedes in-flight builds.
+ invalidateRewriteCache();
this.status.setState(MTMVState.NORMAL);
this.status.setSchemaChangeDetail(null);
this.status.setRefreshState(MTMVRefreshState.SUCCESS);
this.relation = relation;
- if (needUpdateCache) {
- // Initialize cacheWithGuard, cacheWithoutGuard will be
lazily generated when needed
- this.cacheWithGuard = mtmvCacheWithGuard;
- // Clear the other cache to ensure consistency
- this.cacheWithoutGuard = mtmvCacheWithoutGuard;
- }
} else {
this.status.setRefreshState(MTMVRefreshState.FAIL);
}
this.jobInfo.addHistoryTask(task);
compatiblePctSnapshot(partitionSnapshots);
this.refreshSnapshot.updateSnapshots(partitionSnapshots,
getPartitionNames());
+ if (publishCache) {
+ // Publish after recording coverage: NORMAL after a partial
refresh does not prove
+ // that every stored partition already uses the current
definition.
+ if (mtmvCacheWithGuard != null) {
+ this.cacheWithGuard =
cacheWithValidOutputPredicates(mtmvCacheWithGuard);
+ }
+ if (mtmvCacheWithoutGuard != null) {
+ this.cacheWithoutGuard =
cacheWithValidOutputPredicates(mtmvCacheWithoutGuard);
+ }
+ }
Env.getCurrentEnv().getMtmvService()
.refreshComplete(this, relation, task);
return true;
@@ -286,6 +292,7 @@ public class MTMV extends OlapTable {
// snapshots so the next AUTO refresh rebuilds a complete
baseline with the new rules.
this.schemaChangeVersion++;
this.refreshSnapshot = new MTMVRefreshSnapshot();
+ invalidateRewriteCache();
}
}
return this.mvProperties;
@@ -424,38 +431,62 @@ public class MTMV extends OlapTable {
boolean sessionVarsMatch =
SessionVarGuardRewriter.checkSessionVariablesMatch(
currentSessionVars, this.sessionVariables);
- // Select appropriate cache based on session variable match
- readMvLock();
- try {
- if (sessionVarsMatch && cacheWithoutGuard != null) {
- return cacheWithoutGuard;
+ while (true) {
+ long cacheGeneration;
+ readMvLock();
+ try {
+ MTMVCache cached = sessionVarsMatch ? cacheWithoutGuard :
cacheWithGuard;
+ if (cached != null) {
+ return cached;
+ }
+ cacheGeneration = rewriteCacheGeneration;
+ } finally {
+ readMvUnlock();
}
- if (!sessionVarsMatch && cacheWithGuard != null) {
- return cacheWithGuard;
+
+ // Build outside the MV lock to avoid nesting it with table locks
during analysis.
+ MTMVCache generated = createRewriteCache(connectionContext, false,
!sessionVarsMatch);
+ writeMvLock();
+ try {
+ if (cacheGeneration != rewriteCacheGeneration) {
+ continue;
+ }
+ MTMVCache cached = sessionVarsMatch ? cacheWithoutGuard :
cacheWithGuard;
+ if (cached != null) {
+ return cached;
+ }
+ MTMVCache published =
cacheWithValidOutputPredicates(generated);
+ if (sessionVarsMatch) {
+ this.cacheWithoutGuard = published;
+ } else {
+ this.cacheWithGuard = published;
+ }
+ return published;
+ } finally {
+ writeMvUnlock();
}
- } 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 = MTMVCache.from(this.getQuerySql(),
- MTMVPlanUtil.createMTMVContext(this,
MTMVPlanUtil.DISABLE_RULES_WHEN_GENERATE_MTMV_CACHE),
- true, false, connectionContext, !sessionVarsMatch);
+ /** Invalidate both cache variants and reject any older build still in
progress. */
+ public void invalidateRewriteCache() {
writeMvLock();
try {
- if (sessionVarsMatch) {
- this.cacheWithoutGuard = mtmvCache;
- } else {
- this.cacheWithGuard = mtmvCache;
- }
- return mtmvCache;
+ rewriteCacheGeneration++;
+ cacheWithGuard = null;
+ cacheWithoutGuard = null;
} finally {
writeMvUnlock();
}
}
+ protected MTMVCache createRewriteCache(ConnectContext currentContext,
boolean needLock,
+ boolean addSessionVarGuard) {
+ return MTMVCache.from(this.getQuerySql(),
+ MTMVPlanUtil.createMTMVContext(this,
MTMVPlanUtil.DISABLE_RULES_WHEN_GENERATE_MTMV_CACHE),
+ true, needLock, currentContext, addSessionVarGuard);
+ }
+
public Map<String, String> getMvProperties() {
readMvLock();
try {
@@ -601,6 +632,15 @@ public class MTMV extends OlapTable {
this.mvRwLock.writeLock().unlock();
}
+ private MTMVCache cacheWithValidOutputPredicates(MTMVCache cache) {
+ // Callers hold the MV lock while publishing guarantees about the
stored partitions.
+ if (status.getState() == MTMVState.SCHEMA_CHANGE
+ ||
!refreshSnapshot.getPartitionSnapshots().keySet().containsAll(getPartitionNames()))
{
+ return cache.withoutOutputPredicates();
+ }
+ return cache;
+ }
+
// toString() is not easy to find where to call the method
public String toInfoString() {
final StringBuilder sb = new StringBuilder("MTMV{");
@@ -638,8 +678,8 @@ public class MTMV extends OlapTable {
Env.getCurrentEnv().getMtmvService().registerMTMV(this,
this.getDatabase().getId());
} catch (Throwable e) {
LOG.warn("MTMV compatible failed, dbName: {}, mvName: {}, errMsg:
{}", getDBName(), name, e.getMessage());
- status.setState(MTMVState.SCHEMA_CHANGE);
- status.setSchemaChangeDetail("compatible failed, please refresh or
recreate it, reason: " + e.getMessage());
+ alterStatus(new MTMVStatus(MTMVState.SCHEMA_CHANGE,
+ "compatible failed, please refresh or recreate it, reason:
" + e.getMessage()));
}
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVCache.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVCache.java
index 391a9546500..12108927648 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVCache.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVCache.java
@@ -32,6 +32,8 @@ import
org.apache.doris.nereids.rules.exploration.mv.MaterializedViewUtils;
import org.apache.doris.nereids.rules.exploration.mv.StructInfo;
import org.apache.doris.nereids.rules.rewrite.EliminateSort;
import org.apache.doris.nereids.rules.rewrite.MergeProjectable;
+import org.apache.doris.nereids.rules.rewrite.PullUpPredicates;
+import org.apache.doris.nereids.trees.expressions.Expression;
import org.apache.doris.nereids.trees.plans.Plan;
import
org.apache.doris.nereids.trees.plans.commands.ExplainCommand.ExplainLevel;
import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
@@ -44,12 +46,14 @@ import org.apache.doris.qe.OriginStatement;
import org.apache.doris.statistics.Statistics;
import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
+import java.util.Set;
/**
* The cache for materialized view cache
@@ -65,13 +69,17 @@ public class MTMVCache {
private final Plan originalFinalPlan;
private final Statistics statistics;
private final List<Pair<Plan, StructInfo>>
partRulesRewrittenPlanAndStructInfos;
+ // Guarantees expressed only in the definition's output slots, including
session-variable guards.
+ private final Set<Expression> outputPredicates;
public MTMVCache(Pair<Plan, StructInfo>
allRulesRewrittenPlanAndStructInfo, Plan originalFinalPlan,
- Statistics statistics, List<Pair<Plan, StructInfo>>
partRulesRewrittenPlanAndStructInfos) {
+ Statistics statistics, List<Pair<Plan, StructInfo>>
partRulesRewrittenPlanAndStructInfos,
+ Set<Expression> outputPredicates) {
this.allRulesRewrittenPlanAndStructInfo =
allRulesRewrittenPlanAndStructInfo;
this.originalFinalPlan = originalFinalPlan;
this.statistics = statistics;
this.partRulesRewrittenPlanAndStructInfos =
partRulesRewrittenPlanAndStructInfos;
+ this.outputPredicates = ImmutableSet.copyOf(outputPredicates);
}
public Pair<Plan, StructInfo> getAllRulesRewrittenPlanAndStructInfo() {
@@ -82,6 +90,16 @@ public class MTMVCache {
return originalFinalPlan;
}
+ public Set<Expression> getOutputPredicates() {
+ return outputPredicates;
+ }
+
+ /** Keep the definition plan without claiming that its predicates hold for
stored rows. */
+ public MTMVCache withoutOutputPredicates() {
+ return new MTMVCache(allRulesRewrittenPlanAndStructInfo,
originalFinalPlan,
+ statistics, partRulesRewrittenPlanAndStructInfos,
ImmutableSet.of());
+ }
+
public Statistics getStatistics() {
return statistics;
}
@@ -142,6 +160,19 @@ public class MTMVCache {
.orElse(rewritePlan);
Pair<Plan, StructInfo> finalPlanStructInfoPair =
constructPlanAndStructInfo(
addGuardRewritePlan, cascadesContext);
+ // Analysis records nondeterministic functions before constant
folding erases their origin.
+ // Replanning current_date(), for example, cannot establish a
guarantee about rows stored
+ // by an earlier refresh. Do not export output guarantees from
such a definition.
+ // Alias UDF definitions can also change between refresh and cache
reconstruction,
+ // even when the expanded function body is deterministic.
+ // A lower MV may have refreshed since this MV's rows were stored.
Derive guarantees
+ // from this definition alone, without importing the lower MV's
current output facts.
+ Set<Expression> outputPredicates =
(cascadesContext.getStatementContext().hasNondeterministic()
+ || cascadesContext.getStatementContext().hasAliasUdf())
+ ? ImmutableSet.of()
+ : finalPlanStructInfoPair.key().accept(new
PullUpPredicates(true, cascadesContext), null).stream()
+ .filter(predicate ->
!predicate.containsVolatileExpression())
+ .collect(ImmutableSet.toImmutableSet());
List<Pair<Plan, StructInfo>> tmpPlanUsedForRewrite = new
ArrayList<>();
for (Plan plan :
cascadesContext.getStatementContext().getTmpPlanForMvRewrite()) {
Plan addGuardplan = exprRewriter
@@ -150,7 +181,8 @@ public class MTMVCache {
tmpPlanUsedForRewrite.add(constructPlanAndStructInfo(addGuardplan,
cascadesContext));
}
return new MTMVCache(finalPlanStructInfoPair, addGuardRewritePlan,
needCost
- ? cascadesContext.getMemo().getRoot().getStatistics() :
null, tmpPlanUsedForRewrite);
+ ? cascadesContext.getMemo().getRoot().getStatistics() :
null, tmpPlanUsedForRewrite,
+ outputPredicates);
} finally {
createCacheContext.getStatementContext().setForceRecordTmpPlan(false);
mvSqlStatementContext.setForceRecordTmpPlan(false);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPlanUtil.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPlanUtil.java
index 8b6e3aa264a..b99787aca96 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPlanUtil.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPlanUtil.java
@@ -96,11 +96,9 @@ import javax.annotation.Nullable;
public class MTMVPlanUtil {
private static final Logger LOG = LogManager.getLogger(MTMVPlanUtil.class);
- // The rules should be disabled when generate MTMV cache
- // Because these rules may change the plan structure and cause the plan
can not match the mv
- // this is mainly for final CBO phase rewrite, pre RBO phase does not need
to consider, because
- // maintain tmp plan alone for rewrite when pre RBO rewrite
- public static final List<RuleType> DISABLE_RULES_WHEN_GENERATE_MTMV_CACHE
= ImmutableList.of(
+ // These rules keep the generated query plan aligned with the MV
definition while avoiding rewrites
+ // that can change its shape.
+ public static final List<RuleType> DISABLE_RULES_WHEN_RUN_MTMV_TASK =
ImmutableList.of(
RuleType.COMPRESSED_MATERIALIZE_AGG,
RuleType.COMPRESSED_MATERIALIZE_SORT,
RuleType.ELIMINATE_CONST_JOIN_CONDITION,
@@ -113,8 +111,13 @@ public class MTMVPlanUtil {
RuleType.ELIMINATE_GROUP_BY,
RuleType.SALT_JOIN
);
- // The rules should be disabled when run MTMV task
- public static final List<RuleType> DISABLE_RULES_WHEN_RUN_MTMV_TASK =
DISABLE_RULES_WHEN_GENERATE_MTMV_CACHE;
+
+ // Preserve the definition's scans and filters even when source partitions
currently have no data.
+ // Empty-partition pruning would erase predicates needed for MV matching
and output guarantees.
+ public static final List<RuleType> DISABLE_RULES_WHEN_GENERATE_MTMV_CACHE
= ImmutableList.<RuleType>builder()
+ .addAll(DISABLE_RULES_WHEN_RUN_MTMV_TASK)
+ .add(RuleType.PRUNE_EMPTY_PARTITION)
+ .build();
public static ConnectContext createMTMVContext(MTMV mtmv, List<RuleType>
disableRules) {
ConnectContext ctx = createBasicMvContext(null, disableRules,
mtmv.getSessionVariables());
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 1e76e5dde72..9167ed889bc 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
@@ -155,6 +155,9 @@ public class StatementContext implements Closeable {
private boolean isDelete = false;
private boolean hasNondeterministic = false;
+ // Alias UDF calls disappear after expansion; retain their definition
dependency.
+ private boolean hasAliasUdf = false;
+
// hasUnknownColStats true if any column stats in the tables used by this
sql is
// unknown
// the algorithm to derive plan when column stats are unknown is
implemented in
@@ -550,6 +553,14 @@ public class StatementContext implements Closeable {
return hasNondeterministic;
}
+ public void recordAliasUdf() {
+ hasAliasUdf = true;
+ }
+
+ public boolean hasAliasUdf() {
+ return hasAliasUdf;
+ }
+
public ConnectContext getConnectContext() {
return connectContext;
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/ExpressionAnalyzer.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/ExpressionAnalyzer.java
index 77ac4efced5..4467802f8d4 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/ExpressionAnalyzer.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/ExpressionAnalyzer.java
@@ -27,7 +27,6 @@ import org.apache.doris.common.util.Util;
import org.apache.doris.mysql.MysqlCommand;
import org.apache.doris.nereids.CascadesContext;
import org.apache.doris.nereids.SqlCacheContext;
-import org.apache.doris.nereids.StatementContext;
import org.apache.doris.nereids.analyzer.Scope;
import org.apache.doris.nereids.analyzer.UnboundAlias;
import org.apache.doris.nereids.analyzer.UnboundFunction;
@@ -200,7 +199,12 @@ public class ExpressionAnalyzer extends
SubExprAnalyzer<ExpressionRewriteContext
/** analyze */
public Expression analyze(Expression expression, ExpressionRewriteContext
context) {
- return expression.accept(this, context);
+ Expression analyzed = expression.accept(this, context);
+ // Inspect the bound tree to include parser-bound functions and
expanded alias UDF bodies.
+ if (context != null && analyzed.containsNondeterministic()) {
+
context.cascadesContext.getStatementContext().setHasNondeterministic(true);
+ }
+ return analyzed;
}
@Override
@@ -602,14 +606,13 @@ public class ExpressionAnalyzer extends
SubExprAnalyzer<ExpressionRewriteContext
});
}
+ if (builder instanceof AliasUdfBuilder && context != null) {
+ context.cascadesContext.getStatementContext().recordAliasUdf();
+ }
Pair<? extends Expression, ? extends BoundFunction> buildResult =
builder.build(functionName, arguments);
buildResult.second.checkOrderExprIsValid();
Optional<SqlCacheContext> sqlCacheContext = Optional.empty();
- if (!buildResult.second.isDeterministic() && context != null) {
- StatementContext statementContext =
context.cascadesContext.getStatementContext();
- statementContext.setHasNondeterministic(true);
- }
if (wantToParseSqlFromSqlCache) {
sqlCacheContext =
context.cascadesContext.getStatementContext().getSqlCacheContext();
if (builder instanceof AliasUdfBuilder
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/MaterializationContext.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/MaterializationContext.java
index 2ab84c2108b..72c1238165c 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/MaterializationContext.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/MaterializationContext.java
@@ -122,9 +122,10 @@ public abstract class MaterializationContext {
this.enableRecordFailureDetail = parsedStatement != null &&
parsedStatement.isExplain()
&& ExplainLevel.MEMO_PLAN ==
parsedStatement.getExplainOptions().getExplainLevel();
// Construct materialization struct info, catch exception which may
cause planner roll back
- this.structInfo = structInfo == null
- ? constructStructInfo(plan, originalPlan,
cascadesContext).orElseGet(() -> null)
- : structInfo;
+ this.structInfo = (structInfo == null
+ ? constructStructInfo(plan, originalPlan, cascadesContext)
+ : Optional.of(structInfo))
+ .map(info ->
info.withoutRedundantMvFilters(cascadesContext)).orElse(null);
this.available = this.structInfo != null;
if (available) {
this.planOutputShuttledExpressions =
this.structInfo.getPlanOutputShuttledExpressions();
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/MaterializedViewPredicateCollector.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/MaterializedViewPredicateCollector.java
new file mode 100644
index 00000000000..eadb0e7a5ac
--- /dev/null
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/MaterializedViewPredicateCollector.java
@@ -0,0 +1,124 @@
+// 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.nereids.rules.exploration.mv;
+
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.MTMV;
+import org.apache.doris.mtmv.MTMVCache;
+import org.apache.doris.nereids.CascadesContext;
+import org.apache.doris.nereids.rules.rewrite.PullUpPredicates;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
+import org.apache.doris.nereids.util.ExpressionUtils;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableSet;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * Derive predicates for MV matching using guarantees on materialized table
outputs.
+ *
+ * <p>The definition cache owns the guarantees; each visit binds them to this
scan's actual slots.
+ * No expression state is attached to a scan, so ordinary scans, replacement
scans and deep copies
+ * follow the same path. PullUpPredicates retains the operator boundaries,
including null extension
+ * by outer joins and columns removed by projections or aggregation.
+ */
+public final class MaterializedViewPredicateCollector extends PullUpPredicates
{
+ private static final Logger LOG =
LogManager.getLogger(MaterializedViewPredicateCollector.class);
+
+ private final CascadesContext cascadesContext;
+
+ public MaterializedViewPredicateCollector(CascadesContext cascadesContext)
{
+ super(true, cascadesContext);
+ this.cascadesContext = cascadesContext;
+ }
+
+ /** Derive guarantees on a filter's input for the current MV matching
analysis. */
+ public Set<Expression> collect(Plan plan) {
+ return plan.accept(this, null).stream()
+ .filter(predicate -> !predicate.containsVolatileExpression())
+ .collect(ImmutableSet.toImmutableSet());
+ }
+
+ @Override
+ public ImmutableSet<Expression> visitLogicalOlapScan(LogicalOlapScan scan,
Void context) {
+ return cacheOrElse(scan, () -> {
+ if (!(scan.getTable() instanceof MTMV)
+ || scan.getSelectedIndexId() !=
scan.getTable().getBaseIndexId()) {
+ return ImmutableSet.of();
+ }
+ MTMV mtmv = (MTMV) scan.getTable();
+ MTMVCache cache;
+ try {
+ cache =
mtmv.getOrGenerateCache(cascadesContext.getConnectContext());
+ } catch (Exception exception) {
+ // A stored MV remains readable when its definition can no
longer be planned.
+ // As with MV data-trait derivation, unavailable metadata
supplies no rewrite proof.
+ LOG.warn("Cannot derive output predicates for materialized
view {}", mtmv.getName(), exception);
+ return ImmutableSet.of();
+ }
+ if (cache.getOutputPredicates().isEmpty()) {
+ return ImmutableSet.of();
+ }
+ return
ImmutableSet.copyOf(ExpressionUtils.replace(cache.getOutputPredicates(),
+ mapDefinitionOutputs(cache, scan)));
+ });
+ }
+
+ private static Map<Slot, Slot> mapDefinitionOutputs(MTMVCache cache,
LogicalOlapScan scan) {
+ List<Slot> definitionOutputs =
cache.getOriginalFinalPlan().getOutput().stream()
+ .filter(slot -> slot instanceof SlotReference &&
((SlotReference) slot).isVisible())
+ .collect(Collectors.toList());
+ List<Column> physicalColumns = scan.getTable().getBaseSchema(false);
+ Preconditions.checkState(definitionOutputs.size() ==
physicalColumns.size(),
+ "materialized view definition output size %s does not match
visible schema size %s",
+ definitionOutputs.size(), physicalColumns.size());
+
+ Map<String, Slot> scanColumns = new HashMap<>();
+ for (Slot slot : scan.getOutput()) {
+ if (slot instanceof SlotReference) {
+ SlotReference reference = (SlotReference) slot;
+ // Hidden IVM columns, derived variant paths and virtual
columns are not definition outputs.
+ if (reference.isVisible() && reference.getSubPath().isEmpty())
{
+ reference.getOriginalColumn().ifPresent(column ->
+
scanColumns.put(column.getName().toLowerCase(Locale.ROOT), reference));
+ }
+ }
+ }
+ Map<Slot, Slot> outputMapping = new HashMap<>();
+ for (int i = 0; i < physicalColumns.size(); i++) {
+ String columnName = physicalColumns.get(i).getName();
+ Slot scanSlot =
Preconditions.checkNotNull(scanColumns.get(columnName.toLowerCase(Locale.ROOT)),
+ "materialized view scan does not output physical column
%s", columnName);
+ // Repeated definition outputs need one representative, not all
combinations of scan slots.
+ outputMapping.putIfAbsent(definitionOutputs.get(i), scanSlot);
+ }
+ return outputMapping;
+ }
+}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/Predicates.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/Predicates.java
index f22429dc08a..10088fd1b7a 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/Predicates.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/Predicates.java
@@ -424,6 +424,22 @@ public class Predicates {
return ImmutableMap.copyOf(remainingPredicates);
}
+ /** Whether output guarantees prove that a filter predicate is redundant
at its input. */
+ static boolean isImpliedByOutput(Set<Expression> outputPredicates,
Expression predicate) {
+ if (outputPredicates.isEmpty() ||
predicate.containsVolatileExpression()) {
+ return false;
+ }
+ if (outputPredicates.contains(predicate)) {
+ return true;
+ }
+ try {
+ return impliesByDnf(ExpressionUtils.and(outputPredicates),
predicate);
+ } catch (DnfBranchOverflowException exception) {
+ // Keep the filter when proving it would exceed the existing
implication budget.
+ return false;
+ }
+ }
+
private static boolean impliesByDnf(Expression source, Expression target) {
// Check whether source => target.
List<Set<Expression>> sourceBranches = extractDnfBranches(source);
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/StructInfo.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/StructInfo.java
index d4e0e3f5003..b53468c5725 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/StructInfo.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/StructInfo.java
@@ -17,6 +17,7 @@
package org.apache.doris.nereids.rules.exploration.mv;
+import org.apache.doris.catalog.MTMV;
import org.apache.doris.common.Pair;
import org.apache.doris.mtmv.BaseColInfo;
import org.apache.doris.mtmv.BaseTableInfo;
@@ -65,8 +66,10 @@ import
org.apache.doris.nereids.trees.plans.logical.LogicalWindow;
import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanVisitor;
import org.apache.doris.nereids.util.ExpressionUtils;
+import org.apache.doris.nereids.util.PlanUtils;
import com.google.common.collect.HashMultimap;
+import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Multimap;
import com.google.common.collect.Sets;
@@ -338,6 +341,44 @@ public class StructInfo {
relationBitSet, null, null, planOutputShuttledExpressions);
}
+ /**
+ * Simplify a materialization definition using its inputs' current
guarantees.
+ * Query predicates must remain available for compensation when a
candidate supplies stale data.
+ * The result belongs to the current query, not the definition cache: an
input MV can refresh
+ * independently and invalidate the reason why a definition filter was
redundant.
+ */
+ public StructInfo withoutRedundantMvFilters(CascadesContext
cascadesContext) {
+ Plan normalized = removeRedundantMvFilters(topPlan, cascadesContext);
+ return normalized == topPlan ? this : of(normalized, originalPlan,
cascadesContext);
+ }
+
+ private static Plan removeRedundantMvFilters(Plan plan, CascadesContext
cascadesContext) {
+ if (!plan.anyMatch(node -> node instanceof LogicalOlapScan
+ && ((LogicalOlapScan) node).getTable() instanceof MTMV)) {
+ return plan;
+ }
+ MaterializedViewPredicateCollector collector = new
MaterializedViewPredicateCollector(cascadesContext);
+ Plan normalized = plan.accept(new DefaultPlanRewriter<Void>() {
+ @Override
+ public Plan visitLogicalFilter(LogicalFilter<? extends Plan>
filter, Void context) {
+ filter = visitChildren(this, filter, context);
+ // Prove redundancy at the filter's own input, before a parent
Project can discard
+ // the predicate column. Facts from the filter itself must not
participate in this proof.
+ Set<Expression> guaranteed = collector.collect(filter.child());
+ Set<Expression> remaining = filter.getConjuncts().stream()
+ .filter(predicate ->
!Predicates.isImpliedByOutput(guaranteed, predicate))
+ .collect(ImmutableSet.toImmutableSet());
+ if (remaining.size() == filter.getConjuncts().size()) {
+ return filter;
+ }
+ return PlanUtils.filterOrSelf(remaining, filter.child());
+ }
+ }, null);
+ // This equivalent tree is only used to build the matching graph; the
memo is not modified.
+ // Removing a root Filter must not assign its child's group to the
matched expression.
+ return normalized == plan ? plan :
normalized.withGroupExpression(plan.getGroupExpression());
+ }
+
public List<CatalogRelation> getRelations() {
return relations;
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/PullUpPredicates.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/PullUpPredicates.java
index 941cd07f20a..c97ab95bb39 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/PullUpPredicates.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/PullUpPredicates.java
@@ -364,7 +364,7 @@ public class PullUpPredicates extends
PlanVisitor<ImmutableSet<Expression>, Void
});
}
- private ImmutableSet<Expression> cacheOrElse(Plan plan,
Supplier<ImmutableSet<Expression>> predicatesSupplier) {
+ protected ImmutableSet<Expression> cacheOrElse(Plan plan,
Supplier<ImmutableSet<Expression>> predicatesSupplier) {
ImmutableSet<Expression> predicates = cache.get(plan);
if (predicates != null) {
return predicates;
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVOutputPredicatesTest.java
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVOutputPredicatesTest.java
new file mode 100644
index 00000000000..bd4055ed1e1
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVOutputPredicatesTest.java
@@ -0,0 +1,223 @@
+// 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.Column;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.KeysType;
+import org.apache.doris.catalog.MTMV;
+import org.apache.doris.catalog.MaterializedIndex;
+import org.apache.doris.catalog.Partition;
+import org.apache.doris.catalog.PrimitiveType;
+import org.apache.doris.catalog.SinglePartitionInfo;
+import org.apache.doris.common.Pair;
+import org.apache.doris.common.jmockit.Deencapsulation;
+import org.apache.doris.job.common.TaskStatus;
+import org.apache.doris.job.extensions.mtmv.MTMVTask;
+import org.apache.doris.job.extensions.mtmv.MTMVTask.MTMVTaskTriggerMode;
+import org.apache.doris.job.extensions.mtmv.MTMVTaskContext;
+import org.apache.doris.mtmv.MTMVRefreshEnum.BuildMode;
+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.trees.expressions.Alias;
+import org.apache.doris.nereids.trees.expressions.ExprId;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.GreaterThan;
+import org.apache.doris.nereids.trees.expressions.literal.IntegerLiteral;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.RelationId;
+import org.apache.doris.nereids.trees.plans.logical.LogicalOneRowRelation;
+import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.qe.SessionVariable;
+import org.apache.doris.thrift.TStorageType;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+import java.util.HashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+class MTMVOutputPredicatesTest {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testDefinitionChangeInvalidatesPublishedGuarantees(boolean
baseViewChange) {
+ CacheMTMV mtmv = new CacheMTMV();
+ MTMVCache beforeChange = mtmv.getOrGenerateCache(mtmv.context);
+ Assertions.assertEquals(ImmutableSet.of(mtmv.predicate),
beforeChange.getOutputPredicates());
+
+ if (baseViewChange) {
+ mtmv.processBaseViewChange("base view definition changed");
+ } else {
+ mtmv.alterStatus(new MTMVStatus(MTMVState.SCHEMA_CHANGE, "base
table changed"));
+ }
+
+ MTMVCache afterChange = mtmv.getOrGenerateCache(mtmv.context);
+ Assertions.assertNotSame(beforeChange, afterChange);
+ Assertions.assertTrue(afterChange.getOutputPredicates().isEmpty());
+
Assertions.assertTrue(mtmv.getRefreshSnapshot().getPartitionSnapshots().isEmpty());
+ // Return the normalized cache that was published, not the builder's
unqualified guarantees.
+ Assertions.assertSame(afterChange,
mtmv.getOrGenerateCache(mtmv.context));
+ Assertions.assertSame(beforeChange.getOriginalFinalPlan(),
afterChange.getOriginalFinalPlan());
+ }
+
+ @Test
+ void testViewChangeDiscardsInFlightGuarantees() throws Exception {
+ BlockingCacheMTMV mtmv = new BlockingCacheMTMV();
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+ Future<MTMVCache> cacheFuture = executor.submit(() ->
mtmv.getOrGenerateCache(mtmv.context));
+ Assertions.assertTrue(mtmv.firstBuildStarted.await(5,
TimeUnit.SECONDS));
+ mtmv.processBaseViewChange("base view changed while its cache was
building");
+ mtmv.releaseFirstBuild.countDown();
+
+ MTMVCache cache = cacheFuture.get(5, TimeUnit.SECONDS);
+ Assertions.assertEquals(2, mtmv.buildCount.get());
+ Assertions.assertTrue(cache.getOutputPredicates().isEmpty());
+ Assertions.assertSame(cache,
mtmv.getOrGenerateCache(mtmv.context));
+ } finally {
+ mtmv.releaseFirstBuild.countDown();
+ executor.shutdownNow();
+ Assertions.assertTrue(executor.awaitTermination(5,
TimeUnit.SECONDS));
+ }
+ }
+
+ @Test
+ void testGuaranteesRecoverOnlyAfterEveryPartitionUsesCurrentDefinition() {
+ CacheMTMV mtmv = new CacheMTMV();
+ mtmv.processBaseViewChange("base view predicate changed");
+ MTMVCache changedCache = mtmv.getOrGenerateCache(mtmv.context);
+ Assertions.assertTrue(changedCache.getOutputPredicates().isEmpty());
+
+ Env env = Mockito.mock(Env.class);
+ MTMVService service = Mockito.mock(MTMVService.class);
+ Mockito.when(env.getMtmvService()).thenReturn(service);
+
+ try (MockedStatic<Env> mockedEnv = Mockito.mockStatic(Env.class)) {
+ mockedEnv.when(Env::getCurrentEnv).thenReturn(env);
+ // Replay restores NORMAL after refreshing p1, but p2 still
contains the old definition's rows.
+ Assertions.assertTrue(mtmv.addTaskResult(successfulRefresh(mtmv),
mtmv.getRelation(),
+ ImmutableMap.of("p1", new MTMVRefreshPartitionSnapshot()),
true));
+ Assertions.assertEquals(MTMVState.NORMAL,
mtmv.getStatus().getState());
+ Assertions.assertEquals(ImmutableSet.of("p1"),
mtmv.getRefreshSnapshot().getPartitionSnapshots().keySet());
+ MTMVCache partiallyRefreshedCache =
mtmv.getOrGenerateCache(mtmv.context);
+ Assertions.assertNotSame(changedCache, partiallyRefreshedCache);
+
Assertions.assertTrue(partiallyRefreshedCache.getOutputPredicates().isEmpty());
+
+ // A live completion builds both caches before adding p2's
snapshot. Publication must use
+ // the merged snapshot coverage and replace the previous cache
without rebuilding on read.
+ Assertions.assertTrue(mtmv.addTaskResult(successfulRefresh(mtmv),
mtmv.getRelation(),
+ ImmutableMap.of("p2", new MTMVRefreshPartitionSnapshot()),
false));
+ int buildsAtRefreshCompletion = mtmv.buildCount.get();
+ MTMVCache fullyRefreshedCache =
mtmv.getOrGenerateCache(mtmv.context);
+ Assertions.assertNotSame(partiallyRefreshedCache,
fullyRefreshedCache);
+ Assertions.assertEquals(ImmutableSet.of("p1", "p2"),
+
mtmv.getRefreshSnapshot().getPartitionSnapshots().keySet());
+ Assertions.assertEquals(ImmutableSet.of(mtmv.predicate),
fullyRefreshedCache.getOutputPredicates());
+ Assertions.assertEquals(buildsAtRefreshCompletion,
mtmv.buildCount.get());
+ }
+ }
+
+ private static MTMVTask successfulRefresh(CacheMTMV mtmv) {
+ MTMVTask task = new MTMVTask(mtmv, mtmv.getRelation(),
+ new MTMVTaskContext(MTMVTaskTriggerMode.MANUAL));
+ task.setTaskId(1L);
+ task.setStatus(TaskStatus.SUCCESS);
+ Deencapsulation.setField(task, "mtmvSchemaChangeVersion",
mtmv.getSchemaChangeVersion());
+ return task;
+ }
+
+ private static class CacheMTMV extends MTMV {
+ final ConnectContext context = Mockito.mock(ConnectContext.class);
+ final AtomicInteger buildCount = new AtomicInteger();
+ private final Plan plan = new LogicalOneRowRelation(new RelationId(1),
+ ImmutableList.of(new Alias(new ExprId(1), new
IntegerLiteral(20), "id")));
+ private final Expression predicate = new
GreaterThan(plan.getOutput().get(0), new IntegerLiteral(10));
+
+ CacheMTMV() {
+ Mockito.when(context.getSessionVariable()).thenReturn(new
SessionVariable());
+ setId(1L);
+ Deencapsulation.setField(this, "name", "mv");
+ setQualifiedDbName("db");
+ setQuerySql("select id from base_view");
+ MTMVStatus status = new MTMVStatus(MTMVState.NORMAL, null);
+ status.setRefreshState(MTMVRefreshState.SUCCESS);
+ setStatus(status);
+ setRefreshInfo(new MTMVRefreshInfo(BuildMode.IMMEDIATE,
RefreshMethod.COMPLETE,
+ new MTMVRefreshTriggerInfo(RefreshTrigger.MANUAL)));
+ setJobInfo(new MTMVJobInfo("mv_job"));
+ setMvProperties(new HashMap<>());
+ setRelation(new MTMVRelation(ImmutableSet.of(), ImmutableSet.of(),
ImmutableSet.of(),
+ ImmutableSet.of(), ImmutableSet.of()));
+ setMvPartitionInfo(new MTMVPartitionInfo());
+ Deencapsulation.setField(this, "sessionVariables", new HashMap<>(
+
context.getSessionVariable().getAffectQueryResultInPlanVariables()));
+ setBaseIndexId(1L);
+ setIndexMeta(1L, "mv", ImmutableList.of(new Column("id",
PrimitiveType.INT, true)),
+ 0, 0, (short) 1, TStorageType.COLUMN, KeysType.DUP_KEYS);
+ setPartitionInfo(new SinglePartitionInfo());
+ addPartition(new Partition(1L, "p1", new MaterializedIndex(),
null));
+ addPartition(new Partition(2L, "p2", new MaterializedIndex(),
null));
+ MTMVRefreshSnapshot snapshot = new MTMVRefreshSnapshot();
+ snapshot.updateSnapshots(ImmutableMap.of("p1", new
MTMVRefreshPartitionSnapshot(),
+ "p2", new MTMVRefreshPartitionSnapshot()),
getPartitionNames());
+ setRefreshSnapshot(snapshot);
+ }
+
+ @Override
+ protected MTMVCache createRewriteCache(ConnectContext currentContext,
boolean needLock,
+ boolean addSessionVarGuard) {
+ buildCount.incrementAndGet();
+ return new MTMVCache(Pair.of(plan, null), plan, null,
ImmutableList.of(), ImmutableSet.of(predicate));
+ }
+ }
+
+ private static class BlockingCacheMTMV extends CacheMTMV {
+ private final CountDownLatch firstBuildStarted = new CountDownLatch(1);
+ private final CountDownLatch releaseFirstBuild = new CountDownLatch(1);
+
+ @Override
+ protected MTMVCache createRewriteCache(ConnectContext currentContext,
boolean needLock,
+ boolean addSessionVarGuard) {
+ MTMVCache cache = super.createRewriteCache(currentContext,
needLock, addSessionVarGuard);
+ if (buildCount.get() == 1) {
+ firstBuildStarted.countDown();
+ try {
+ Assertions.assertTrue(releaseFirstBuild.await(5,
TimeUnit.SECONDS));
+ } catch (InterruptedException exception) {
+ Thread.currentThread().interrupt();
+ throw new AssertionError(exception);
+ }
+ }
+ return cache;
+ }
+ }
+}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/mv/MTMVCacheTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/mv/MTMVCacheTest.java
index 5cb57665d59..2691c57daf1 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/nereids/mv/MTMVCacheTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/mv/MTMVCacheTest.java
@@ -19,29 +19,49 @@ package org.apache.doris.nereids.mv;
import org.apache.doris.catalog.MTMV;
import org.apache.doris.mtmv.MTMVCache;
+import org.apache.doris.mtmv.MTMVPlanUtil;
+import org.apache.doris.mtmv.MTMVRefreshEnum.MTMVState;
+import org.apache.doris.mtmv.MTMVRefreshSnapshot;
import org.apache.doris.mtmv.MTMVRelationManager;
+import org.apache.doris.mtmv.MTMVStatus;
import org.apache.doris.nereids.CascadesContext;
import
org.apache.doris.nereids.rules.exploration.mv.AsyncMaterializationContext;
import org.apache.doris.nereids.rules.exploration.mv.MaterializationContext;
import org.apache.doris.nereids.sqltest.SqlTestBase;
+import org.apache.doris.nereids.trees.expressions.GreaterThan;
import org.apache.doris.nereids.trees.expressions.SessionVarGuardExpr;
+import org.apache.doris.nereids.trees.expressions.literal.Literal;
import org.apache.doris.nereids.trees.plans.Plan;
import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
+import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
import org.apache.doris.nereids.util.PlanChecker;
import org.apache.doris.qe.ConnectContext;
import org.apache.doris.qe.SessionVariable;
import org.apache.doris.qe.SqlModeHelper;
+import com.google.common.collect.ImmutableList;
import mockit.Mock;
import mockit.MockUp;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.mockito.Mockito;
+import java.time.LocalDate;
+import java.time.ZoneOffset;
import java.util.BitSet;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
/**
* Relevant test case about mtmv cache.
@@ -129,4 +149,131 @@ public class MTMVCacheTest extends SqlTestBase {
.anyMatch(expr ->
expr.containsType(SessionVarGuardExpr.class)));
dropMvByNereids("drop materialized view mv1");
}
+
+ @Test
+ void testInvalidateShouldNotPublishInFlightRewriteCache() throws Exception
{
+ ControlledCacheMTMV mtmv = new ControlledCacheMTMV();
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+ Future<MTMVCache> cacheFuture = executor.submit(() ->
mtmv.getOrGenerateCache(connectContext));
+ Assertions.assertTrue(mtmv.firstBuildStarted.await(5,
TimeUnit.SECONDS));
+
+ mtmv.invalidateRewriteCache();
+ mtmv.releaseFirstBuild.countDown();
+
+ MTMVCache generatedCache = cacheFuture.get(5, TimeUnit.SECONDS);
+ Assertions.assertNotSame(mtmv.firstCache, generatedCache);
+ Assertions.assertSame(mtmv.secondCache, generatedCache);
+ Assertions.assertSame(generatedCache,
mtmv.getOrGenerateCache(connectContext));
+ } finally {
+ executor.shutdownNow();
+ }
+ }
+
+ @Test
+ void testNondeterministicDefinitionDoesNotExportFoldedPredicates() throws
Exception {
+ createFunction("create alias function mv_cache_day(int) with
parameter(n) "
+ + "as dayofmonth(days_add(current_date(), n))");
+ try {
+ for (String expression : ImmutableList.of(
+ "dayofmonth(current_date())", "dayofmonth(CURRENT_DATE)",
"mv_cache_day(0)")) {
+ for (LocalDate date : ImmutableList.of(LocalDate.of(2026, 9,
9), LocalDate.of(2026, 9, 10))) {
+ ConnectContext cacheContext =
Mockito.spy(MTMVPlanUtil.createBasicMvContext(connectContext,
+
MTMVPlanUtil.DISABLE_RULES_WHEN_GENERATE_MTMV_CACHE,
+
connectContext.getSessionVariable().getAffectQueryResultInPlanVariables()));
+ cacheContext.getSessionVariable().setTimeZone("UTC");
+
Mockito.doReturn(date.atStartOfDay(ZoneOffset.UTC).toInstant())
+ .when(cacheContext).getStartTimeInstant();
+ cacheContext.setThreadLocalInfo();
+ MTMVCache cache = MTMVCache.from("select id, score from T1
where score > " + expression,
+ cacheContext, false, true, connectContext, false);
+ LogicalFilter<? extends Plan> filter =
cache.getOriginalFinalPlan()
+ .<LogicalFilter<? extends
Plan>>collectFirst(LogicalFilter.class::isInstance)
+ .orElseThrow(AssertionError::new);
+ // The same definition folds to a different bound when its
cache is rebuilt tomorrow.
+
Assertions.assertTrue(filter.getConjuncts().stream().anyMatch(predicate ->
+ predicate instanceof GreaterThan &&
predicate.child(1) instanceof Literal
+ && ((Literal)
predicate.child(1)).getStringValue()
+
.equals(Integer.toString(date.getDayOfMonth()))),
+ cache.getOriginalFinalPlan()::treeString);
+
Assertions.assertTrue(cache.getOutputPredicates().isEmpty(),
cache.getOutputPredicates()::toString);
+ }
+ }
+ } finally {
+ connectContext.setThreadLocalInfo();
+ executeNereidsSql("drop function mv_cache_day(int)");
+ }
+ }
+
+ @ParameterizedTest(name = "throughView={0}")
+ @ValueSource(booleans = {false, true})
+ void testReplacedAliasDefinitionDoesNotExportFoldedPredicates(boolean
throughView) throws Exception {
+ createFunction("create alias function mv_cache_bound(int) with
parameter(n) as abs(n)");
+ try {
+ String definitionSql = "select id from T1 where id >
mv_cache_bound(0)";
+ if (throughView) {
+ // View analysis must retain the alias dependency in its
parent's statement.
+ createView("create view mv_cache_alias_view as " +
definitionSql);
+ definitionSql = "select id from mv_cache_alias_view";
+ }
+ for (int bound : ImmutableList.of(0, 10)) {
+ if (bound == 10) {
+ executeNereidsSql("drop function mv_cache_bound(int)");
+ createFunction("create alias function mv_cache_bound(int)
with parameter(n) as abs(n + 10)");
+ }
+ ConnectContext cacheContext =
MTMVPlanUtil.createBasicMvContext(connectContext,
+ MTMVPlanUtil.DISABLE_RULES_WHEN_GENERATE_MTMV_CACHE,
+
connectContext.getSessionVariable().getAffectQueryResultInPlanVariables());
+ cacheContext.setThreadLocalInfo();
+ MTMVCache cache = MTMVCache.from(definitionSql, cacheContext,
false, true, connectContext, false);
+ LogicalFilter<? extends Plan> filter =
cache.getOriginalFinalPlan()
+ .<LogicalFilter<? extends
Plan>>collectFirst(LogicalFilter.class::isInstance)
+ .orElseThrow(AssertionError::new);
+ // A rebuild binds the new deterministic body, although stored
MV rows have not been refreshed.
+
Assertions.assertTrue(filter.getConjuncts().stream().anyMatch(predicate ->
+ predicate instanceof GreaterThan &&
predicate.child(1) instanceof Literal
+ && ((Literal)
predicate.child(1)).getStringValue()
+
.equals(Integer.toString(bound))),
+ cache.getOriginalFinalPlan()::treeString);
+ Assertions.assertTrue(cache.getOutputPredicates().isEmpty(),
cache.getOutputPredicates()::toString);
+ }
+ } finally {
+ connectContext.setThreadLocalInfo();
+ if (throughView) {
+ dropView("drop view if exists mv_cache_alias_view");
+ }
+ executeNereidsSql("drop function if exists mv_cache_bound(int)");
+ }
+ }
+
+ private static class ControlledCacheMTMV extends MTMV {
+ private final CountDownLatch firstBuildStarted = new CountDownLatch(1);
+ private final CountDownLatch releaseFirstBuild = new CountDownLatch(1);
+ private final AtomicInteger buildCount = new AtomicInteger();
+ private final MTMVCache firstCache = new MTMVCache(
+ null, null, null, Collections.emptyList(),
Collections.emptySet());
+ private final MTMVCache secondCache = new MTMVCache(
+ null, null, null, Collections.emptyList(),
Collections.emptySet());
+
+ private ControlledCacheMTMV() {
+ setStatus(new MTMVStatus(MTMVState.NORMAL, null));
+ setRefreshSnapshot(new MTMVRefreshSnapshot());
+ }
+
+ @Override
+ protected MTMVCache createRewriteCache(ConnectContext currentContext,
boolean needLock,
+ boolean addSessionVarGuard) {
+ if (buildCount.incrementAndGet() == 1) {
+ firstBuildStarted.countDown();
+ try {
+ Assertions.assertTrue(releaseFirstBuild.await(5,
TimeUnit.SECONDS));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ }
+ return firstCache;
+ }
+ return secondCache;
+ }
+ }
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/mv/RelationImpliedPredicateTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/mv/RelationImpliedPredicateTest.java
new file mode 100644
index 00000000000..378fed01ec2
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/mv/RelationImpliedPredicateTest.java
@@ -0,0 +1,627 @@
+// 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.nereids.mv;
+
+import org.apache.doris.catalog.Database;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.MTMV;
+import org.apache.doris.catalog.Partition;
+import org.apache.doris.common.jmockit.Deencapsulation;
+import org.apache.doris.mtmv.BaseTableInfo;
+import org.apache.doris.mtmv.MTMVCache;
+import org.apache.doris.mtmv.MTMVRefreshEnum.MTMVRefreshState;
+import org.apache.doris.mtmv.MTMVRefreshEnum.MTMVState;
+import org.apache.doris.mtmv.MTMVRefreshPartitionSnapshot;
+import org.apache.doris.mtmv.MTMVRelationManager;
+import org.apache.doris.nereids.CascadesContext;
+import org.apache.doris.nereids.rules.exploration.mv.MaterializationContext;
+import
org.apache.doris.nereids.rules.exploration.mv.MaterializedViewPredicateCollector;
+import
org.apache.doris.nereids.rules.exploration.mv.PreMaterializedViewRewriter.PreRewriteStrategy;
+import org.apache.doris.nereids.rules.exploration.mv.StructInfo;
+import org.apache.doris.nereids.sqltest.SqlTestBase;
+import org.apache.doris.nereids.trees.copier.DeepCopierContext;
+import org.apache.doris.nereids.trees.copier.LogicalPlanDeepCopier;
+import org.apache.doris.nereids.trees.expressions.EqualTo;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.GreaterThan;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.literal.Literal;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
+import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalFilter;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalOlapScan;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalPlan;
+import org.apache.doris.nereids.util.ExpressionUtils;
+import org.apache.doris.nereids.util.PlanChecker;
+import org.apache.doris.qe.ConnectContext;
+
+import com.google.common.collect.ImmutableList;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * Tests derivation and use of predicates guaranteed by materialized view
outputs.
+ */
+public class RelationImpliedPredicateTest extends SqlTestBase {
+
+ @Override
+ protected void runBeforeAll() throws Exception {
+ super.runBeforeAll();
+ MTMVRelationManager existingManager =
Env.getCurrentEnv().getMtmvService().getRelationManager();
+ MTMVRelationManager spyManager = Mockito.spy(existingManager);
+ Mockito.doAnswer(invocation -> {
+ MTMV mtmv = invocation.getArgument(0);
+ ConnectContext ctx = invocation.getArgument(1);
+ ctx.getStatementContext().getMvCanRewritePartitionsMap()
+ .putIfAbsent(new BaseTableInfo(mtmv),
mtmv.getPartitions());
+ return true;
+ }).when(spyManager).isMVPartitionValid(Mockito.any(MTMV.class),
+ Mockito.any(ConnectContext.class),
+ Mockito.anyBoolean(), Mockito.any(Map.class));
+ Deencapsulation.setField(Env.getCurrentEnv().getMtmvService(),
"relationManager", spyManager);
+
Env.getCurrentEnv().getMtmvService().registerHook("MTMVRelationManager",
spyManager);
+ }
+
+ @Test
+ void testOutputPredicatesBindToGeneratedDirectAndCopiedScans() throws
Exception {
+ initMvRewriteSession();
+ dropMvByNereids("drop materialized view if exists
mv_relation_implied_fact");
+ try {
+ createMvByNereids("create materialized view
mv_relation_implied_fact "
+ + "BUILD IMMEDIATE REFRESH COMPLETE ON MANUAL\n"
+ + "DISTRIBUTED BY RANDOM BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1') \n"
+ + "as select T1.id, T1.score, T1.score as score_copy from
T1 where T1.score = 1;");
+ mockRefreshedMtmv("mv_relation_implied_fact");
+
+ CascadesContext context = createCascadesContext(
+ "select T1.id, T1.score, T1.score as score_copy from T1
where T1.score = 1",
+ connectContext);
+ PlanChecker.from(context)
+ .setIsQuery()
+ .analyze()
+ .rewrite();
+
+ Assertions.assertEquals(1,
context.getMaterializationContexts().size());
+ MaterializationContext materializationContext =
context.getMaterializationContexts().get(0);
+ Plan scanPlan = materializationContext.getScanPlan(null, context);
+ Assertions.assertNotNull(scanPlan);
+ LogicalOlapScan generatedScan =
scanPlan.<LogicalOlapScan>collectFirst(LogicalOlapScan.class::isInstance)
+ .orElseThrow(() -> new
AssertionError(scanPlan.treeString()));
+
+ CascadesContext directContext = createCascadesContext("select *
from mv_relation_implied_fact");
+ PlanChecker.from(directContext).setIsQuery().analyze();
+ LogicalOlapScan directScan = directContext.getRewritePlan()
+
.<LogicalOlapScan>collectFirst(LogicalOlapScan.class::isInstance).orElseThrow(AssertionError::new);
+ LogicalOlapScan copiedScan = (LogicalOlapScan) directScan.accept(
+ LogicalPlanDeepCopier.INSTANCE, new DeepCopierContext());
+
Assertions.assertTrue(directScan.getOutputSet().stream().noneMatch(copiedScan.getOutputSet()::contains));
+
+ // Duplicate definition expressions need one output
representative, not a product of slot mappings.
+ for (LogicalOlapScan scan : ImmutableList.of(generatedScan,
directScan, copiedScan)) {
+ Set<Expression> predicates = new
MaterializedViewPredicateCollector(directContext).collect(scan);
+ Assertions.assertTrue(predicates.stream()
+ .allMatch(predicate ->
scan.getOutputSet().containsAll(predicate.getInputSlots())),
+ predicates::toString);
+ Slot score = scan.getOutput().stream().filter(slot ->
slot.getName().equals("score"))
+ .findFirst().orElseThrow(AssertionError::new);
+ Assertions.assertTrue(predicates.stream().anyMatch(predicate
-> predicate instanceof EqualTo
+ && predicate.child(0).equals(score) &&
predicate.child(1) instanceof Literal
+ && ((Literal)
predicate.child(1)).getStringValue().equals("1")),
+ predicates::toString);
+ }
+ } finally {
+ dropMvByNereids("drop materialized view if exists
mv_relation_implied_fact");
+ }
+ }
+
+ @Test
+ void testNestedMvRewriteByOutputPredicate() throws Exception {
+ initMvRewriteSession();
+ dropMvByNereids("drop materialized view if exists
mv_relation_implied_explicit_join");
+ dropMvByNereids("drop materialized view if exists
mv_relation_implied_fact_with_score");
+ try {
+ createMvByNereids("create materialized view
mv_relation_implied_fact_with_score "
+ + "BUILD IMMEDIATE REFRESH COMPLETE ON MANUAL\n"
+ + "DISTRIBUTED BY RANDOM BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1') \n"
+ + "as select T1.id, T1.score from T1 where T1.score = 1;");
+ mockRefreshedMtmv("mv_relation_implied_fact_with_score");
+
+ createMvByNereids("create materialized view
mv_relation_implied_explicit_join "
+ + "BUILD IMMEDIATE REFRESH COMPLETE ON MANUAL\n"
+ + "DISTRIBUTED BY RANDOM BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1') \n"
+ + "as select f.id, f.score as fact_score, T2.score as
dim_score "
+ + "from mv_relation_implied_fact_with_score f "
+ + "inner join T2 on f.id = T2.id "
+ + "where f.score = 1;");
+ mockRefreshedMtmv("mv_relation_implied_explicit_join");
+
+ CascadesContext context = createCascadesContext(
+ "select T1.id, T1.score as fact_score, T2.score as
dim_score "
+ + "from T1 inner join T2 on T1.id = T2.id "
+ + "where T1.score = 1",
+ connectContext);
+ PhysicalPlan bestPlan = PlanChecker.from(context)
+ .setIsQuery()
+ .analyze()
+ .rewrite()
+ .preMvRewrite()
+ .optimize()
+ .getBestPlanTree();
+ String plan = bestPlan.treeString();
+
Assertions.assertTrue(plan.contains("mv_relation_implied_explicit_join"),
+ () -> plan +
MaterializationContext.toSummaryString(context, bestPlan));
+ } finally {
+ dropMvByNereids("drop materialized view if exists
mv_relation_implied_explicit_join");
+ dropMvByNereids("drop materialized view if exists
mv_relation_implied_fact_with_score");
+ }
+ }
+
+ @Test
+ void testNestedMvRewriteAfterPredicateColumnIsProjectedOut() throws
Exception {
+ initMvRewriteSession();
+ dropMvByNereids("drop materialized view if exists
mv_relation_implied_projected_upper");
+ dropMvByNereids("drop materialized view if exists
mv_relation_implied_projected_lower");
+ try {
+ createMvByNereids("create materialized view
mv_relation_implied_projected_lower "
+ + "BUILD IMMEDIATE REFRESH COMPLETE ON MANUAL\n"
+ + "DISTRIBUTED BY RANDOM BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1')\n"
+ + "as select id, score from T1 where score = 1;");
+ mockRefreshedMtmv("mv_relation_implied_projected_lower");
+ createMvByNereids("create materialized view
mv_relation_implied_projected_upper "
+ + "BUILD IMMEDIATE REFRESH COMPLETE ON MANUAL\n"
+ + "DISTRIBUTED BY RANDOM BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1')\n"
+ + "as select f.id, T2.score as dim_score "
+ + "from mv_relation_implied_projected_lower f inner join
T2 on f.id = T2.id "
+ + "where f.score > 0;");
+ MTMV upper =
mockRefreshedMtmv("mv_relation_implied_projected_upper");
+ StructInfo cachedDefinition =
upper.getOrGenerateCache(connectContext)
+ .getAllRulesRewrittenPlanAndStructInfo().value();
+
Assertions.assertTrue(cachedDefinition.getTopPlan().anyMatch(LogicalFilter.class::isInstance));
+
+ CascadesContext context = createCascadesContext(
+ "select T1.id, T2.score as dim_score from T1 inner join T2
on T1.id = T2.id "
+ + "where T1.score = 1");
+ PhysicalPlan bestPlan =
PlanChecker.from(context).setIsQuery().analyze().rewrite()
+ .preMvRewrite().optimize().getBestPlanTree();
+ Assertions.assertTrue(bestPlan.anyMatch(plan -> plan instanceof
PhysicalOlapScan
+ && ((PhysicalOlapScan) plan).getTable().getName()
+
.equals("mv_relation_implied_projected_upper")),
+ () -> bestPlan.treeString() +
MaterializationContext.toSummaryString(context, bestPlan));
+ // The successful candidate uses a query-local proof; the stored
definition keeps its filter.
+
Assertions.assertTrue(cachedDefinition.getTopPlan().anyMatch(LogicalFilter.class::isInstance));
+ List<PhysicalFilter<? extends Plan>> filters =
bestPlan.collectToList(PhysicalFilter.class::isInstance);
+ Assertions.assertTrue(filters.stream().allMatch(filter ->
filter.getConjuncts().stream()
+ .allMatch(predicate ->
filter.child().getOutputSet()
+ .containsAll(predicate.getInputSlots()))),
+ bestPlan::treeString);
+ } finally {
+ dropMvByNereids("drop materialized view if exists
mv_relation_implied_projected_upper");
+ dropMvByNereids("drop materialized view if exists
mv_relation_implied_projected_lower");
+ }
+ }
+
+ @Test
+ void testStructInfoRetainsFilterAboveNullExtendedMvScan() throws Exception
{
+ initMvRewriteSession();
+ dropMvByNereids("drop materialized view if exists
mv_relation_implied_nullable_input");
+ try {
+ createMvByNereids("create materialized view
mv_relation_implied_nullable_input "
+ + "BUILD IMMEDIATE REFRESH COMPLETE ON MANUAL\n"
+ + "DISTRIBUTED BY RANDOM BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1')\n"
+ + "as select id, score from T1 where score = 1;");
+ mockRefreshedMtmv("mv_relation_implied_nullable_input");
+ CascadesContext context = createCascadesContext(
+ "select f.id from T2 left join
mv_relation_implied_nullable_input f on T2.id = f.id "
+ + "where f.score = 1");
+ // Analyze only: rewriting would turn this null-rejecting LEFT
JOIN into an INNER JOIN first.
+ PlanChecker.from(context).setIsQuery().analyze();
+ Plan analyzedPlan = context.getRewritePlan();
+ LogicalFilter<? extends Plan> filter = analyzedPlan
+ .<LogicalFilter<? extends
Plan>>collectFirst(LogicalFilter.class::isInstance)
+ .orElseThrow(AssertionError::new);
+ StructInfo structInfo = StructInfo.of(analyzedPlan,
context).withoutRedundantMvFilters(context);
+
Assertions.assertTrue(structInfo.getPredicates().getPulledUpPredicates()
+ .containsAll(filter.getConjuncts()),
structInfo.getTopPlan()::treeString);
+ } finally {
+ dropMvByNereids("drop materialized view if exists
mv_relation_implied_nullable_input");
+ }
+ }
+
+ @Test
+ void testNestedMvRewriteWithNonOutputDefinitionPredicate() throws
Exception {
+ initMvRewriteSession();
+ dropMvByNereids("drop materialized view if exists
mv_relation_implied_join");
+ dropMvByNereids("drop materialized view if exists
mv_relation_implied_fact");
+ try {
+ createMvByNereids("create materialized view
mv_relation_implied_fact "
+ + "BUILD IMMEDIATE REFRESH COMPLETE ON MANUAL\n"
+ + "DISTRIBUTED BY RANDOM BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1') \n"
+ + "as select T1.id from T1 where T1.score = 1;");
+ mockRefreshedMtmv("mv_relation_implied_fact");
+
+ createMvByNereids("create materialized view
mv_relation_implied_join "
+ + "BUILD IMMEDIATE REFRESH COMPLETE ON MANUAL\n"
+ + "DISTRIBUTED BY RANDOM BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1') \n"
+ + "as select f.id, T2.score as dim_score "
+ + "from mv_relation_implied_fact f "
+ + "inner join T2 on f.id = T2.id;");
+ mockRefreshedMtmv("mv_relation_implied_join");
+
+ // Both sides scan the same lower MV. Its identity guarantees
score = 1 even though score is not output.
+ CascadesContext context = createCascadesContext(
+ "select T1.id, T2.score as dim_score "
+ + "from T1 inner join T2 on T1.id = T2.id "
+ + "where T1.score = 1",
+ connectContext);
+ PhysicalPlan bestPlan = PlanChecker.from(context)
+ .setIsQuery()
+ .analyze()
+ .rewrite()
+ .preMvRewrite()
+ .optimize()
+ .getBestPlanTree();
+ String plan = bestPlan.treeString();
+ Assertions.assertTrue(plan.contains("mv_relation_implied_join"),
+ () -> plan +
MaterializationContext.toSummaryString(context, bestPlan));
+ } finally {
+ dropMvByNereids("drop materialized view if exists
mv_relation_implied_join");
+ dropMvByNereids("drop materialized view if exists
mv_relation_implied_fact");
+ }
+ }
+
+ @Test
+ void testNestedInnerJoinRewriteRetainsNullRejectFilter() throws Exception {
+ initMvRewriteSession();
+ dropMvByNereids("drop materialized view if exists
mv_relation_implied_left_join");
+ dropMvByNereids("drop materialized view if exists
mv_relation_implied_positive_id");
+ try {
+ createMvByNereids("create materialized view
mv_relation_implied_positive_id "
+ + "BUILD IMMEDIATE REFRESH COMPLETE ON MANUAL\n"
+ + "DISTRIBUTED BY RANDOM BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1')\n"
+ + "as select id from T1 where id > 0;");
+ mockRefreshedMtmv("mv_relation_implied_positive_id");
+ createMvByNereids("create materialized view
mv_relation_implied_left_join "
+ + "BUILD IMMEDIATE REFRESH COMPLETE ON MANUAL\n"
+ + "DISTRIBUTED BY RANDOM BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1')\n"
+ + "as select T2.id as dim_id, f.id as fact_id "
+ + "from T2 left join mv_relation_implied_positive_id f on
T2.id = f.id;");
+ mockRefreshedMtmv("mv_relation_implied_left_join");
+
+ CascadesContext context = createCascadesContext(
+ "select T2.id as dim_id, T1.id as fact_id "
+ + "from T2 inner join T1 on T2.id = T1.id where
T1.id > 0");
+ PhysicalPlan bestPlan =
PlanChecker.from(context).setIsQuery().analyze().rewrite()
+ .preMvRewrite().optimize().getBestPlanTree();
+ Optional<PhysicalOlapScan> mvScan = bestPlan.collectFirst(plan ->
plan instanceof PhysicalOlapScan
+ && ((PhysicalOlapScan)
plan).getTable().getName().equals("mv_relation_implied_left_join"));
+ Assertions.assertTrue(mvScan.isPresent(),
+ () -> bestPlan.treeString() +
MaterializationContext.toSummaryString(context, bestPlan));
+ Slot factId = mvScan.get().getOutput().stream().filter(slot ->
slot.getName().equals("fact_id"))
+ .findFirst().orElseThrow(AssertionError::new);
+ List<PhysicalFilter<? extends Plan>> filters =
bestPlan.collectToList(PhysicalFilter.class::isInstance);
+ Set<Expression> compensation = filters.stream().flatMap(filter ->
filter.getConjuncts().stream())
+ .collect(Collectors.toSet());
+ // A LEFT JOIN MV can contain (dim_id = 2, fact_id = NULL); its
scan must not be used unfiltered.
+
Assertions.assertTrue(ExpressionUtils.inferNotNullSlots(compensation,
context).contains(factId),
+ bestPlan::treeString);
+ } finally {
+ dropMvByNereids("drop materialized view if exists
mv_relation_implied_left_join");
+ dropMvByNereids("drop materialized view if exists
mv_relation_implied_positive_id");
+ }
+ }
+
+ @Test
+ void testNestedInnerJoinRewriteDoesNotRequireUnprojectedJoinKey() throws
Exception {
+ initMvRewriteSession();
+ dropMvByNereids("drop materialized view if exists
mv_relation_implied_unprojected_join_key");
+ dropMvByNereids("drop materialized view if exists
mv_relation_implied_join_key");
+ try {
+ createMvByNereids("create materialized view
mv_relation_implied_join_key "
+ + "BUILD IMMEDIATE REFRESH COMPLETE ON MANUAL\n"
+ + "DISTRIBUTED BY RANDOM BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1')\n"
+ + "as select id from T1 where id > 0;");
+ mockRefreshedMtmv("mv_relation_implied_join_key");
+ createMvByNereids("create materialized view
mv_relation_implied_unprojected_join_key "
+ + "BUILD IMMEDIATE REFRESH COMPLETE ON MANUAL\n"
+ + "DISTRIBUTED BY RANDOM BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1')\n"
+ + "as select f.id as fact_id "
+ + "from T2 left join mv_relation_implied_join_key f on
T2.id = f.id;");
+ mockRefreshedMtmv("mv_relation_implied_unprojected_join_key");
+
+ CascadesContext context = createCascadesContext(
+ "select T1.id as fact_id from T2 inner join T1 on T2.id =
T1.id where T1.id > 0");
+ List<Plan> alternatives =
PlanChecker.from(context).setIsQuery().analyze().rewrite()
+ .preMvRewrite().optimize().getAllPlan();
+ List<Plan> rewrittenPlans = alternatives.stream().filter(plan ->
plan.anyMatch(node ->
+ node instanceof LogicalOlapScan &&
((LogicalOlapScan) node).getTable().getName()
+
.equals("mv_relation_implied_unprojected_join_key")))
+ .collect(Collectors.toList());
+ Assertions.assertFalse(rewrittenPlans.isEmpty(),
+ () ->
alternatives.stream().map(Plan::treeString).collect(Collectors.joining("\n")));
+ // HyperGraph already matched T2.id = f.id. Requiring it as a
filter would need the absent T2.id output.
+ for (Plan rewrittenPlan : rewrittenPlans) {
+ LogicalOlapScan scan =
rewrittenPlan.<LogicalOlapScan>collectFirst(node ->
+ node instanceof LogicalOlapScan && ((LogicalOlapScan)
node).getTable().getName()
+
.equals("mv_relation_implied_unprojected_join_key")).orElseThrow(AssertionError::new);
+ Slot factId = scan.getOutput().stream().filter(slot ->
slot.getName().equals("fact_id"))
+ .findFirst().orElseThrow(AssertionError::new);
+ List<LogicalFilter<? extends Plan>> filters =
+
rewrittenPlan.collectToList(LogicalFilter.class::isInstance);
+ Set<Expression> compensation = filters.stream().flatMap(filter
-> filter.getConjuncts().stream())
+ .collect(Collectors.toSet());
+
Assertions.assertTrue(ExpressionUtils.inferNotNullSlots(compensation,
context).contains(factId),
+ rewrittenPlan::treeString);
+ }
+ } finally {
+ dropMvByNereids("drop materialized view if exists
mv_relation_implied_unprojected_join_key");
+ dropMvByNereids("drop materialized view if exists
mv_relation_implied_join_key");
+ }
+ }
+
+ @Test
+ void testDirectLowerMvQueryUsesUpperMv() throws Exception {
+ initMvRewriteSession();
+ dropMvByNereids("drop materialized view if exists
mv_relation_implied_direct_upper");
+ dropMvByNereids("drop materialized view if exists
mv_relation_implied_direct_lower");
+ try {
+ createMvByNereids("create materialized view
mv_relation_implied_direct_lower "
+ + "BUILD IMMEDIATE REFRESH COMPLETE ON MANUAL\n"
+ + "DISTRIBUTED BY RANDOM BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1')\n"
+ + "as select id from T1 where score = 1;");
+ mockRefreshedMtmv("mv_relation_implied_direct_lower");
+ createMvByNereids("create materialized view
mv_relation_implied_direct_upper "
+ + "BUILD IMMEDIATE REFRESH COMPLETE ON MANUAL\n"
+ + "DISTRIBUTED BY RANDOM BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1')\n"
+ + "as select id from mv_relation_implied_direct_lower;");
+ mockRefreshedMtmv("mv_relation_implied_direct_upper");
+
+ CascadesContext context = createCascadesContext("select id from
mv_relation_implied_direct_lower");
+ List<Plan> alternatives =
PlanChecker.from(context).setIsQuery().analyze().rewrite()
+ .preMvRewrite().optimize().getAllPlan();
+ // Both one-table plans can have the same estimated cost; assert
rewrite availability.
+ Assertions.assertTrue(alternatives.stream().anyMatch(plan ->
plan.anyMatch(node ->
+ node instanceof LogicalOlapScan &&
((LogicalOlapScan) node).getTable().getName()
+
.equals("mv_relation_implied_direct_upper"))),
+ () ->
alternatives.stream().map(Plan::treeString).collect(Collectors.joining("\n")));
+ } finally {
+ dropMvByNereids("drop materialized view if exists
mv_relation_implied_direct_upper");
+ dropMvByNereids("drop materialized view if exists
mv_relation_implied_direct_lower");
+ }
+ }
+
+ @Test
+ void testSelfJoinNonOutputPredicatesAreNotExported() throws Exception {
+ initMvRewriteSession();
+ dropMvByNereids("drop materialized view if exists
mv_relation_implied_self_join");
+ try {
+ createMvByNereids("create materialized view
mv_relation_implied_self_join "
+ + "BUILD IMMEDIATE REFRESH COMPLETE ON MANUAL\n"
+ + "DISTRIBUTED BY RANDOM BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1')\n"
+ + "as select a.id from T1 a cross join T1 b where a.score
= 1 and b.score = 2;");
+ mockRefreshedMtmv("mv_relation_implied_self_join");
+ CascadesContext context = createCascadesContext("select id from
mv_relation_implied_self_join");
+ PlanChecker.from(context).setIsQuery().analyze();
+ LogicalOlapScan scan = context.getRewritePlan()
+
.<LogicalOlapScan>collectFirst(LogicalOlapScan.class::isInstance).orElseThrow(AssertionError::new);
+ // The predicates constrain different rows of T1 and say nothing
about the MV's only output, id.
+ Set<Expression> predicates = ((MTMV)
scan.getTable()).getOrGenerateCache(connectContext)
+ .getOutputPredicates();
+ Assertions.assertTrue(predicates.isEmpty(), predicates::toString);
+ } finally {
+ dropMvByNereids("drop materialized view if exists
mv_relation_implied_self_join");
+ }
+ }
+
+ @Test
+ void testTimeDependentMvDefinitionDoesNotRemoveQueryFilter() throws
Exception {
+ initMvRewriteSession();
+ dropMvByNereids("drop materialized view if exists
mv_relation_implied_current_date");
+ try {
+ createMvByNereids("create materialized view
mv_relation_implied_current_date "
+ + "BUILD IMMEDIATE REFRESH COMPLETE ON MANUAL\n"
+ + "DISTRIBUTED BY RANDOM BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1',
'enable_nondeterministic_function' = 'true')\n"
+ + "as select id, score from T1 where score >
dayofmonth(current_date());");
+ mockRefreshedMtmv("mv_relation_implied_current_date");
+ CascadesContext context = createCascadesContext(
+ "select id from mv_relation_implied_current_date where
score > dayofmonth(current_date())");
+ PlanChecker.from(context).setIsQuery().analyze().rewrite();
+ Plan queryPlan = context.getRewritePlan();
+ LogicalFilter<? extends Plan> filter = queryPlan
+ .<LogicalFilter<? extends
Plan>>collectFirst(LogicalFilter.class::isInstance)
+ .orElseThrow(AssertionError::new);
+ StructInfo structInfo = StructInfo.of(queryPlan,
context).withoutRedundantMvFilters(context);
+ // Stored rows were selected at refresh time. Re-evaluating the
date while planning
+ // cannot prove that today's query filter is redundant on those
rows.
+
Assertions.assertTrue(structInfo.getPredicates().getPulledUpPredicates()
+ .containsAll(filter.getConjuncts()),
structInfo.getTopPlan()::treeString);
+ } finally {
+ dropMvByNereids("drop materialized view if exists
mv_relation_implied_current_date");
+ }
+ }
+
+ @Test
+ void testChangedBaseViewDoesNotSupplyOutputGuarantees() throws Exception {
+ initMvRewriteSession();
+ dropMvByNereids("drop materialized view if exists
mv_relation_implied_changed_view");
+ dropView("drop view if exists relation_implied_base_view");
+ try {
+ createView("create view relation_implied_base_view as select id
from T1 where id > 0");
+ createMvByNereids("create materialized view
mv_relation_implied_changed_view "
+ + "BUILD IMMEDIATE REFRESH COMPLETE ON MANUAL\n"
+ + "DISTRIBUTED BY RANDOM BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1')\n"
+ + "as select id from relation_implied_base_view;");
+ MTMV mtmv = mockRefreshedMtmv("mv_relation_implied_changed_view");
+ MTMVCache originalCache = mtmv.getOrGenerateCache(connectContext);
+
Assertions.assertFalse(originalCache.getOutputPredicates().isEmpty());
+
+ executeNereidsSql("alter view relation_implied_base_view as select
id from T1 where id > 10");
+ Assertions.assertEquals(MTMVState.SCHEMA_CHANGE,
mtmv.getStatus().getState());
+ MTMVCache changedCache = mtmv.getOrGenerateCache(connectContext);
+ Assertions.assertNotSame(originalCache, changedCache);
+
Assertions.assertTrue(changedCache.getOutputPredicates().isEmpty());
+
Assertions.assertFalse(originalCache.getOutputPredicates().isEmpty());
+ Assertions.assertNotNull(changedCache.getOriginalFinalPlan());
+
+ CascadesContext context = createCascadesContext(
+ "select id from mv_relation_implied_changed_view where id
> 10");
+ PlanChecker.from(context).setIsQuery().analyze().rewrite();
+ Plan queryPlan = context.getRewritePlan();
+ LogicalFilter<? extends Plan> filter = queryPlan
+ .<LogicalFilter<? extends
Plan>>collectFirst(LogicalFilter.class::isInstance)
+ .orElseThrow(AssertionError::new);
+ StructInfo structInfo = StructInfo.of(queryPlan,
context).withoutRedundantMvFilters(context);
+
Assertions.assertTrue(structInfo.getPredicates().getPulledUpPredicates()
+ .containsAll(filter.getConjuncts()),
structInfo.getTopPlan()::treeString);
+ } finally {
+ dropMvByNereids("drop materialized view if exists
mv_relation_implied_changed_view");
+ dropView("drop view if exists relation_implied_base_view");
+ }
+ }
+
+ @Test
+ void testRefreshedLowerMvDoesNotRemoveCompensationForStaleUpper() throws
Exception {
+ initMvRewriteSession();
+ dropMvByNereids("drop materialized view if exists
mv_relation_implied_stored_upper");
+ dropMvByNereids("drop materialized view if exists
mv_relation_implied_stored_lower");
+ dropView("drop view if exists relation_implied_refresh_view");
+ try {
+ createView("create view relation_implied_refresh_view as select id
from T1 where id > 0");
+ createMvByNereids("create materialized view
mv_relation_implied_stored_lower "
+ + "BUILD IMMEDIATE REFRESH COMPLETE ON MANUAL\n"
+ + "DISTRIBUTED BY RANDOM BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1')\n"
+ + "as select id from relation_implied_refresh_view;");
+ MTMV lower = mockRefreshedMtmv("mv_relation_implied_stored_lower");
+ createMvByNereids("create materialized view
mv_relation_implied_stored_upper "
+ + "BUILD IMMEDIATE REFRESH COMPLETE ON MANUAL\n"
+ + "DISTRIBUTED BY RANDOM BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1', 'grace_period' =
'31536000')\n"
+ + "as select id from mv_relation_implied_stored_lower;");
+ MTMV upper = mockRefreshedMtmv("mv_relation_implied_stored_upper");
+
+ executeNereidsSql("alter view relation_implied_refresh_view as
select id from T1 where id > 10");
+ Assertions.assertEquals(MTMVState.SCHEMA_CHANGE,
lower.getStatus().getState());
+ // Only the lower MV is refreshed. The upper still contains rows
from before the view changed.
+ lower.getStatus().setState(MTMVState.NORMAL);
+ mockRefreshedPartitions(lower);
+
Assertions.assertFalse(lower.getOrGenerateCache(connectContext).getOutputPredicates().isEmpty());
+ Assertions.assertEquals(MTMVState.NORMAL,
upper.getStatus().getState());
+ upper.invalidateRewriteCache();
+
Assertions.assertTrue(upper.getOrGenerateCache(connectContext).getOutputPredicates().isEmpty());
+
+ CascadesContext context = createCascadesContext(
+ "select id from mv_relation_implied_stored_upper where id
> 10");
+ PlanChecker.from(context).setIsQuery().analyze().rewrite();
+ LogicalFilter<? extends Plan> filter = context.getRewritePlan()
+ .<LogicalFilter<? extends
Plan>>collectFirst(LogicalFilter.class::isInstance)
+ .orElseThrow(AssertionError::new);
+ StructInfo structInfo = StructInfo.of(context.getRewritePlan(),
context)
+ .withoutRedundantMvFilters(context);
+
Assertions.assertTrue(structInfo.getPredicates().getPulledUpPredicates()
+ .containsAll(filter.getConjuncts()),
structInfo.getTopPlan()::treeString);
+
+ // Let the real grace-period check admit the old upper MV, instead
of the fixture's
+ // unconditional partition-valid mock. Its old rows must still
satisfy the query filter.
+ MTMVRelationManager relationManager =
Env.getCurrentEnv().getMtmvService().getRelationManager();
+
Mockito.doCallRealMethod().when(relationManager).isMVPartitionValid(Mockito.eq(upper),
+ Mockito.any(ConnectContext.class), Mockito.anyBoolean(),
Mockito.anyMap());
+ CascadesContext lowerQueryContext = createCascadesContext(
+ "select id from mv_relation_implied_stored_lower where id
> 10");
+ List<Plan> alternatives =
PlanChecker.from(lowerQueryContext).setIsQuery().analyze().rewrite()
+ .preMvRewrite().optimize().getAllPlan();
+ List<Plan> upperPlans = alternatives.stream().filter(plan ->
plan.anyMatch(node ->
+ node instanceof LogicalOlapScan &&
((LogicalOlapScan) node).getTable().getName()
+
.equals("mv_relation_implied_stored_upper")))
+ .collect(Collectors.toList());
+ Assertions.assertFalse(upperPlans.isEmpty(),
+ () ->
alternatives.stream().map(Plan::treeString).collect(Collectors.joining("\n")));
+ for (Plan upperPlan : upperPlans) {
+ List<LogicalFilter<? extends Plan>> filters =
upperPlan.collectToList(LogicalFilter.class::isInstance);
+
Assertions.assertTrue(filters.stream().anyMatch(candidateFilter ->
candidateFilter.getConjuncts().stream()
+ .anyMatch(predicate -> predicate instanceof
GreaterThan
+ &&
candidateFilter.child().getOutputSet()
+
.containsAll(predicate.getInputSlots())
+ && predicate.child(1) instanceof
Literal
+ && ((Literal)
predicate.child(1)).getStringValue().equals("10"))),
+ upperPlan::treeString);
+ }
+ } finally {
+ dropMvByNereids("drop materialized view if exists
mv_relation_implied_stored_upper");
+ dropMvByNereids("drop materialized view if exists
mv_relation_implied_stored_lower");
+ dropView("drop view if exists relation_implied_refresh_view");
+ }
+ }
+
+ private MTMV mockRefreshedMtmv(String name) {
+ Database db = (Database)
Env.getCurrentEnv().getInternalCatalog().getDbNullable(connectContext.getDatabase());
+ MTMV mtmv = (MTMV) db.getTableNullable(name);
+ mtmv.getStatus().setState(MTMVState.NORMAL);
+ mtmv.getStatus().setRefreshState(MTMVRefreshState.SUCCESS);
+ mockRefreshedPartitions(mtmv);
+ return mtmv;
+ }
+
+ private void mockRefreshedPartitions(MTMV mtmv) {
+ // Candidate validity alone does not model a completed refresh.
Populate definition coverage
+ // and publish a data version so rewrite normalization does not prune
the MV as empty.
+ Map<String, MTMVRefreshPartitionSnapshot> snapshots = new HashMap<>();
+ for (Partition partition : mtmv.getPartitions()) {
+ long refreshVersion = partition.getNextVersion();
+ partition.setNextVersion(refreshVersion + 1);
+ partition.updateVisibleVersion(refreshVersion);
+ snapshots.put(partition.getName(), new
MTMVRefreshPartitionSnapshot());
+ }
+ mtmv.getRefreshSnapshot().updateSnapshots(snapshots,
mtmv.getPartitionNames());
+ mtmv.invalidateRewriteCache();
+ }
+
+ private void initMvRewriteSession() {
+
connectContext.getSessionVariable().setDisableNereidsRules("PRUNE_EMPTY_PARTITION");
+ connectContext.getState().setIsQuery(true);
+ connectContext.getSessionVariable().enableMaterializedViewRewrite =
true;
+ connectContext.getSessionVariable().enableMaterializedViewNestRewrite
= true;
+
connectContext.getSessionVariable().setPreMaterializedViewRewriteStrategy(PreRewriteStrategy.NOT_IN_RBO.name());
+
connectContext.getSessionVariable().materializedViewRewriteDurationThresholdMs
= 1000000;
+ }
+}
diff --git
a/regression-test/data/nereids_rules_p0/mv/nested/relation_implied_predicate_nested_mtmv.out
b/regression-test/data/nereids_rules_p0/mv/nested/relation_implied_predicate_nested_mtmv.out
new file mode 100644
index 00000000000..5fde8180fcd
--- /dev/null
+++
b/regression-test/data/nereids_rules_p0/mv/nested/relation_implied_predicate_nested_mtmv.out
@@ -0,0 +1,8 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !query_before --
+1 10 100 A
+3 20 300 B
+
+-- !query_after --
+1 10 100 A
+3 20 300 B
diff --git
a/regression-test/suites/nereids_rules_p0/mv/nested/relation_implied_predicate_nested_mtmv.groovy
b/regression-test/suites/nereids_rules_p0/mv/nested/relation_implied_predicate_nested_mtmv.groovy
new file mode 100644
index 00000000000..61d29ebc5ea
--- /dev/null
+++
b/regression-test/suites/nereids_rules_p0/mv/nested/relation_implied_predicate_nested_mtmv.groovy
@@ -0,0 +1,120 @@
+// 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("relation_implied_predicate_nested_mtmv") {
+ String db = context.config.getDbNameByFile(context.file)
+ sql "use ${db}"
+ sql "SET enable_nereids_planner=true"
+ sql "SET enable_fallback_to_original_planner=false"
+ sql "SET enable_materialized_view_rewrite=true"
+ sql "SET enable_materialized_view_nest_rewrite=true"
+ sql "SET pre_materialized_view_rewrite_strategy=NOT_IN_RBO"
+
+ sql """DROP MATERIALIZED VIEW IF EXISTS rip_mv_join"""
+ sql """DROP MATERIALIZED VIEW IF EXISTS rip_mv_fact"""
+ sql """DROP VIEW IF EXISTS rip_fact_view"""
+ sql """DROP TABLE IF EXISTS rip_fact"""
+ sql """DROP TABLE IF EXISTS rip_dim"""
+
+ sql """
+ CREATE TABLE rip_fact (
+ k1 BIGINT,
+ dim_id BIGINT,
+ sku_type VARCHAR(8),
+ amount BIGINT
+ )
+ DUPLICATE KEY(k1)
+ DISTRIBUTED BY HASH(k1) BUCKETS 1
+ PROPERTIES ("replication_num" = "1")
+ """
+
+ sql """
+ CREATE TABLE rip_dim (
+ dim_id BIGINT,
+ tag VARCHAR(16)
+ )
+ DUPLICATE KEY(dim_id)
+ DISTRIBUTED BY HASH(dim_id) BUCKETS 1
+ PROPERTIES ("replication_num" = "1")
+ """
+
+ sql """
+ INSERT INTO rip_fact VALUES
+ (1, 10, '1', 100),
+ (2, 10, '2', 200),
+ (3, 20, '1', 300)
+ """
+
+ sql """
+ INSERT INTO rip_dim VALUES
+ (10, 'A'),
+ (20, 'B')
+ """
+
+ sql """ANALYZE TABLE rip_fact WITH SYNC"""
+ sql """ANALYZE TABLE rip_dim WITH SYNC"""
+ sql """ALTER TABLE rip_fact MODIFY COLUMN amount SET STATS
('row_count'='3')"""
+ sql """ALTER TABLE rip_dim MODIFY COLUMN tag SET STATS ('row_count'='2')"""
+ sql """CREATE VIEW rip_fact_view AS SELECT * FROM rip_fact WHERE k1 > 0"""
+
+ def query = """
+ SELECT f.k1, f.dim_id, f.amount, d.tag
+ FROM rip_fact f
+ INNER JOIN rip_dim d ON f.dim_id = d.dim_id
+ WHERE f.sku_type = '1' AND f.k1 > 0
+ ORDER BY 1, 2, 3, 4
+ """
+
+ sql "SET enable_materialized_view_rewrite=false"
+ order_qt_query_before "${query}"
+
+ sql "SET enable_materialized_view_rewrite=true"
+ create_async_mv(db, "rip_mv_fact", """
+ SELECT k1, dim_id, amount, sku_type
+ FROM rip_fact_view
+ WHERE sku_type = '1'
+ """)
+ create_async_mv(db, "rip_mv_join", """
+ SELECT f.k1, f.dim_id, f.amount, d.tag
+ FROM rip_mv_fact f
+ INNER JOIN rip_dim d ON f.dim_id = d.dim_id
+ WHERE f.sku_type = '1'
+ """)
+
+ mv_rewrite_success(query, "rip_mv_join")
+ order_qt_query_after "${query}"
+
+ // Only the lower MV adopts the new view definition. The upper still
stores k1 = 1 and 3.
+ sql """ALTER MATERIALIZED VIEW rip_mv_join SET ('grace_period' =
'31536000')"""
+ sql """ALTER VIEW rip_fact_view AS SELECT * FROM rip_fact WHERE k1 > 2"""
+ sql """REFRESH MATERIALIZED VIEW rip_mv_fact COMPLETE"""
+ waitingMTMVTaskFinished(getJobName(db, "rip_mv_fact"))
+
+ def lowerMvQuery = """
+ SELECT f.k1, f.dim_id, f.amount, d.tag
+ FROM rip_mv_fact f
+ INNER JOIN rip_dim d ON f.dim_id = d.dim_id
+ WHERE f.k1 > 2
+ ORDER BY 1, 2, 3, 4
+ """
+ def staleUpperQuery = lowerMvQuery.replaceFirst("SELECT", "SELECT /*+
use_mv(rip_mv_join) */")
+ def currentLowerQuery = lowerMvQuery.replaceFirst("SELECT",
+ "SELECT /*+ SET_VAR(enable_materialized_view_rewrite=false) */")
+ // The explicit query filter must still reject k1 = 1 when grace_period
permits the old upper MV.
+ mv_rewrite_success(staleUpperQuery, "rip_mv_join")
+ check_sqls_result_equal(staleUpperQuery, currentLowerQuery)
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
