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 0f6600d2053 [fix](mv rewrite) Keep the scalar aggregate when a grouped
MV answers it (#68861)
0f6600d2053 is described below
commit 0f6600d2053fe14a7836a48ea04f4722410d8a68
Author: yujun <[email protected]>
AuthorDate: Thu Oct 22 21:56:32 2026 +0800
[fix](mv rewrite) Keep the scalar aggregate when a grouped MV answers it
(#68861)
A scalar aggregate query can be answered by a **grouped** aggregate
materialized view.
`AbstractMaterializedViewAggregateRule#isGroupByEquals` treats the query
and the view grouping as
equal when a uniform function dependency proves the filtered view holds
at most one group, and then
eliminates the aggregate entirely, leaving a bare view scan. With
```sql
select count(*) from base where k = 9999; -- k = 9999 matches no row
```
the rewritten plan was `filter -> OlapScan[mv]` with no aggregate node
at all, so the query returned
an **empty result set** instead of one row of `0`. "At most one group"
does not imply "one group
exists", while a scalar aggregate must always emit exactly one row.
COUNT is the function that hits this: its argument is not a real column,
so it is the only aggregate
whose output the fast path can substitute directly with a view slot.
`SUM`/`MIN`/`MAX` over the same
view already fall through to the roll up path and return the correct
single row.
The roll up path has a second, related gap: the slots the project above
the rewritten aggregate is
built from come from the rolled up expressions before
`LogicalAggregate`'s constructor forces a
scalar `NullableAggregateFunction` to nullable, so `AdjustNullable`
rejects the nullability flip and,
with `fe_debug` on, throws away the whole rewrite candidate.
### What is changed and how does it work?
1. `isGroupByEquals`: reject the aggregate-eliminating fast path when
the query is a scalar aggregate
and the view is grouped, so the rewrite goes through the roll up path
and keeps the aggregate node.
The `query scalar + view scalar` combination is untouched — there the
view always holds exactly one
row, so eliminating the aggregate is safe.
2. roll up: build the query output to rewritten output mapping, and the
`needTopProject` comparison,
from the constructed aggregate's output expressions, so the slots carry
the nullability the node
actually produces.
3. `Predicates.compensatePredicates`: stop pruning a compensation
candidate as redundant merely because
the view states the same condition, when that candidate constrains an
aggregate or a window output and
the rewrite computes that value again. A view with a `HAVING` of its own
enforced the condition over the
value *it* aggregated, while the query needs it over the value *its own*
aggregate produces from the
view rows, and a roll up changes that value: over the empty row set it
is `0`, which the retained scalar
aggregate would answer with a row while the query yields none. Such a
predicate now stays in the
compensation, where the existing `rejectUnsafeResidualCompensation`
check sees it and refuses the
rewrite, so a view carrying a `HAVING` is rejected for these queries the
same way it already is for
every other query against it. Whether the value is computed again comes
from a new
`aggregateOutputIsRecomputed` hook which the aggregate rule sets for
exactly the shape item 1 keeps the
aggregate for, a query without grouping over a view which groups by
something; a query and a view that
group by the same dimensions and repeat the same condition still read
the stored value, so the view
still answers them.
The empty-input guarantee itself needs no compensation here: #68863,
already merged, made
`Count.constructRollUp` produce `SUM0`, which is a
`NotNullableAggregateFunction` and returns `0` for
an empty input just like `COUNT` does. That is why this PR no longer
wraps the rolled up aggregate in
`Nvl` — it would only add a redundant `ifnull` on top of a roll up that
already guarantees the value.
A query `HAVING` over the retained scalar aggregate keeps working for a
view without a `HAVING` of
its own: the filter stays above the aggregate and is evaluated on the
rolled up value, so a `HAVING`
the single rolled up row does not pass yields no row at all. For a view
which does have one, the query
is now answered by the base table instead of by that view (see 3 above);
before this change such a
pair was rewritten by dropping the query `HAVING`, which answered
`select count(*) from t where k = 9999 having count(*) > 0` with a row
of `0` where the query yields
no row.
Before / after for the query above:
```
empty set
```
```
+----------+
| count(*) |
+----------+
| 0 |
+----------+
```
and the plan now keeps the aggregate over the view:
```
PhysicalResultSink
--PhysicalProject
----hashAgg[GLOBAL]
------PhysicalProject
--------filter((mv.k = 9999))
----------PhysicalOlapScan[mv]
```
### Release note
Two related changes for scalar aggregates rewritten by a grouped
aggregate materialized view:
- `select count(*) from t where <predicate matching no row>` now returns
`0` instead of an empty result
set.
- a scalar query whose `HAVING` repeats the `HAVING` of an aggregate
materialized view is no longer
rewritten by that view, because the roll up does not preserve the
condition the view filtered on.
---
.../mv/AbstractMaterializedViewAggregateRule.java | 47 +++++-
.../mv/AbstractMaterializedViewRule.java | 12 +-
.../nereids/rules/exploration/mv/Predicates.java | 83 +++++++---
.../apache/doris/nereids/mv/PredicatesTest.java | 47 +++++-
.../exploration/mv/MvExplorationSuiteTest.java | 55 +++++++
.../agg_scalar_empty_input.out | 138 +++++++++++++++++
.../agg_scalar_empty_input.groovy | 171 +++++++++++++++++++++
7 files changed, 527 insertions(+), 26 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/AbstractMaterializedViewAggregateRule.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/AbstractMaterializedViewAggregateRule.java
index 046793de494..77e3cde0117 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/AbstractMaterializedViewAggregateRule.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/AbstractMaterializedViewAggregateRule.java
@@ -294,6 +294,7 @@ public abstract class AbstractMaterializedViewAggregateRule
extends AbstractMate
// The shuttled query bottom aggregate outputs are used as the map
keys and are passed to the
// rewriter directly, and the shuttled query top plan outputs restore
the projection expressions,
// so the whole top plan is traversed only twice instead of once per
output expression.
+ // The map values are filled after the rewritten aggregate is
constructed, see the comment there.
List<? extends Expression> shuttledBottomAggOutputs =
ExpressionUtils.shuttleExpressionWithLineage(
queryAggregate.getOutputExpressions(), queryTopPlan);
List<? extends Expression> shuttledTopPlanOutputs =
ExpressionUtils.shuttleExpressionWithLineage(
@@ -319,7 +320,6 @@ public abstract class AbstractMaterializedViewAggregateRule
extends AbstractMate
? (NamedExpression) rewrittenGroupByExpression : new
Alias(rewrittenGroupByExpression);
finalGroupExpressions.add(groupByOutput);
finalOutputExpressions.add(groupByOutput);
- bottomAggOutputToNewExprMap.put(shuttledQueryAggregateOutput,
groupByOutput.toSlot());
} else {
// if it is an aggregate function, try to roll up and rewrite
Expression rewrittenFunction =
rewriteShuttledExpression(queryStructInfo,
@@ -333,13 +333,23 @@ public abstract class
AbstractMaterializedViewAggregateRule extends AbstractMate
}
NamedExpression functionOutput = new Alias(rewrittenFunction);
finalOutputExpressions.add(functionOutput);
- bottomAggOutputToNewExprMap.put(shuttledQueryAggregateOutput,
functionOutput.toSlot());
}
}
LogicalAggregate<Plan> rewrittenAggregate =
new LogicalAggregate<>(finalGroupExpressions,
finalOutputExpressions, tempRewritedPlan);
+ // Each rewritten aggregate output corresponds to one query aggregate
output, so map the query
+ // aggregate output to the new output slot by position. The slot must
be taken from the constructed
+ // aggregate: its constructor forces a NullableAggregateFunction of an
aggregate without group by
+ // to nullable, since a scalar aggregate over an empty input returns
NULL, and a slot built from
+ // the expression before that adjustment keeps the old nullability,
which AdjustNullable rejects.
+ List<NamedExpression> rewrittenAggregateOutputs =
rewrittenAggregate.getOutputExpressions();
+ for (int i = 0; i < queryAggregateOutputs.size(); i++) {
+ NamedExpression rewrittenOutput = rewrittenAggregateOutputs.get(i);
+ bottomAggOutputToNewExprMap.put(shuttledBottomAggOutputs.get(i),
rewrittenOutput.toSlot());
+ }
+
// rewrite the query top plan output expressions to reference the
rewritten aggregate output,
// the query top plan output slot is shuttled by lineage firstly to
restore the projection
// expression, so a projection of the group by key in the query top
plan can be recomputed
@@ -366,10 +376,10 @@ public abstract class
AbstractMaterializedViewAggregateRule extends AbstractMate
// (e.g. `select k1 from t group by k1, k2`), in which case the
redundant aggregate outputs must be
// projected away. Otherwise the rewritten plan output count differs
from the query and the candidate
// is rejected by MaterializedViewUtils.normalizeExpressions, so a
valid sync MV is silently not used.
- boolean needTopProject = topProjectExpressions.size() !=
finalOutputExpressions.size();
+ boolean needTopProject = topProjectExpressions.size() !=
rewrittenAggregateOutputs.size();
for (int i = 0; i < topProjectExpressions.size(); i++) {
- if (i >= finalOutputExpressions.size()
- ||
!topProjectExpressions.get(i).toSlot().equals(finalOutputExpressions.get(i).toSlot()))
{
+ if (i >= rewrittenAggregateOutputs.size()
+ ||
!topProjectExpressions.get(i).toSlot().equals(rewrittenAggregateOutputs.get(i).toSlot()))
{
needTopProject = true;
break;
}
@@ -570,6 +580,23 @@ public abstract class
AbstractMaterializedViewAggregateRule extends AbstractMate
return true;
}
+ @Override
+ protected boolean aggregateOutputIsRecomputed(StructInfo queryStructInfo,
StructInfo viewStructInfo) {
+ // A predicate the view enforces over its own aggregate output answers
the query only while the query
+ // reads the value the view stored. isGroupByEquals keeps the
aggregate for a query without grouping
+ // over a view which groups by something, and that roll up aggregates
the stored values, of no row at
+ // all when the filter matches none. The roll ups of the other shapes
are the ones upstream has.
+ Pair<Plan, LogicalAggregate<Plan>> queryTopPlanAndAggPair =
splitToTopPlanAndAggregate(queryStructInfo);
+ Pair<Plan, LogicalAggregate<Plan>> viewTopPlanAndAggPair =
splitToTopPlanAndAggregate(viewStructInfo);
+ if (queryTopPlanAndAggPair == null || viewTopPlanAndAggPair == null) {
+ // Without an aggregate on both sides, the view has no aggregate
output of its own to have
+ // enforced a predicate over, so there is no recomputed value to
protect.
+ return false;
+ }
+ return queryTopPlanAndAggPair.value().getGroupByExpressions().isEmpty()
+ &&
!viewTopPlanAndAggPair.value().getGroupByExpressions().isEmpty();
+ }
+
private boolean isGroupByEquals(Pair<Plan, LogicalAggregate<Plan>>
queryTopPlanAndAggPair,
Pair<Plan, LogicalAggregate<Plan>> viewTopPlanAndAggPair,
SlotMapping viewToQuerySlotMapping,
@@ -588,6 +615,16 @@ public abstract class
AbstractMaterializedViewAggregateRule extends AbstractMate
LogicalAggregate<Plan> queryAggregate = queryTopPlanAndAggPair.value();
LogicalAggregate<Plan> viewAggregate = viewTopPlanAndAggPair.value();
+ // If the query is a scalar aggregate but the view groups by extra
dimensions, the dimensions
+ // removed by the uniform function dependency below only guarantee at
most one group, not that a
+ // group exists. Eliminating the aggregate into a bare project would
lose the single row a scalar
+ // aggregate must always emit when the filter matches zero rows. Keep
the aggregate by rejecting
+ // this fast path and falling back to the roll up rewrite.
+ if (queryAggregate.getGroupByExpressions().isEmpty()
+ && !viewAggregate.getGroupByExpressions().isEmpty()) {
+ return false;
+ }
+
Set<Expression> queryGroupByShuttledExpression = new
HashSet<>(ExpressionUtils.shuttleExpressionWithLineage(
queryAggregate.getGroupByExpressions(), queryTopPlan));
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/AbstractMaterializedViewRule.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/AbstractMaterializedViewRule.java
index 1302176bd9a..2b4abd0aca7 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/AbstractMaterializedViewRule.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/AbstractMaterializedViewRule.java
@@ -853,13 +853,23 @@ public abstract class AbstractMaterializedViewRule
implements ExplorationRuleFac
}
PredicateCompensation finalPredicateCompensation =
Predicates.compensatePredicates(queryStructInfo,
viewStructInfo,
- viewToQuerySlotMapping, comparisonResult,
cascadesContext);
+ viewToQuerySlotMapping, comparisonResult,
cascadesContext,
+ aggregateOutputIsRecomputed(queryStructInfo,
viewStructInfo));
if (finalPredicateCompensation == null) {
return SplitPredicate.INVALID_INSTANCE;
}
return finalPredicateCompensation.toSplitPredicate();
}
+ /**
+ * Whether the rewrite answers the query by aggregating the view rows
again, instead of reading the
+ * aggregate values that the view stored. A predicate the view enforced
over its own aggregate output
+ * says nothing about such a recomputed value, so compensation can not
treat it as redundant.
+ */
+ protected boolean aggregateOutputIsRecomputed(StructInfo queryStructInfo,
StructInfo viewStructInfo) {
+ return false;
+ }
+
/**
* Check whether query-side null-reject evidence covers each required
view-side slot set.
*
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 10088fd1b7a..7b1412a168b 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
@@ -283,13 +283,33 @@ public class Predicates {
return new PredicateCompensation(equalCandidates, rangeCandidates,
residualCandidates);
}
- /** Compensate predicates in one step. */
+ /**
+ * Compensate predicates in one step, for a rewrite which reads the
aggregate values of the view as the
+ * view stored them.
+ */
public static PredicateCompensation compensatePredicates(
StructInfo queryStructInfo,
StructInfo viewStructInfo,
SlotMapping viewToQuerySlotMapping,
ComparisonResult comparisonResult,
CascadesContext cascadesContext) {
+ return compensatePredicates(queryStructInfo, viewStructInfo,
viewToQuerySlotMapping, comparisonResult,
+ cascadesContext, false);
+ }
+
+ /**
+ * Compensate predicates in one step.
+ *
+ * @param aggregateOutputRecomputed whether the rewrite answers the query
by aggregating the view rows
+ * again, rather than reading the aggregate values that the view
already stored
+ */
+ public static PredicateCompensation compensatePredicates(
+ StructInfo queryStructInfo,
+ StructInfo viewStructInfo,
+ SlotMapping viewToQuerySlotMapping,
+ ComparisonResult comparisonResult,
+ CascadesContext cascadesContext,
+ boolean aggregateOutputRecomputed) {
PredicateCompensation compensationCandidates =
collectCompensationCandidates(
queryStructInfo, viewStructInfo, viewToQuerySlotMapping,
comparisonResult, cascadesContext);
if (compensationCandidates == null) {
@@ -297,13 +317,12 @@ public class Predicates {
}
Set<Expression> queryBasedViewResidualPredicates =
collectNonInferredQueryBasedExpressions(
viewStructInfo.getSplitPredicate().getResidualPredicateMap().keySet(),
viewToQuerySlotMapping);
- Set<Expression> exactCoveredPredicates = new LinkedHashSet<>();
- exactCoveredPredicates.addAll(Sets.intersection(
- compensationCandidates.getEquals().keySet(),
queryBasedViewResidualPredicates));
- exactCoveredPredicates.addAll(Sets.intersection(
- compensationCandidates.getRanges().keySet(),
queryBasedViewResidualPredicates));
- exactCoveredPredicates.addAll(Sets.intersection(
- compensationCandidates.getResiduals().keySet(),
queryBasedViewResidualPredicates));
+ // A candidate the view already answers is pruned here; one over a
value the query computes again is
+ // kept, so that the check at the end refuses the rewrite instead of
dropping the filter.
+ Set<Expression> prunableCandidates = collectPrunableCandidates(
+ compensationCandidates, aggregateOutputRecomputed);
+ Set<Expression> exactCoveredPredicates = new LinkedHashSet<>(
+ Sets.intersection(prunableCandidates,
queryBasedViewResidualPredicates));
Set<Expression> remainingQueryBasedViewResidualPredicates = new
LinkedHashSet<>(
Sets.difference(queryBasedViewResidualPredicates,
exactCoveredPredicates));
@@ -333,12 +352,12 @@ public class Predicates {
}
PredicateCompensation finalCompensation = new
PredicateCompensation(
- removePredicatesImpliedByViewResidual(
- exactPrunedCompensationCandidates.getEquals(),
combinedQueryBasedViewResidual),
- removePredicatesImpliedByViewResidual(
- exactPrunedCompensationCandidates.getRanges(),
combinedQueryBasedViewResidual),
- removePredicatesImpliedByViewResidual(
- exactPrunedCompensationCandidates.getResiduals(),
combinedQueryBasedViewResidual));
+
removePredicatesImpliedByViewResidual(exactPrunedCompensationCandidates.getEquals(),
+ combinedQueryBasedViewResidual,
aggregateOutputRecomputed),
+
removePredicatesImpliedByViewResidual(exactPrunedCompensationCandidates.getRanges(),
+ combinedQueryBasedViewResidual,
aggregateOutputRecomputed),
+
removePredicatesImpliedByViewResidual(exactPrunedCompensationCandidates.getResiduals(),
+ combinedQueryBasedViewResidual,
aggregateOutputRecomputed));
return rejectUnsafeResidualCompensation(finalCompensation);
} catch (DnfBranchOverflowException e) {
// DNF branch expansion may explode exponentially; fail
compensation conservatively.
@@ -356,10 +375,27 @@ public class Predicates {
return ImmutableMap.copyOf(residualCandidates);
}
+ private static Set<Expression> collectPrunableCandidates(
+ PredicateCompensation candidates, boolean
aggregateOutputRecomputed) {
+ Set<Expression> prunableCandidates = new LinkedHashSet<>();
+ for (Map<Expression, ExpressionInfo> predicateMap : ImmutableList.of(
+ candidates.getEquals(), candidates.getRanges(),
candidates.getResiduals())) {
+ for (Expression candidate : predicateMap.keySet()) {
+ if (isPrunable(candidate, aggregateOutputRecomputed)) {
+ prunableCandidates.add(candidate);
+ }
+ }
+ }
+ return prunableCandidates;
+ }
+
+ private static boolean constrainsAggregateOrWindowOutput(Expression
expression) {
+ return expression.containsType(AggregateFunction.class,
WindowExpression.class);
+ }
+
private static PredicateCompensation
rejectUnsafeResidualCompensation(PredicateCompensation compensation) {
for (Expression expression : compensation.getResiduals().keySet()) {
- if (expression.anyMatch(WindowExpression.class::isInstance)
- ||
expression.anyMatch(AggregateFunction.class::isInstance)) {
+ if (constrainsAggregateOrWindowOutput(expression)) {
// Aggregate and window residuals are not safe as detail-MV
compensation predicates.
return null;
}
@@ -399,18 +435,27 @@ public class Predicates {
private static Map<Expression, ExpressionInfo>
removePredicatesImpliedByViewResidual(
Map<Expression, ExpressionInfo> predicates,
- Expression viewResidual) {
+ Expression viewResidual,
+ boolean aggregateOutputRecomputed) {
// Remove candidates already implied by the view residual predicate.
Map<Expression, ExpressionInfo> remainingPredicates = new
LinkedHashMap<>();
for (Map.Entry<Expression, ExpressionInfo> entry :
predicates.entrySet()) {
- // If viewResidual => candidate, candidate is redundant and can be
dropped.
- if (!impliesByDnf(viewResidual, entry.getKey())) {
+ if (!isPrunable(entry.getKey(), aggregateOutputRecomputed)
+ || !impliesByDnf(viewResidual, entry.getKey())) {
remainingPredicates.put(entry.getKey(), entry.getValue());
}
}
return ImmutableMap.copyOf(remainingPredicates);
}
+ /**
+ * Whether the view already answers the candidate. A candidate over an
aggregate or window output does
+ * not follow from the view while the query computes that value again from
the view rows.
+ */
+ private static boolean isPrunable(Expression candidate, boolean
aggregateOutputRecomputed) {
+ return !(aggregateOutputRecomputed &&
constrainsAggregateOrWindowOutput(candidate));
+ }
+
private static Map<Expression, ExpressionInfo>
removeExactCoveredPredicates(
Map<Expression, ExpressionInfo> predicates,
Set<Expression> exactCoveredPredicates) {
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/mv/PredicatesTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/mv/PredicatesTest.java
index 4c70811dcb5..c3dad1506dc 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/nereids/mv/PredicatesTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/mv/PredicatesTest.java
@@ -29,6 +29,7 @@ import
org.apache.doris.nereids.rules.exploration.mv.mapping.SlotMapping;
import org.apache.doris.nereids.sqltest.SqlTestBase;
import org.apache.doris.nereids.trees.expressions.Expression;
import org.apache.doris.nereids.trees.expressions.WindowExpression;
+import
org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
import org.apache.doris.nereids.trees.plans.Plan;
import org.apache.doris.nereids.util.PlanChecker;
@@ -365,6 +366,44 @@ public class PredicatesTest extends SqlTestBase {
Assertions.assertTrue(finalPredicateCompensation.getResiduals().isEmpty());
}
+ @Test
+ public void
testCoveredAggregateResidualIsRejectedWhenAggregateOutputIsRecomputed() {
+ // The view groups by id and has count(*) > 10; the scalar query
aggregates the view rows again.
+ PredicateRewriteContext rewriteContext = buildRewriteContext(
+ "select id, count(*) as c from T1 group by id having count(*)
> 10",
+ "select count(*) from T1 where id = 5 having count(*) > 10");
+
+ PredicateCompensation compensationCandidates =
Predicates.collectCompensationCandidates(
+ rewriteContext.queryStructInfo,
+ rewriteContext.viewStructInfo,
+ rewriteContext.viewToQuerySlotMapping,
+ rewriteContext.comparisonResult,
+ rewriteContext.queryContext);
+ Assertions.assertNotNull(compensationCandidates);
+
Assertions.assertTrue(compensationCandidates.getResiduals().keySet().stream()
+ .anyMatch(expression ->
expression.containsType(AggregateFunction.class)));
+
+ // The view enforced count(*) > 10 over the values it stored, while
the roll up produces a new
+ // value, zero when the filter matches no view row. The covered
residual can not be pruned, so the
+ // unsafe residual check refuses the rewrite.
+ Assertions.assertNull(compensatePredicates(rewriteContext, true));
+
+ // A rewrite which reads the stored values may prune the covered
residual, the filter stays above a
+ // single aggregate output in both the query and the view.
+ PredicateCompensation valuePreservingCompensation =
compensatePredicates(rewriteContext, false);
+ Assertions.assertNotNull(valuePreservingCompensation);
+
Assertions.assertTrue(valuePreservingCompensation.getResiduals().isEmpty());
+ }
+
+ @Test
+ public void
testCoveredWindowResidualIsRejectedWhenWindowOutputIsRecomputed() {
+ String sql = "select id, score from T1 qualify sum(score) over
(partition by id) > 10";
+ PredicateRewriteContext rewriteContext = buildRewriteContext(sql, sql);
+
+ // The same condition over a window output which the query computes
again is not covered by the view.
+ Assertions.assertNull(compensatePredicates(rewriteContext, true));
+ }
+
@Test
public void testQueryOnlyWindowResidualIsRejectedAsCompensation() {
PredicateRewriteContext rewriteContext = buildRewriteContext(
@@ -507,12 +546,18 @@ public class PredicatesTest extends SqlTestBase {
}
private PredicateCompensation compensatePredicates(PredicateRewriteContext
rewriteContext) {
+ return compensatePredicates(rewriteContext, false);
+ }
+
+ private PredicateCompensation compensatePredicates(PredicateRewriteContext
rewriteContext,
+ boolean aggregateOutputRecomputed) {
return Predicates.compensatePredicates(
rewriteContext.queryStructInfo,
rewriteContext.viewStructInfo,
rewriteContext.viewToQuerySlotMapping,
rewriteContext.comparisonResult,
- rewriteContext.queryContext);
+ rewriteContext.queryContext,
+ aggregateOutputRecomputed);
}
private void assertResidualCompensationSucceeds(String viewPredicate,
String queryPredicate,
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/exploration/mv/MvExplorationSuiteTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/exploration/mv/MvExplorationSuiteTest.java
index b59982b082b..ade4a3aeeff 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/exploration/mv/MvExplorationSuiteTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/exploration/mv/MvExplorationSuiteTest.java
@@ -973,6 +973,61 @@ public class MvExplorationSuiteTest extends SqlTestBase {
Assertions.assertTrue(compensatePredicates.isInvalid());
}
+ //
-------------------------------------------------------------------------
+ // scalar aggregate over a grouped view
+ //
-------------------------------------------------------------------------
+
+ @Test
+ void testAggregateOutputRecomputedForScalarQueryOverGroupedView() {
+ connectContext.getSessionVariable().setDisableNereidsRules(
+ "OLAP_SCAN_PARTITION_PRUNE"
+ + ",PRUNE_EMPTY_PARTITION"
+ + ",ELIMINATE_GROUP_BY_KEY_BY_UNIFORM"
+ + ",ELIMINATE_CONST_JOIN_CONDITION"
+ + ",CONSTANT_PROPAGATION"
+ );
+ TestAggregateMaterializedViewRule rule = new
TestAggregateMaterializedViewRule();
+ StructInfo groupedView = buildStructInfo("select id, count(*) as c
from T1 group by id");
+
+ // The query has no grouping, so the roll up aggregates the values the
view stored again.
+ Assertions.assertTrue(rule.aggregateOutputIsRecomputedForTest(
+ buildStructInfo("select count(*) from T1"), groupedView));
+
+ // The same grouping reads those values as they are stored.
+ Assertions.assertFalse(rule.aggregateOutputIsRecomputedForTest(
+ buildStructInfo("select id, count(*) as c from T1 group by
id"), groupedView));
+
+ // A query without an aggregate has no aggregate output to compute
again.
+ Assertions.assertFalse(rule.aggregateOutputIsRecomputedForTest(
+ buildStructInfo("select id from T1"), groupedView));
+
+ // A scalar view stores the single value which the scalar query reads.
+ Assertions.assertFalse(rule.aggregateOutputIsRecomputedForTest(
+ buildStructInfo("select count(*) from T1"),
+ buildStructInfo("select count(*) as c from T1")));
+ }
+
+ private StructInfo buildStructInfo(String sql) {
+ CascadesContext cascadesContext = createCascadesContext(sql,
connectContext);
+ Plan plan = PlanChecker.from(cascadesContext)
+ .analyze()
+ .rewrite()
+ .getPlan().child(0);
+ return StructInfo.of(plan, plan, cascadesContext);
+ }
+
+ private static class TestAggregateMaterializedViewRule extends
AbstractMaterializedViewAggregateRule {
+ @Override
+ public List<Rule> buildRules() {
+ return ImmutableList.of();
+ }
+
+ private boolean aggregateOutputIsRecomputedForTest(StructInfo
queryStructInfo,
+ StructInfo viewStructInfo) {
+ return aggregateOutputIsRecomputed(queryStructInfo,
viewStructInfo);
+ }
+ }
+
private static boolean isNotNullOnSlot(Expression expression, String
slotName) {
if (!(expression instanceof Not) || ((Not)
expression).isGeneratedIsNotNull()
|| !(((Not) expression).child() instanceof IsNull)) {
diff --git
a/regression-test/data/nereids_rules_p0/mv/agg_scalar_empty_input/agg_scalar_empty_input.out
b/regression-test/data/nereids_rules_p0/mv/agg_scalar_empty_input/agg_scalar_empty_input.out
new file mode 100644
index 00000000000..390a29f6030
--- /dev/null
+++
b/regression-test/data/nereids_rules_p0/mv/agg_scalar_empty_input/agg_scalar_empty_input.out
@@ -0,0 +1,138 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !count_off_empty --
+0
+
+-- !count_off_matched --
+20
+
+-- !count_off_all --
+2000
+
+-- !count_off_projected --
+1
+
+-- !count_off_distinct --
+0
+
+-- !sum_off_empty --
+\N
+
+-- !sum_off_matched --
+19020
+
+-- !sum_off_all --
+1999000
+
+-- !min_off_empty --
+\N
+
+-- !min_off_matched --
+1
+
+-- !max_off_empty --
+\N
+
+-- !max_off_matched --
+1901
+
+-- !count_on_empty --
+0
+
+-- !count_on_matched --
+20
+
+-- !count_on_all --
+2000
+
+-- !count_on_projected --
+1
+
+-- !count_on_distinct --
+0
+
+-- !sum_on_empty --
+\N
+
+-- !sum_on_matched --
+19020
+
+-- !sum_on_all --
+1999000
+
+-- !min_on_empty --
+\N
+
+-- !min_on_matched --
+1
+
+-- !max_on_empty --
+\N
+
+-- !max_on_matched --
+1901
+
+-- !shape_count_on --
+PhysicalResultSink
+--PhysicalProject
+----hashAgg[GLOBAL]
+------PhysicalProject
+--------filter((scalar_agg_empty_mv.k = 9999))
+----------PhysicalOlapScan[scalar_agg_empty_mv]
+
+-- !shape_min_on --
+PhysicalResultSink
+--PhysicalProject
+----hashAgg[GLOBAL]
+------PhysicalProject
+--------filter((scalar_agg_empty_mv.k = 9999))
+----------PhysicalOlapScan[scalar_agg_empty_mv]
+
+-- !having_off_passes --
+20
+
+-- !having_off_rejects --
+
+-- !having_off_empty --
+
+-- !having_on_passes --
+20
+
+-- !having_on_rejects --
+
+-- !having_on_empty --
+
+-- !shape_having --
+PhysicalResultSink
+--PhysicalProject
+----filter((sum0(cnt) > 100000))
+------hashAgg[GLOBAL]
+--------PhysicalProject
+----------filter((scalar_agg_empty_mv.k = 1))
+------------PhysicalOlapScan[scalar_agg_empty_mv]
+
+-- !having_mv_off_matched --
+20
+
+-- !having_mv_off_empty --
+
+-- !having_mv_on_matched --
+20
+
+-- !having_mv_on_empty --
+
+-- !uniform_off --
+1 1
+
+-- !uniform_off_empty --
+
+-- !uniform_on --
+1 1
+
+-- !uniform_on_empty --
+
+-- !shape_uniform --
+PhysicalResultSink
+--PhysicalProject
+----filter((scalar_agg_uniform_mv.v = 1001))
+------PhysicalOlapScan[scalar_agg_uniform_mv]
+
diff --git
a/regression-test/suites/nereids_rules_p0/mv/agg_scalar_empty_input/agg_scalar_empty_input.groovy
b/regression-test/suites/nereids_rules_p0/mv/agg_scalar_empty_input/agg_scalar_empty_input.groovy
new file mode 100644
index 00000000000..10b07d38d13
--- /dev/null
+++
b/regression-test/suites/nereids_rules_p0/mv/agg_scalar_empty_input/agg_scalar_empty_input.groovy
@@ -0,0 +1,171 @@
+// 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("agg_scalar_empty_input") {
+ String db = context.config.getDbNameByFile(context.file)
+ sql "use ${db}"
+ sql "set pre_materialized_view_rewrite_strategy = TRY_IN_RBO"
+ sql "set runtime_filter_mode=OFF"
+ sql "SET ignore_shape_nodes='PhysicalDistribute'"
+ // The same query text runs once with the rewrite off and once with it on,
so the cached result of the
+ // first run would be replayed for the second one instead of the rewritten
plan being executed.
+ sql "SET enable_sql_cache=false"
+
+ sql """DROP MATERIALIZED VIEW IF EXISTS scalar_agg_empty_mv"""
+ sql """DROP MATERIALIZED VIEW IF EXISTS scalar_agg_uniform_mv"""
+ sql """DROP TABLE IF EXISTS scalar_agg_empty_base"""
+
+ sql """
+ CREATE TABLE scalar_agg_empty_base (k INT NOT NULL, v INT NOT NULL)
+ DUPLICATE KEY(k)
+ DISTRIBUTED BY HASH(k) BUCKETS 1
+ PROPERTIES ("replication_num" = "1")
+ """
+
+ // 2000 rows over 100 distinct k values, so the pre-aggregated mv is much
cheaper to read than
+ // the base table and the rewritten plan is the one the optimizer picks. k
= 9999 matches nothing,
+ // and k = 1 holds v = 1, 101, ... 1901.
+ sql """INSERT INTO scalar_agg_empty_base SELECT number % 100, number FROM
numbers("number" = "2000")"""
+ sql """ANALYZE TABLE scalar_agg_empty_base WITH SYNC"""
+
+ def mvSql = """select k, count(*) as cnt, sum(v) as s, min(v) as mn,
max(v) as mx """ +
+ """from scalar_agg_empty_base group by k"""
+
+ def qCountEmpty = """select count(*) from scalar_agg_empty_base where k =
9999"""
+ def qCountMatched = """select count(*) from scalar_agg_empty_base where k
= 1"""
+ def qCountAll = """select count(*) from scalar_agg_empty_base"""
+ def qCountProjected = """select count(*) + 1 from scalar_agg_empty_base
where k = 9999"""
+ def qCountDistinct = """select count(distinct v) from
scalar_agg_empty_base where k = 9999"""
+ def qSumEmpty = """select sum(v) from scalar_agg_empty_base where k =
9999"""
+ def qSumMatched = """select sum(v) from scalar_agg_empty_base where k =
1"""
+ def qSumAll = """select sum(v) from scalar_agg_empty_base"""
+ def qMinEmpty = """select min(v) from scalar_agg_empty_base where k =
9999"""
+ def qMinMatched = """select min(v) from scalar_agg_empty_base where k =
1"""
+ def qMaxEmpty = """select max(v) from scalar_agg_empty_base where k =
9999"""
+ def qMaxMatched = """select max(v) from scalar_agg_empty_base where k =
1"""
+
+ // Reference results, computed on the base table: a scalar aggregate
always emits exactly one
+ // row, COUNT of an empty input is 0 while SUM/MIN/MAX of an empty input
are NULL.
+ sql "set enable_materialized_view_rewrite=false"
+ order_qt_count_off_empty "${qCountEmpty}"
+ order_qt_count_off_matched "${qCountMatched}"
+ order_qt_count_off_all "${qCountAll}"
+ order_qt_count_off_projected "${qCountProjected}"
+ order_qt_count_off_distinct "${qCountDistinct}"
+ order_qt_sum_off_empty "${qSumEmpty}"
+ order_qt_sum_off_matched "${qSumMatched}"
+ order_qt_sum_off_all "${qSumAll}"
+ order_qt_min_off_empty "${qMinEmpty}"
+ order_qt_min_off_matched "${qMinMatched}"
+ order_qt_max_off_empty "${qMaxEmpty}"
+ order_qt_max_off_matched "${qMaxMatched}"
+
+ // The mv groups by k while these queries are scalar, so the mv dimensions
may only be dropped
+ // if the aggregate node is kept: rolling COUNT up must yield 0 and
SUM/MIN/MAX must yield NULL.
+ sql "set enable_materialized_view_rewrite=true"
+ async_mv_rewrite_success(db, mvSql, qCountEmpty, "scalar_agg_empty_mv")
+ sql """ANALYZE TABLE scalar_agg_empty_mv WITH SYNC"""
+ mv_rewrite_success(qCountMatched, "scalar_agg_empty_mv")
+ mv_rewrite_success(qCountProjected, "scalar_agg_empty_mv")
+ mv_rewrite_success(qSumEmpty, "scalar_agg_empty_mv")
+ mv_rewrite_success(qSumMatched, "scalar_agg_empty_mv")
+ mv_rewrite_success(qSumAll, "scalar_agg_empty_mv")
+ mv_rewrite_success(qMinEmpty, "scalar_agg_empty_mv")
+ mv_rewrite_success(qMinMatched, "scalar_agg_empty_mv")
+ mv_rewrite_success(qMaxEmpty, "scalar_agg_empty_mv")
+ mv_rewrite_success(qMaxMatched, "scalar_agg_empty_mv")
+ // count(distinct v) is stored in the mv as an already computed value and
can not be rolled up,
+ // so the mv must not answer this query at all.
+ mv_rewrite_fail(qCountDistinct, "scalar_agg_empty_mv")
+ // count(*) without a predicate is answered straight from the FE table
metadata by
+ // REWRITE_SIMPLE_AGG_TO_CONSTANT, so the mv is not expected to be used
for it.
+
+ // The scalar aggregate must survive the rewrite: an mv scan alone would
return no row at all.
+ order_qt_count_on_empty "${qCountEmpty}"
+ order_qt_count_on_matched "${qCountMatched}"
+ order_qt_count_on_all "${qCountAll}"
+ order_qt_count_on_projected "${qCountProjected}"
+ order_qt_count_on_distinct "${qCountDistinct}"
+ order_qt_sum_on_empty "${qSumEmpty}"
+ order_qt_sum_on_matched "${qSumMatched}"
+ order_qt_sum_on_all "${qSumAll}"
+ order_qt_min_on_empty "${qMinEmpty}"
+ order_qt_min_on_matched "${qMinMatched}"
+ order_qt_max_on_empty "${qMaxEmpty}"
+ order_qt_max_on_matched "${qMaxMatched}"
+ // Whether the scalar aggregate is split into a local and a global
aggregate is a cost based
+ // choice which depends on the number of backends, so pin the one phase
aggregation here to keep
+ // the shape assertion about the aggregate itself independent of the
cluster it runs on.
+ sql "set agg_phase=1"
+ qt_shape_count_on """explain shape plan ${qCountEmpty}"""
+ qt_shape_min_on """explain shape plan ${qMinEmpty}"""
+
+ // A having clause filters the rows the top aggregate produces. Keeping
the aggregate is what makes the
+ // roll up possible, and the filter then stays above it and is evaluated
on the rolled up value, so the
+ // mv is still used and a having which the single rolled up row does not
pass yields no row at all.
+ def qHavingPasses = """select count(*) from scalar_agg_empty_base where k
= 1 having count(*) > 0"""
+ def qHavingRejects = """select count(*) from scalar_agg_empty_base where k
= 1 having count(*) > 100000"""
+ def qHavingEmptyInput = """select count(*) from scalar_agg_empty_base
where k = 9999 having count(*) > 0"""
+ sql "set enable_materialized_view_rewrite=false"
+ order_qt_having_off_passes "${qHavingPasses}"
+ order_qt_having_off_rejects "${qHavingRejects}"
+ order_qt_having_off_empty "${qHavingEmptyInput}"
+ sql "set enable_materialized_view_rewrite=true"
+ mv_rewrite_success(qHavingPasses, "scalar_agg_empty_mv")
+ mv_rewrite_success(qHavingRejects, "scalar_agg_empty_mv")
+ mv_rewrite_success(qHavingEmptyInput, "scalar_agg_empty_mv")
+ order_qt_having_on_passes "${qHavingPasses}"
+ order_qt_having_on_rejects "${qHavingRejects}"
+ order_qt_having_on_empty "${qHavingEmptyInput}"
+ qt_shape_having """explain shape plan ${qHavingRejects}"""
+
+ // An mv with a having clause of its own stores only the groups which
passed that clause, so the clause
+ // is a condition on the value the mv aggregated. A scalar query which
repeats the very same condition
+ // does not ask for the same value: its aggregate is computed over the mv
rows its own filter selects,
+ // and over nothing at all an aggregate rolls up to 0 while the query
computed on the base table yields
+ // no row. The mv can not answer such a query, so it must not be used for
it.
+ def havingMvSql = """select k, count(*) as cnt from scalar_agg_empty_base
group by k having count(*) > 0"""
+ def qHavingMvMatched = """select count(*) from scalar_agg_empty_base where
k = 1 having count(*) > 0"""
+ def qHavingMvEmpty = """select count(*) from scalar_agg_empty_base where k
= 9999 having count(*) > 0"""
+ async_mv_rewrite_fail(db, havingMvSql, qHavingMvEmpty,
"scalar_agg_having_mv")
+ sql """ANALYZE TABLE scalar_agg_having_mv WITH SYNC"""
+ mv_rewrite_fail(qHavingMvMatched, "scalar_agg_having_mv")
+ sql "set enable_materialized_view_rewrite=false"
+ order_qt_having_mv_off_matched "${qHavingMvMatched}"
+ order_qt_having_mv_off_empty "${qHavingMvEmpty}"
+ sql "set enable_materialized_view_rewrite=true"
+ order_qt_having_mv_on_matched "${qHavingMvMatched}"
+ order_qt_having_mv_on_empty "${qHavingMvEmpty}"
+
+ // A query which keeps one of the mv dimensions may still drop the
aggregate node: the equality
+ // filter makes the remaining mv dimension uniform, so the mv holds
exactly one row per group.
+ def uniformMvSql = """select k, v, count(*) as cnt from
scalar_agg_empty_base group by k, v"""
+ def qUniform = """select k, count(*) from scalar_agg_empty_base where v =
1001 group by k"""
+ def qUniformEmpty = """select k, count(*) from scalar_agg_empty_base where
v = 999999 group by k"""
+ sql """DROP MATERIALIZED VIEW IF EXISTS scalar_agg_empty_mv"""
+ async_mv_rewrite_success(db, uniformMvSql, qUniform,
"scalar_agg_uniform_mv")
+ sql """ANALYZE TABLE scalar_agg_uniform_mv WITH SYNC"""
+ mv_rewrite_success(qUniformEmpty, "scalar_agg_uniform_mv")
+
+ sql "set enable_materialized_view_rewrite=false"
+ order_qt_uniform_off "${qUniform}"
+ order_qt_uniform_off_empty "${qUniformEmpty}"
+ sql "set enable_materialized_view_rewrite=true"
+ order_qt_uniform_on "${qUniform}"
+ order_qt_uniform_on_empty "${qUniformEmpty}"
+ qt_shape_uniform """explain shape plan ${qUniform}"""
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
