This is an automated email from the ASF dual-hosted git repository.
yujun777 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 9e60fab4db0 [fix](mtmv) Eliminate redundant filters in nested MV
candidate definitions (#63639)
9e60fab4db0 is described below
commit 9e60fab4db00998d490d916c4bf304b15490c1b9
Author: foxtail463 <[email protected]>
AuthorDate: Wed Oct 21 12:03:50 2026 +0800
[fix](mtmv) Eliminate redundant filters in nested MV candidate definitions
(#63639)
Nested MV rewriting can fail when an upper MV repeats a filter that is
already guaranteed by its lower MV.
For example:
```sql
mv_fact_snapshot_v1:
SELECT k1, dim_id, sku_type, amount
FROM fact
WHERE sku_type = '1';
```
```sql
mv_detail_trade_joined_v1:
SELECT t.k1, t.dim_id, t.sku_type, t.amount, d.tag
FROM mv_fact_snapshot_v1 t
JOIN dim d ON t.dim_id = d.dim_id
WHERE t.sku_type = '1';
```
For the query:
```sql
SELECT f.k1, f.dim_id, f.sku_type, f.amount, d.tag
FROM fact f
JOIN dim d ON f.dim_id = d.dim_id
WHERE f.sku_type = '1';
```
The query can match mv_fact_snapshot_v1, but matching
mv_detail_trade_joined_v1 may be rejected because of the repeated
t.sku_type = '1' filter. However, mv_fact_snapshot_v1 already guarantees
that every output row satisfies this predicate, so the upper-MV filter
is redundant.
### Solution:
Use reliable predicates guaranteed by the lower MV to prove redundant
filters in upper-MV definitions during candidate matching. This allows
nested rewriting to match mv_detail_trade_joined_v1 without requiring
the upper MV to expose additional filter columns.
Query predicates must still be preserved for compensation. Cached MV
definitions remain unchanged. Output-predicate guarantees are withheld
whenever the MV definition or refresh state cannot reliably establish
them.
---------
Co-authored-by: yangtao555 <[email protected]>
---
.../main/java/org/apache/doris/catalog/MTMV.java | 63 +-
.../main/java/org/apache/doris/mtmv/MTMVCache.java | 36 +-
.../org/apache/doris/nereids/StatementContext.java | 25 +-
.../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 | 242 ++++++++
.../test/java/org/apache/doris/mtmv/MTMVTest.java | 18 +-
.../org/apache/doris/nereids/mv/MTMVCacheTest.java | 103 +++-
.../nereids/mv/RelationImpliedPredicateTest.java | 659 +++++++++++++++++++++
.../relation_implied_predicate_nested_mtmv.out | 8 +
.../relation_implied_predicate_nested_mtmv.groovy | 120 ++++
15 files changed, 1431 insertions(+), 48 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 9805be4e157..6572a0edca9 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
@@ -297,6 +297,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);
}
@@ -426,7 +427,12 @@ public class MTMV extends OlapTable {
applyRefreshedEpochs(capturedEpochs);
}
}
+ boolean publishCache = false;
if (task.getStatus() == TaskStatus.SUCCESS) {
+ publishCache = needUpdateCache && cacheGeneration ==
rewriteCacheGeneration && !isDropped;
+ // A successful refresh can restore a changed definition,
including after several
+ // partial refreshes. Discard caches and builds from the
previous snapshot coverage.
+ invalidateRewriteCache();
this.status.setState(MTMVState.NORMAL);
this.status.setSchemaChangeDetail(null);
this.status.setRefreshState(MTMVRefreshState.SUCCESS);
@@ -437,21 +443,6 @@ public class MTMV extends OlapTable {
ivmInfo.setPlanSignature(refreshedIvmPlanSignature);
}
}
- // 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) {
- 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 {
this.status.setRefreshState(MTMVRefreshState.FAIL);
}
@@ -465,6 +456,17 @@ public class MTMV extends OlapTable {
snapshotsToWrite =
snapshotsOfCleanPartitions(partitionSnapshots);
}
this.refreshSnapshot.updateSnapshots(snapshotsToWrite,
getPartitionNames());
+ if (publishCache) {
+ // Publish only after the refreshed partitions have been
recorded. A partial
+ // refresh may restore NORMAL while other partitions still use
the old definition.
+ MTMVCacheManager manager =
Env.getCurrentEnv().getMtmvCacheManager();
+ if (mtmvCacheWithGuard != null) {
+ manager.put(this.id, true,
cacheWithValidOutputPredicates(mtmvCacheWithGuard));
+ }
+ if (mtmvCacheWithoutGuard != null) {
+ manager.put(this.id, false,
cacheWithValidOutputPredicates(mtmvCacheWithoutGuard));
+ }
+ }
Env.getCurrentEnv().getMtmvService()
.refreshComplete(this, relation, task);
if (isReplay) {
@@ -726,11 +728,11 @@ public class MTMV extends OlapTable {
MTMVCache cached;
readMvLock();
try {
+ cacheGeneration = rewriteCacheGeneration;
cached = manager.isEnabled() ? manager.getIfPresent(this.id,
guarded) : null;
if (cached == null && statementContext != null) {
- cached = statementContext.getQueryLocalMtmvCache(this.id,
guarded);
+ cached = statementContext.getQueryLocalMtmvCache(this.id,
guarded, cacheGeneration);
}
- cacheGeneration = rewriteCacheGeneration;
} finally {
readMvUnlock();
}
@@ -744,23 +746,24 @@ public class MTMV extends OlapTable {
// Someone invalidated between our snapshot and now; drop
the stale build and retry.
continue;
}
+ MTMVCache published =
cacheWithValidOutputPredicates(generated);
if (manager.isEnabled()) {
MTMVCache existing = manager.getIfPresent(this.id,
guarded);
if (existing != null) {
return existing;
}
if (!isDropped) {
- manager.put(this.id, guarded, generated);
+ manager.put(this.id, guarded, published);
}
} 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);
+ MTMVCache existing =
statementContext.getQueryLocalMtmvCache(this.id, guarded, cacheGeneration);
if (existing != null) {
return existing;
}
- statementContext.putQueryLocalMtmvCache(this.id, guarded,
generated);
+ statementContext.putQueryLocalMtmvCache(this.id, guarded,
cacheGeneration, published);
}
- return generated;
+ return published;
} finally {
readMvUnlock();
}
@@ -1769,6 +1772,16 @@ public class MTMV extends OlapTable {
this.mvRwLock.writeLock().unlock();
}
+ private MTMVCache cacheWithValidOutputPredicates(MTMVCache cache) {
+ // A cache built from the current definition can prove facts about
stored rows only after
+ // every partition has been refreshed. Callers hold the MV lock while
publishing it.
+ 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{");
@@ -1806,8 +1819,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()));
}
}
@@ -1840,8 +1853,8 @@ public class MTMV extends OlapTable {
}
if (refreshInfo != null && refreshInfo.getRefreshMethod() == null) {
LOG.warn("MTMV {} has unknown refresh method, marking as schema
change", name);
- status.setState(MTMVState.SCHEMA_CHANGE);
- status.setSchemaChangeDetail("Unknown refresh method detected
during deserialization");
+ alterStatus(new MTMVStatus(MTMVState.SCHEMA_CHANGE,
+ "Unknown refresh method detected during deserialization"));
}
Map<String, MTMVRefreshPartitionSnapshot> partitionSnapshots =
refreshSnapshot.getPartitionSnapshots();
compatiblePctSnapshot(partitionSnapshots);
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 8b168cac931..60f512abd4c 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.model.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/nereids/StatementContext.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/StatementContext.java
index 2913af5f004..4401e038d1a 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
@@ -166,6 +166,9 @@ public class StatementContext implements Closeable {
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
@@ -339,8 +342,9 @@ public class StatementContext implements Closeable {
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<>();
+ // in the same statement so multiple rewrite paths do not rebuild the same
MV plan. Each value
+ // carries the MV's cache generation because a planning retry can reuse
this statement context.
+ private final Map<Pair<Long, Boolean>, Pair<Long, MTMVCache>>
queryLocalMtmvCaches = new HashMap<>();
/// for dictionary sink.
private List<Backend> usedBackendsDistributing; // report used backends
after done distribute planning.
@@ -628,6 +632,14 @@ public class StatementContext implements Closeable {
return hasNondeterministic;
}
+ public void recordAliasUdf() {
+ hasAliasUdf = true;
+ }
+
+ public boolean hasAliasUdf() {
+ return hasAliasUdf;
+ }
+
public ConnectContext getConnectContext() {
return connectContext;
}
@@ -1677,12 +1689,13 @@ public class StatementContext implements Closeable {
this.materializationRewrittenSuccessSet.add(materializationQualifier);
}
- public MTMVCache getQueryLocalMtmvCache(long mtmvId, boolean guarded) {
- return queryLocalMtmvCaches.get(Pair.of(mtmvId, guarded));
+ public MTMVCache getQueryLocalMtmvCache(long mtmvId, boolean guarded, long
generation) {
+ Pair<Long, MTMVCache> entry = queryLocalMtmvCaches.get(Pair.of(mtmvId,
guarded));
+ return entry != null && entry.first == generation ? entry.second :
null;
}
- public void putQueryLocalMtmvCache(long mtmvId, boolean guarded, MTMVCache
cache) {
- queryLocalMtmvCaches.put(Pair.of(mtmvId, guarded), cache);
+ public void putQueryLocalMtmvCache(long mtmvId, boolean guarded, long
generation, MTMVCache cache) {
+ queryLocalMtmvCaches.put(Pair.of(mtmvId, guarded), Pair.of(generation,
cache));
}
public Multimap<List<String>, Pair<RelationId, Set<String>>>
getTableUsedPartitionNameMap() {
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 18e32ddedaa..48cb5050ef6 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
@@ -26,7 +26,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;
@@ -199,7 +198,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
@@ -620,14 +624,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 ac6c6879929..03c733c7c47 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
@@ -123,9 +123,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 e5109b5516c..41766a0cafc 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..c9d4202be36
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVOutputPredicatesTest.java
@@ -0,0 +1,242 @@
+// 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.catalog.info.TableNameInfo;
+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.mtmv.ivm.IvmInfo;
+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.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.qe.ConnectContext;
+import org.apache.doris.qe.SessionVariable;
+import org.apache.doris.thrift.TStorageType;
+
+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.List;
+import java.util.Map;
+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;
+import java.util.concurrent.atomic.AtomicLong;
+
+class MTMVOutputPredicatesTest {
+ private static final AtomicLong NEXT_MTMV_ID = new
AtomicLong(1_000_000_000L);
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testDefinitionChangeInvalidatesPublishedGuarantees(boolean
baseViewChange) {
+ CacheMTMV mtmv = new CacheMTMV();
+ MTMVCache beforeChange = mtmv.getOrGenerateCache(mtmv.context);
+ Assertions.assertEquals(Set.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);
+ MTMVCacheManager manager = new MTMVCacheManager();
+ MTMVService service = Mockito.mock(MTMVService.class);
+ EditLog editLog = Mockito.mock(EditLog.class);
+ EditLogItem editLogItem = Mockito.mock(EditLogItem.class);
+ Mockito.when(env.getMtmvService()).thenReturn(service);
+ Mockito.when(env.getMtmvCacheManager()).thenReturn(manager);
+ Mockito.when(env.getEditLog()).thenReturn(editLog);
+
Mockito.when(editLog.submitEdit(Mockito.eq(OperationType.OP_ALTER_MTMV),
Mockito.any(AlterMTMV.class)))
+ .thenReturn(editLogItem);
+
+ 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,
"p1"), true));
+ Assertions.assertEquals(MTMVState.NORMAL,
mtmv.getStatus().getState());
+ Assertions.assertEquals(Set.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,
"p2"), false));
+ int buildsAtRefreshCompletion = mtmv.buildCount.get();
+ MTMVCache fullyRefreshedCache =
mtmv.getOrGenerateCache(mtmv.context);
+ Assertions.assertNotSame(partiallyRefreshedCache,
fullyRefreshedCache);
+ Assertions.assertEquals(Set.of("p1", "p2"),
+
mtmv.getRefreshSnapshot().getPartitionSnapshots().keySet());
+ Assertions.assertEquals(Set.of(mtmv.predicate),
fullyRefreshedCache.getOutputPredicates());
+ Assertions.assertEquals(buildsAtRefreshCompletion,
mtmv.buildCount.get());
+ }
+ Mockito.verify(editLogItem).await();
+ }
+
+ private static AlterMTMV successfulRefresh(CacheMTMV mtmv, String
partition) {
+ MTMVTask task = new MTMVTask(mtmv, mtmv.getRelation(),
+ new MTMVTaskContext(MTMVTaskTriggerMode.MANUAL));
+ task.setTaskId(1L);
+ task.setStatus(TaskStatus.SUCCESS);
+ Deencapsulation.setField(task, "mtmvSchemaChangeVersion",
mtmv.getSchemaChangeVersion());
+ AlterMTMV result = new AlterMTMV(new TableNameInfo("db", "mv"),
MTMVAlterOpType.ADD_TASK);
+ result.setTask(task);
+ result.setRelation(mtmv.getRelation());
+ result.setPartitionSnapshots(Map.of(partition, new
MTMVRefreshPartitionSnapshot()));
+ return result;
+ }
+
+ 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),
+ List.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(NEXT_MTMV_ID.incrementAndGet());
+ 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(Set.of(), Set.of(), Set.of(),
Set.of(), Set.of()));
+ setMvPartitionInfo(new MTMVPartitionInfo());
+ Deencapsulation.setField(this, "ivmInfo", new IvmInfo());
+ Deencapsulation.setField(this, "sessionVariables", new HashMap<>(
+
context.getSessionVariable().getAffectQueryResultInPlanVariables()));
+ setBaseIndexId(1L);
+ setIndexMeta(1L, "mv", List.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(Map.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, List.of(),
Set.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/mtmv/MTMVTest.java
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVTest.java
index 1636d09517f..62dd652f68f 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
@@ -408,6 +408,9 @@ public class MTMVTest {
EditLog editLog = Mockito.mock(EditLog.class);
EditLogItem editLogItem = Mockito.mock(EditLogItem.class);
Mockito.when(env.getEditLog()).thenReturn(editLog);
+ // The invalidation path drops the rewrite cache under the MV lock, so
the mocked env
+ // has to serve a real cache manager, otherwise invalidateRewriteCache
hits a null one.
+ Mockito.when(env.getMtmvCacheManager()).thenReturn(new
MTMVCacheManager());
Mockito.when(editLog.submitEdit(Mockito.eq(OperationType.OP_ALTER_MTMV),
Mockito.any(AlterMTMV.class)))
.thenAnswer(invocation -> {
Assertions.assertTrue(mvRwLock.isWriteLockedByCurrentThread());
@@ -553,7 +556,7 @@ public class MTMVTest {
}
@Test
- public void testDisabledCacheReusesPlanWithinSameStatement() throws
Exception {
+ public void testDisabledCacheReusesPlanOnlyWithinCurrentGeneration()
throws Exception {
int originalMaxSize = Config.mtmv_cache_manage_num;
try {
Config.mtmv_cache_manage_num = 0;
@@ -562,7 +565,9 @@ public class MTMVTest {
HookedMTMV mtmv = buildHookedMTMV();
MTMVCache plan = Mockito.mock(MTMVCache.class);
+ MTMVCache refreshedPlan = Mockito.mock(MTMVCache.class);
mtmv.lazyCaches.add(plan);
+ mtmv.lazyCaches.add(refreshedPlan);
ConnectContext context = mockConnectContext();
StatementContext statementContext = new StatementContext();
@@ -575,11 +580,18 @@ public class MTMVTest {
MTMVCache second = mtmv.getOrGenerateCache(context);
Assertions.assertSame(plan, first);
Assertions.assertSame(first, second);
+ Assertions.assertEquals(1, mtmv.lazyBuildCount);
+
+ // INSERT may plan again with this same StatementContext after
releasing its table locks.
+ // A refresh between attempts must invalidate the local copy
as well as the global cache.
+
Assertions.assertTrue(mtmv.addTaskResult(buildSuccessTaskResult(mtmv), false));
+ MTMVCache afterRefresh = mtmv.getOrGenerateCache(context);
+ Assertions.assertSame(refreshedPlan, afterRefresh);
+ Assertions.assertSame(afterRefresh,
mtmv.getOrGenerateCache(context));
}
- Assertions.assertEquals(1, mtmv.lazyBuildCount);
+ Assertions.assertEquals(2, mtmv.lazyBuildCount);
Assertions.assertNull(manager.getIfPresent(mtmv.getId(), false));
- Assertions.assertSame(plan,
statementContext.getQueryLocalMtmvCache(mtmv.getId(), false));
} finally {
Config.mtmv_cache_manage_num = originalMaxSize;
}
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 08c54591192..1e43a1c5afa 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,20 +19,33 @@ 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.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.SqlModeHelper;
+import com.google.common.collect.ImmutableList;
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.Collections;
import java.util.List;
import java.util.Optional;
@@ -42,6 +55,7 @@ import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
/**
* Relevant test case about mtmv cache.
@@ -143,12 +157,97 @@ public class MTMVCacheTest extends SqlTestBase {
}
}
+ @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 static final AtomicLong NEXT_MTMV_ID = new
AtomicLong(2_000_000_000L);
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());
- private final MTMVCache secondCache = new MTMVCache(null, null, null,
Collections.emptyList());
+ 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() {
+ setId(NEXT_MTMV_ID.incrementAndGet());
+ setStatus(new MTMVStatus(MTMVState.NORMAL, null));
+ setRefreshSnapshot(new MTMVRefreshSnapshot());
+ }
@Override
protected MTMVCache createRewriteCache(ConnectContext currentContext,
boolean needLock,
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..72c3b257b36
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/mv/RelationImpliedPredicateTest.java
@@ -0,0 +1,659 @@
+// 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.Column;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.MTMV;
+import org.apache.doris.common.Config;
+import org.apache.doris.mtmv.MTMVCache;
+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.SlotReference;
+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.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+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 {
+
+ @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");
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testIvmOutputPredicatesSkipHiddenColumns(boolean showHiddenColumns)
throws Exception {
+ initMvRewriteSession();
+ boolean originalEnableTableStream = Config.enable_table_stream;
+ boolean originalEnableFeatureBinlog = Config.enable_feature_binlog;
+ boolean originalShowHiddenColumns =
connectContext.getSessionVariable().showHiddenColumns();
+ Config.enable_table_stream = true;
+ // master gates ROW-binlog table creation behind
enable_feature_binlog; the IVM base table needs it.
+ Config.enable_feature_binlog = true;
+ try {
+ dropMvByNereids("drop materialized view if exists
mv_relation_implied_ivm");
+ dropTableWithSql("drop table if exists relation_implied_ivm_base");
+ createTable("create table relation_implied_ivm_base (id bigint,
score bigint) "
+ + "duplicate key(id) distributed by hash(id) buckets 1 "
+ + "properties('replication_num' = '1', 'binlog.enable' =
'true', 'binlog.format' = 'ROW')");
+ createMvByNereids("create materialized view
mv_relation_implied_ivm "
+ + "BUILD DEFERRED REFRESH INCREMENTAL ON MANUAL\n"
+ + "DISTRIBUTED BY RANDOM BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1', 'ivm_force_mow' =
'true')\n"
+ + "as select id, score from relation_implied_ivm_base
where score = 1;");
+ mockRefreshedMtmv("mv_relation_implied_ivm");
+ CascadesContext context = createCascadesContext("select id, score
from mv_relation_implied_ivm");
+ PlanChecker.from(context).setIsQuery().analyze();
+ LogicalOlapScan scan = context.getRewritePlan()
+
.<LogicalOlapScan>collectFirst(LogicalOlapScan.class::isInstance).orElseThrow(AssertionError::new);
+ Assertions.assertTrue(scan.getOutput().stream().anyMatch(slot ->
slot instanceof SlotReference
+ && slot.getName().equals(Column.IVM_ROW_ID_COL) &&
!((SlotReference) slot).isVisible()));
+ Set<Slot> visibleOutput = scan.getOutput().stream()
+ .filter(slot -> ((SlotReference)
slot).isVisible()).collect(Collectors.toSet());
+ MTMVCache cache = ((MTMV)
scan.getTable()).getOrGenerateCache(connectContext);
+ Assertions.assertFalse(cache.getOutputPredicates().isEmpty(),
+ () -> cache.getOriginalFinalPlan().treeString() + "\n"
+ +
cache.getAllRulesRewrittenPlanAndStructInfo().key().treeString());
+ // Definition output binding always uses visible physical columns,
regardless of
+ // whether this query's session asks to display hidden columns.
+
connectContext.getSessionVariable().setShowHiddenColumns(showHiddenColumns);
+ Set<Expression> predicates = new
MaterializedViewPredicateCollector(context).collect(scan);
+ Assertions.assertFalse(predicates.isEmpty());
+ Assertions.assertTrue(predicates.stream()
+ .allMatch(predicate ->
visibleOutput.containsAll(predicate.getInputSlots())), predicates::toString);
+ Slot score = visibleOutput.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 {
+
connectContext.getSessionVariable().setShowHiddenColumns(originalShowHiddenColumns);
+ Config.enable_table_stream = originalEnableTableStream;
+ Config.enable_feature_binlog = originalEnableFeatureBinlog;
+ dropMvByNereids("drop materialized view if exists
mv_relation_implied_ivm");
+ dropTableWithSql("drop table if exists relation_implied_ivm_base");
+ }
+ }
+
+ @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) {
+ MTMV mtmv = mockCandidateMtmv(name);
+ mockRefreshedPartitions(mtmv);
+ return mtmv;
+ }
+
+ private void mockRefreshedPartitions(MTMV mtmv) {
+ // Candidate validity alone does not model a completed refresh.
Populate definition coverage
+ // for every fixture partition, as addTaskResult does after successful
refreshes.
+ Map<String, MTMVRefreshPartitionSnapshot> snapshots = new HashMap<>();
+ for (String partitionName : mtmv.getPartitionNames()) {
+ snapshots.put(partitionName, new MTMVRefreshPartitionSnapshot());
+ }
+ mtmv.getRefreshSnapshot().updateSnapshots(snapshots,
mtmv.getPartitionNames());
+ mtmv.invalidateRewriteCache();
+ }
+
+ private void initMvRewriteSession() {
+
connectContext.getSessionVariable().setDisableNereidsRules("PRUNE_EMPTY_PARTITION");
+ installValidRelationManager();
+ 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]
