[feat](mtmv) Support grouping_sets rewrite when query rewrite by materialized view (#36056) (#37436)
cherry pick from master pr: #36056 commitId: 569c9772
This commit is contained in:
seawinde
@ -89,16 +89,23 @@ public class NormalizeRepeat extends OneAnalysisRuleFactory {
|
||||
return new LogicalAggregate<>(repeat.getGroupByExpressions(),
|
||||
repeat.getOutputExpressions(), repeat.child());
|
||||
}
|
||||
checkRepeatLegality(repeat);
|
||||
repeat = removeDuplicateColumns(repeat);
|
||||
// add virtual slot, LogicalAggregate and LogicalProject for normalize
|
||||
LogicalAggregate<Plan> agg = normalizeRepeat(repeat);
|
||||
return dealSlotAppearBothInAggFuncAndGroupingSets(agg);
|
||||
return doNormalize(repeat);
|
||||
})
|
||||
);
|
||||
}
|
||||
|
||||
private LogicalRepeat<Plan> removeDuplicateColumns(LogicalRepeat<Plan> repeat) {
|
||||
/**
|
||||
* Normalize repeat, this can be used directly, if optimize the repeat
|
||||
*/
|
||||
public static LogicalAggregate<Plan> doNormalize(LogicalRepeat<Plan> repeat) {
|
||||
checkRepeatLegality(repeat);
|
||||
repeat = removeDuplicateColumns(repeat);
|
||||
// add virtual slot, LogicalAggregate and LogicalProject for normalize
|
||||
LogicalAggregate<Plan> agg = normalizeRepeat(repeat);
|
||||
return dealSlotAppearBothInAggFuncAndGroupingSets(agg);
|
||||
}
|
||||
|
||||
private static LogicalRepeat<Plan> removeDuplicateColumns(LogicalRepeat<Plan> repeat) {
|
||||
List<List<Expression>> groupingSets = repeat.getGroupingSets();
|
||||
ImmutableList.Builder<List<Expression>> builder = ImmutableList.builder();
|
||||
for (List<Expression> sets : groupingSets) {
|
||||
@ -108,11 +115,11 @@ public class NormalizeRepeat extends OneAnalysisRuleFactory {
|
||||
return repeat.withGroupSets(builder.build());
|
||||
}
|
||||
|
||||
private void checkRepeatLegality(LogicalRepeat<Plan> repeat) {
|
||||
private static void checkRepeatLegality(LogicalRepeat<Plan> repeat) {
|
||||
checkGroupingSetsSize(repeat);
|
||||
}
|
||||
|
||||
private void checkGroupingSetsSize(LogicalRepeat<Plan> repeat) {
|
||||
private static void checkGroupingSetsSize(LogicalRepeat<Plan> repeat) {
|
||||
Set<Expression> flattenGroupingSetExpr = ImmutableSet.copyOf(
|
||||
ExpressionUtils.flatExpressions(repeat.getGroupingSets()));
|
||||
if (flattenGroupingSetExpr.size() > LogicalRepeat.MAX_GROUPING_SETS_NUM) {
|
||||
@ -122,7 +129,7 @@ public class NormalizeRepeat extends OneAnalysisRuleFactory {
|
||||
}
|
||||
}
|
||||
|
||||
private LogicalAggregate<Plan> normalizeRepeat(LogicalRepeat<Plan> repeat) {
|
||||
private static LogicalAggregate<Plan> normalizeRepeat(LogicalRepeat<Plan> repeat) {
|
||||
Set<Expression> needToSlotsGroupingExpr = collectNeedToSlotGroupingExpr(repeat);
|
||||
NormalizeToSlotContext groupingExprContext = buildContext(repeat, needToSlotsGroupingExpr);
|
||||
Map<Expression, NormalizeToSlotTriplet> groupingExprMap = groupingExprContext.getNormalizeToSlotMap();
|
||||
@ -198,14 +205,14 @@ public class NormalizeRepeat extends OneAnalysisRuleFactory {
|
||||
Optional.of(normalizedRepeat), normalizedRepeat);
|
||||
}
|
||||
|
||||
private Set<Expression> collectNeedToSlotGroupingExpr(LogicalRepeat<Plan> repeat) {
|
||||
private static Set<Expression> collectNeedToSlotGroupingExpr(LogicalRepeat<Plan> repeat) {
|
||||
// grouping sets should be pushed down, e.g. grouping sets((k + 1)),
|
||||
// we should push down the `k + 1` to the bottom plan
|
||||
return ImmutableSet.copyOf(
|
||||
ExpressionUtils.flatExpressions(repeat.getGroupingSets()));
|
||||
}
|
||||
|
||||
private Set<Expression> collectNeedToSlotArgsOfGroupingScalarFuncAndAggFunc(LogicalRepeat<Plan> repeat) {
|
||||
private static Set<Expression> collectNeedToSlotArgsOfGroupingScalarFuncAndAggFunc(LogicalRepeat<Plan> repeat) {
|
||||
Set<GroupingScalarFunction> groupingScalarFunctions = ExpressionUtils.collect(
|
||||
repeat.getOutputExpressions(), GroupingScalarFunction.class::isInstance);
|
||||
ImmutableSet.Builder<Expression> argumentsSetBuilder = ImmutableSet.builder();
|
||||
@ -237,7 +244,7 @@ public class NormalizeRepeat extends OneAnalysisRuleFactory {
|
||||
.build();
|
||||
}
|
||||
|
||||
private Plan pushDownProject(Set<NamedExpression> pushedExprs, Plan originBottomPlan) {
|
||||
private static Plan pushDownProject(Set<NamedExpression> pushedExprs, Plan originBottomPlan) {
|
||||
if (!pushedExprs.equals(originBottomPlan.getOutputSet()) && !pushedExprs.isEmpty()) {
|
||||
return new LogicalProject<>(ImmutableList.copyOf(pushedExprs), originBottomPlan);
|
||||
}
|
||||
@ -245,7 +252,7 @@ public class NormalizeRepeat extends OneAnalysisRuleFactory {
|
||||
}
|
||||
|
||||
/** buildContext */
|
||||
public NormalizeToSlotContext buildContext(Repeat<? extends Plan> repeat,
|
||||
public static NormalizeToSlotContext buildContext(Repeat<? extends Plan> repeat,
|
||||
Set<? extends Expression> sourceExpressions) {
|
||||
Set<Alias> aliases = ExpressionUtils.collect(repeat.getOutputExpressions(), Alias.class::isInstance);
|
||||
Map<Expression, Alias> existsAliasMap = Maps.newLinkedHashMap();
|
||||
@ -271,7 +278,7 @@ public class NormalizeRepeat extends OneAnalysisRuleFactory {
|
||||
return new NormalizeToSlotContext(normalizeToSlotMap);
|
||||
}
|
||||
|
||||
private Optional<NormalizeToSlotTriplet> toGroupingSetExpressionPushDownTriplet(
|
||||
private static Optional<NormalizeToSlotTriplet> toGroupingSetExpressionPushDownTriplet(
|
||||
Expression expression, @Nullable Alias existsAlias) {
|
||||
NormalizeToSlotTriplet originTriplet = NormalizeToSlotTriplet.toTriplet(expression, existsAlias);
|
||||
SlotReference remainSlot = (SlotReference) originTriplet.remainExpr;
|
||||
@ -279,7 +286,8 @@ public class NormalizeRepeat extends OneAnalysisRuleFactory {
|
||||
return Optional.of(new NormalizeToSlotTriplet(expression, newSlot, originTriplet.pushedExpr));
|
||||
}
|
||||
|
||||
private Expression normalizeAggFuncChildrenAndGroupingScalarFunc(NormalizeToSlotContext context, Expression expr) {
|
||||
private static Expression normalizeAggFuncChildrenAndGroupingScalarFunc(NormalizeToSlotContext context,
|
||||
Expression expr) {
|
||||
if (expr instanceof AggregateFunction) {
|
||||
AggregateFunction function = (AggregateFunction) expr;
|
||||
List<Expression> normalizedRealExpressions = context.normalizeToUseSlotRef(function.getArguments());
|
||||
@ -296,7 +304,7 @@ public class NormalizeRepeat extends OneAnalysisRuleFactory {
|
||||
}
|
||||
}
|
||||
|
||||
private Set<Alias> getExistsAlias(LogicalRepeat<Plan> repeat,
|
||||
private static Set<Alias> getExistsAlias(LogicalRepeat<Plan> repeat,
|
||||
Map<Expression, NormalizeToSlotTriplet> groupingExprMap) {
|
||||
Set<Alias> existsAlias = Sets.newHashSet();
|
||||
Set<Alias> aliases = ExpressionUtils.collect(repeat.getOutputExpressions(), Alias.class::isInstance);
|
||||
@ -323,7 +331,7 @@ public class NormalizeRepeat extends OneAnalysisRuleFactory {
|
||||
* +--LogicalRepeat (groupingSets=[[a#0]], outputExpr=[a#0, a#3, GROUPING_ID#1]
|
||||
* +--LogicalProject (projects =[a#0, a#0 as `a`#3])
|
||||
*/
|
||||
private LogicalAggregate<Plan> dealSlotAppearBothInAggFuncAndGroupingSets(
|
||||
private static LogicalAggregate<Plan> dealSlotAppearBothInAggFuncAndGroupingSets(
|
||||
@NotNull LogicalAggregate<Plan> aggregate) {
|
||||
LogicalRepeat<Plan> repeat = (LogicalRepeat<Plan>) aggregate.child();
|
||||
Map<Slot, Alias> commonSlotToAliasMap = getCommonSlotToAliasMap(repeat, aggregate);
|
||||
@ -370,7 +378,8 @@ public class NormalizeRepeat extends OneAnalysisRuleFactory {
|
||||
return aggregate.withAggOutput(newOutputExpressions);
|
||||
}
|
||||
|
||||
private Map<Slot, Alias> getCommonSlotToAliasMap(LogicalRepeat<Plan> repeat, LogicalAggregate<Plan> aggregate) {
|
||||
private static Map<Slot, Alias> getCommonSlotToAliasMap(LogicalRepeat<Plan> repeat,
|
||||
LogicalAggregate<Plan> aggregate) {
|
||||
List<AggregateFunction> aggregateFunctions =
|
||||
CollectNonWindowedAggFuncs.collect(aggregate.getOutputExpressions());
|
||||
ImmutableSet.Builder<Slot> aggUsedSlotBuilder = ImmutableSet.builder();
|
||||
|
||||
@ -19,6 +19,7 @@ package org.apache.doris.nereids.rules.exploration.mv;
|
||||
|
||||
import org.apache.doris.common.Pair;
|
||||
import org.apache.doris.nereids.CascadesContext;
|
||||
import org.apache.doris.nereids.rules.analysis.NormalizeRepeat;
|
||||
import org.apache.doris.nereids.rules.exploration.mv.StructInfo.PlanCheckContext;
|
||||
import org.apache.doris.nereids.rules.exploration.mv.StructInfo.PlanSplitContext;
|
||||
import org.apache.doris.nereids.rules.exploration.mv.mapping.SlotMapping;
|
||||
@ -32,13 +33,16 @@ import org.apache.doris.nereids.trees.expressions.ExprId;
|
||||
import org.apache.doris.nereids.trees.expressions.Expression;
|
||||
import org.apache.doris.nereids.trees.expressions.NamedExpression;
|
||||
import org.apache.doris.nereids.trees.expressions.Slot;
|
||||
import org.apache.doris.nereids.trees.expressions.VirtualSlotReference;
|
||||
import org.apache.doris.nereids.trees.expressions.functions.Function;
|
||||
import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
|
||||
import org.apache.doris.nereids.trees.expressions.functions.agg.RollUpTrait;
|
||||
import org.apache.doris.nereids.trees.expressions.functions.scalar.GroupingScalarFunction;
|
||||
import org.apache.doris.nereids.trees.expressions.visitor.DefaultExpressionRewriter;
|
||||
import org.apache.doris.nereids.trees.plans.Plan;
|
||||
import org.apache.doris.nereids.trees.plans.algebra.Repeat;
|
||||
import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
|
||||
import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
|
||||
import org.apache.doris.nereids.trees.plans.logical.LogicalRepeat;
|
||||
import org.apache.doris.nereids.trees.plans.visitor.ExpressionLineageReplacer;
|
||||
import org.apache.doris.nereids.util.ExpressionUtils;
|
||||
|
||||
@ -51,7 +55,9 @@ import java.util.BitSet;
|
||||
import java.util.HashSet;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Optional;
|
||||
import java.util.Set;
|
||||
import java.util.function.Supplier;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
/**
|
||||
@ -91,10 +97,16 @@ public abstract class AbstractMaterializedViewAggregateRule extends AbstractMate
|
||||
() -> String.format("query plan = %s\n", queryStructInfo.getOriginalPlan().treeString()));
|
||||
return null;
|
||||
}
|
||||
// Firstly,if group by expression between query and view is equals, try to rewrite expression directly
|
||||
Plan queryTopPlan = queryTopPlanAndAggPair.key();
|
||||
if (isGroupByEquals(queryTopPlanAndAggPair, viewTopPlanAndAggPair, viewToQuerySlotMapping, queryStructInfo,
|
||||
viewStructInfo, materializationContext)) {
|
||||
LogicalAggregate<Plan> queryAggregate = queryTopPlanAndAggPair.value();
|
||||
if (!checkCompatibility(queryStructInfo, queryAggregate, viewTopPlanAndAggPair.value(),
|
||||
materializationContext)) {
|
||||
return null;
|
||||
}
|
||||
boolean queryContainsGroupSets = queryAggregate.getSourceRepeat().isPresent();
|
||||
// If group by expression between query and view is equals, try to rewrite expression directly
|
||||
if (!queryContainsGroupSets && isGroupByEquals(queryTopPlanAndAggPair, viewTopPlanAndAggPair,
|
||||
viewToQuerySlotMapping, queryStructInfo, viewStructInfo, materializationContext)) {
|
||||
List<Expression> rewrittenQueryExpressions = rewriteExpression(queryTopPlan.getOutput(),
|
||||
queryTopPlan,
|
||||
materializationContext.getShuttledExprToScanExprMapping(),
|
||||
@ -124,20 +136,6 @@ public abstract class AbstractMaterializedViewAggregateRule extends AbstractMate
|
||||
materializationContext.getShuttledExprToScanExprMapping(),
|
||||
viewToQuerySlotMapping));
|
||||
}
|
||||
// if view is scalar aggregate but query is not. Or if query is scalar aggregate but view is not
|
||||
// Should not rewrite
|
||||
List<Expression> queryGroupByExpressions = queryTopPlanAndAggPair.value().getGroupByExpressions();
|
||||
List<Expression> viewGroupByExpressions = viewTopPlanAndAggPair.value().getGroupByExpressions();
|
||||
if ((queryGroupByExpressions.isEmpty() && !viewGroupByExpressions.isEmpty())
|
||||
|| (!queryGroupByExpressions.isEmpty() && viewGroupByExpressions.isEmpty())) {
|
||||
materializationContext.recordFailReason(queryStructInfo,
|
||||
"only one the of query or view is scalar aggregate and "
|
||||
+ "can not rewrite expression meanwhile",
|
||||
() -> String.format("query aggregate = %s,\n view aggregate = %s,\n",
|
||||
queryTopPlanAndAggPair.value().treeString(),
|
||||
viewTopPlanAndAggPair.value().treeString()));
|
||||
return null;
|
||||
}
|
||||
// try to roll up.
|
||||
// split the query top plan expressions to group expressions and functions, if can not, bail out.
|
||||
Pair<Set<? extends Expression>, Set<? extends Expression>> queryGroupAndFunctionPair
|
||||
@ -147,46 +145,31 @@ public abstract class AbstractMaterializedViewAggregateRule extends AbstractMate
|
||||
// try to rewrite, contains both roll up aggregate functions and aggregate group expression
|
||||
List<NamedExpression> finalOutputExpressions = new ArrayList<>();
|
||||
List<Expression> finalGroupExpressions = new ArrayList<>();
|
||||
List<? extends Expression> queryExpressions = queryTopPlan.getOutput();
|
||||
// permute the mv expr mapping to query based
|
||||
Map<Expression, Expression> mvExprToMvScanExprQueryBased =
|
||||
materializationContext.getShuttledExprToScanExprMapping().keyPermute(viewToQuerySlotMapping)
|
||||
.flattenMap().get(0);
|
||||
for (Expression topExpression : queryExpressions) {
|
||||
// if agg function, try to roll up and rewrite
|
||||
for (Expression topExpression : queryTopPlan.getOutput()) {
|
||||
if (queryTopPlanFunctionSet.contains(topExpression)) {
|
||||
Expression queryFunctionShuttled = ExpressionUtils.shuttleExpressionWithLineage(
|
||||
topExpression,
|
||||
queryTopPlan,
|
||||
queryStructInfo.getTableBitSet());
|
||||
AggregateExpressionRewriteContext context = new AggregateExpressionRewriteContext(
|
||||
false, mvExprToMvScanExprQueryBased, queryTopPlan, queryStructInfo.getTableBitSet());
|
||||
// queryFunctionShuttled maybe sum(column) + count(*), so need to use expression rewriter
|
||||
Expression rollupedExpression = queryFunctionShuttled.accept(AGGREGATE_EXPRESSION_REWRITER,
|
||||
context);
|
||||
if (!context.isValid()) {
|
||||
materializationContext.recordFailReason(queryStructInfo,
|
||||
"Query function roll up fail",
|
||||
() -> String.format("queryFunctionShuttled = %s,\n mvExprToMvScanExprQueryBased = %s",
|
||||
queryFunctionShuttled, mvExprToMvScanExprQueryBased));
|
||||
// if agg function, try to roll up and rewrite
|
||||
Expression rollupedExpression = tryRewriteExpression(queryStructInfo, topExpression,
|
||||
mvExprToMvScanExprQueryBased, false, materializationContext,
|
||||
"Query function roll up fail",
|
||||
() -> String.format("queryExpression = %s,\n mvExprToMvScanExprQueryBased = %s",
|
||||
topExpression, mvExprToMvScanExprQueryBased));
|
||||
if (rollupedExpression == null) {
|
||||
return null;
|
||||
}
|
||||
finalOutputExpressions.add(new Alias(rollupedExpression));
|
||||
} else {
|
||||
// if group by expression, try to rewrite group by expression
|
||||
Expression queryGroupShuttledExpr = ExpressionUtils.shuttleExpressionWithLineage(
|
||||
topExpression, queryTopPlan, queryStructInfo.getTableBitSet());
|
||||
AggregateExpressionRewriteContext context = new AggregateExpressionRewriteContext(true,
|
||||
mvExprToMvScanExprQueryBased, queryTopPlan, queryStructInfo.getTableBitSet());
|
||||
// group by expression maybe group by a + b, so we need expression rewriter
|
||||
Expression rewrittenGroupByExpression = queryGroupShuttledExpr.accept(AGGREGATE_EXPRESSION_REWRITER,
|
||||
context);
|
||||
if (!context.isValid()) {
|
||||
// if group by dimension, try to rewrite
|
||||
Expression rewrittenGroupByExpression = tryRewriteExpression(queryStructInfo, topExpression,
|
||||
mvExprToMvScanExprQueryBased, true, materializationContext,
|
||||
"View dimensions doesn't not cover the query dimensions",
|
||||
() -> String.format("mvExprToMvScanExprQueryBased is %s,\n queryExpression is %s",
|
||||
mvExprToMvScanExprQueryBased, topExpression));
|
||||
if (rewrittenGroupByExpression == null) {
|
||||
// group expr can not rewrite by view
|
||||
materializationContext.recordFailReason(queryStructInfo,
|
||||
"View dimensions doesn't not cover the query dimensions",
|
||||
() -> String.format("mvExprToMvScanExprQueryBased is %s,\n queryGroupShuttledExpr is %s",
|
||||
mvExprToMvScanExprQueryBased, queryGroupShuttledExpr));
|
||||
return null;
|
||||
}
|
||||
NamedExpression groupByExpression = rewrittenGroupByExpression instanceof NamedExpression
|
||||
@ -195,26 +178,19 @@ public abstract class AbstractMaterializedViewAggregateRule extends AbstractMate
|
||||
finalGroupExpressions.add(groupByExpression);
|
||||
}
|
||||
}
|
||||
// add project to guarantee group by column ref is slot reference,
|
||||
// this is necessary because physical createHash will need slotReference later
|
||||
List<Expression> queryGroupByExpressions = queryAggregate.getGroupByExpressions();
|
||||
// handle the scene that query top plan not use the group by in query bottom aggregate
|
||||
if (queryGroupByExpressions.size() != queryTopPlanGroupBySet.size()) {
|
||||
for (Expression expression : queryGroupByExpressions) {
|
||||
if (queryTopPlanGroupBySet.contains(expression)) {
|
||||
continue;
|
||||
}
|
||||
Expression queryGroupShuttledExpr = ExpressionUtils.shuttleExpressionWithLineage(
|
||||
expression, queryTopPlan, queryStructInfo.getTableBitSet());
|
||||
AggregateExpressionRewriteContext context = new AggregateExpressionRewriteContext(true,
|
||||
mvExprToMvScanExprQueryBased, queryTopPlan, queryStructInfo.getTableBitSet());
|
||||
// group by expression maybe group by a + b, so we need expression rewriter
|
||||
Expression rewrittenGroupByExpression = queryGroupShuttledExpr.accept(AGGREGATE_EXPRESSION_REWRITER,
|
||||
context);
|
||||
if (!context.isValid()) {
|
||||
// group expr can not rewrite by view
|
||||
materializationContext.recordFailReason(queryStructInfo,
|
||||
"View dimensions doesn't not cover the query dimensions in bottom agg ",
|
||||
() -> String.format("mvExprToMvScanExprQueryBased is %s,\n queryGroupShuttledExpr is %s",
|
||||
mvExprToMvScanExprQueryBased, queryGroupShuttledExpr));
|
||||
Expression rewrittenGroupByExpression = tryRewriteExpression(queryStructInfo, expression,
|
||||
mvExprToMvScanExprQueryBased, true, materializationContext,
|
||||
"View dimensions doesn't not cover the query dimensions in bottom agg ",
|
||||
() -> String.format("mvExprToMvScanExprQueryBased is %s,\n expression is %s",
|
||||
mvExprToMvScanExprQueryBased, expression));
|
||||
if (rewrittenGroupByExpression == null) {
|
||||
return null;
|
||||
}
|
||||
NamedExpression groupByExpression = rewrittenGroupByExpression instanceof NamedExpression
|
||||
@ -222,31 +198,90 @@ public abstract class AbstractMaterializedViewAggregateRule extends AbstractMate
|
||||
finalGroupExpressions.add(groupByExpression);
|
||||
}
|
||||
}
|
||||
List<Expression> copiedFinalGroupExpressions = new ArrayList<>(finalGroupExpressions);
|
||||
List<NamedExpression> projectsUnderAggregate = copiedFinalGroupExpressions.stream()
|
||||
.map(NamedExpression.class::cast)
|
||||
.collect(Collectors.toList());
|
||||
projectsUnderAggregate.addAll(tempRewritedPlan.getOutput());
|
||||
LogicalProject<Plan> mvProject = new LogicalProject<>(projectsUnderAggregate, tempRewritedPlan);
|
||||
// add agg rewrite
|
||||
Map<ExprId, Slot> projectOutPutExprIdMap = mvProject.getOutput().stream()
|
||||
.distinct()
|
||||
.collect(Collectors.toMap(NamedExpression::getExprId, slot -> slot));
|
||||
// make the expressions to re reference project output
|
||||
finalGroupExpressions = finalGroupExpressions.stream()
|
||||
.map(expr -> {
|
||||
ExprId exprId = ((NamedExpression) expr).getExprId();
|
||||
if (projectOutPutExprIdMap.containsKey(exprId)) {
|
||||
return projectOutPutExprIdMap.get(exprId);
|
||||
if (queryContainsGroupSets) {
|
||||
// construct group sets for repeat
|
||||
List<List<Expression>> rewrittenGroupSetsExpressions = new ArrayList<>();
|
||||
List<List<Expression>> groupingSets = queryAggregate.getSourceRepeat().get().getGroupingSets();
|
||||
for (List<Expression> groupingSet : groupingSets) {
|
||||
if (groupingSet.isEmpty()) {
|
||||
rewrittenGroupSetsExpressions.add(ImmutableList.of());
|
||||
} else {
|
||||
List<Expression> rewrittenGroupSetExpressions = new ArrayList<>();
|
||||
for (Expression expression : groupingSet) {
|
||||
Expression rewrittenGroupByExpression = tryRewriteExpression(queryStructInfo, expression,
|
||||
mvExprToMvScanExprQueryBased, true, materializationContext,
|
||||
"View dimensions doesn't not cover the query group set dimensions",
|
||||
() -> String.format("mvExprToMvScanExprQueryBased is %s,\n queryExpression is %s",
|
||||
mvExprToMvScanExprQueryBased, expression));
|
||||
if (rewrittenGroupByExpression == null) {
|
||||
return null;
|
||||
}
|
||||
rewrittenGroupSetExpressions.add(rewrittenGroupByExpression);
|
||||
}
|
||||
return (NamedExpression) expr;
|
||||
})
|
||||
.collect(Collectors.toList());
|
||||
finalOutputExpressions = finalOutputExpressions.stream()
|
||||
.map(expr -> projectOutPutExprIdMap.containsKey(expr.getExprId())
|
||||
? projectOutPutExprIdMap.get(expr.getExprId()) : expr)
|
||||
.collect(Collectors.toList());
|
||||
return new LogicalAggregate(finalGroupExpressions, finalOutputExpressions, mvProject);
|
||||
rewrittenGroupSetsExpressions.add(rewrittenGroupSetExpressions);
|
||||
}
|
||||
}
|
||||
LogicalRepeat<Plan> repeat = new LogicalRepeat<>(rewrittenGroupSetsExpressions,
|
||||
finalOutputExpressions, tempRewritedPlan);
|
||||
return NormalizeRepeat.doNormalize(repeat);
|
||||
}
|
||||
return new LogicalAggregate<>(finalGroupExpressions, finalOutputExpressions, tempRewritedPlan);
|
||||
}
|
||||
|
||||
/**
|
||||
* Try to rewrite query expression by view, contains both group by dimension and aggregate function
|
||||
*/
|
||||
protected Expression tryRewriteExpression(StructInfo queryStructInfo, Expression queryExpression,
|
||||
Map<Expression, Expression> mvShuttledExprToMvScanExprQueryBased, boolean isGroupBy,
|
||||
MaterializationContext materializationContext, String summaryIfFail, Supplier<String> detailIfFail) {
|
||||
Expression queryFunctionShuttled = ExpressionUtils.shuttleExpressionWithLineage(
|
||||
queryExpression,
|
||||
queryStructInfo.getTopPlan(),
|
||||
queryStructInfo.getTableBitSet());
|
||||
AggregateExpressionRewriteContext expressionRewriteContext = new AggregateExpressionRewriteContext(
|
||||
isGroupBy, mvShuttledExprToMvScanExprQueryBased, queryStructInfo.getTopPlan(),
|
||||
queryStructInfo.getTableBitSet());
|
||||
Expression rewrittenExpression = queryFunctionShuttled.accept(AGGREGATE_EXPRESSION_REWRITER,
|
||||
expressionRewriteContext);
|
||||
if (!expressionRewriteContext.isValid()) {
|
||||
materializationContext.recordFailReason(queryStructInfo, summaryIfFail, detailIfFail);
|
||||
return null;
|
||||
}
|
||||
return rewrittenExpression;
|
||||
}
|
||||
|
||||
/**
|
||||
* Check query and view aggregate compatibility
|
||||
*/
|
||||
private static boolean checkCompatibility(
|
||||
StructInfo queryStructInfo,
|
||||
LogicalAggregate<Plan> queryAggregate, LogicalAggregate<Plan> viewAggregate,
|
||||
MaterializationContext materializationContext) {
|
||||
// if view is scalar aggregate but query is not. Or if query is scalar aggregate but view is not
|
||||
// Should not rewrite
|
||||
List<Expression> queryGroupByExpressions = queryAggregate.getGroupByExpressions();
|
||||
List<Expression> viewGroupByExpressions = viewAggregate.getGroupByExpressions();
|
||||
if (!queryGroupByExpressions.isEmpty() && viewGroupByExpressions.isEmpty()) {
|
||||
materializationContext.recordFailReason(queryStructInfo,
|
||||
"only one the of query or view is scalar aggregate and "
|
||||
+ "can not rewrite expression meanwhile",
|
||||
() -> String.format("query aggregate = %s,\n view aggregate = %s,\n",
|
||||
queryAggregate.treeString(),
|
||||
viewAggregate.treeString()));
|
||||
return false;
|
||||
}
|
||||
boolean viewHasGroupSets = viewAggregate.getSourceRepeat()
|
||||
.map(repeat -> repeat.getGroupingSets().size()).orElse(0) > 0;
|
||||
// if both query and view has group sets, or query doesn't hava, mv have, not supported
|
||||
if (viewHasGroupSets) {
|
||||
materializationContext.recordFailReason(queryStructInfo,
|
||||
"both query and view have group sets, or query doesn't have but view have, not supported",
|
||||
() -> String.format("query aggregate = %s,\n view aggregate = %s,\n",
|
||||
queryAggregate.treeString(),
|
||||
viewAggregate.treeString()));
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
private boolean isGroupByEquals(Pair<Plan, LogicalAggregate<Plan>> queryTopPlanAndAggPair,
|
||||
@ -259,14 +294,18 @@ public abstract class AbstractMaterializedViewAggregateRule extends AbstractMate
|
||||
Plan viewTopPlan = viewTopPlanAndAggPair.key();
|
||||
LogicalAggregate<Plan> queryAggregate = queryTopPlanAndAggPair.value();
|
||||
LogicalAggregate<Plan> viewAggregate = viewTopPlanAndAggPair.value();
|
||||
Set<? extends Expression> queryGroupShuttledExpression = new HashSet<>(
|
||||
ExpressionUtils.shuttleExpressionWithLineage(
|
||||
queryAggregate.getGroupByExpressions(), queryTopPlan, queryStructInfo.getTableBitSet()));
|
||||
Set<? extends Expression> viewGroupShuttledExpressionQueryBased = ExpressionUtils.shuttleExpressionWithLineage(
|
||||
viewAggregate.getGroupByExpressions(), viewTopPlan, viewStructInfo.getTableBitSet())
|
||||
.stream()
|
||||
.map(expr -> ExpressionUtils.replace(expr, viewToQuerySlotMapping.toSlotReferenceMap()))
|
||||
.collect(Collectors.toSet());
|
||||
|
||||
Set<Expression> queryGroupShuttledExpression = new HashSet<>();
|
||||
for (Expression queryExpression : ExpressionUtils.shuttleExpressionWithLineage(
|
||||
queryAggregate.getGroupByExpressions(), queryTopPlan, queryStructInfo.getTableBitSet())) {
|
||||
queryGroupShuttledExpression.add(queryExpression);
|
||||
}
|
||||
Set<Expression> viewGroupShuttledExpressionQueryBased = new HashSet<>();
|
||||
for (Expression viewExpression : ExpressionUtils.shuttleExpressionWithLineage(
|
||||
viewAggregate.getGroupByExpressions(), viewTopPlan, viewStructInfo.getTableBitSet())) {
|
||||
viewGroupShuttledExpressionQueryBased.add(
|
||||
ExpressionUtils.replace(viewExpression, viewToQuerySlotMapping.toSlotReferenceMap()));
|
||||
}
|
||||
return queryGroupShuttledExpression.equals(viewGroupShuttledExpressionQueryBased);
|
||||
}
|
||||
|
||||
@ -305,23 +344,6 @@ public abstract class AbstractMaterializedViewAggregateRule extends AbstractMate
|
||||
return null;
|
||||
}
|
||||
|
||||
// Check the aggregate function can roll up or not, return true if could roll up
|
||||
// if view aggregate function is distinct or is in the un supported rollup functions, it doesn't support
|
||||
// roll up.
|
||||
private static boolean canRollup(Expression rollupExpression) {
|
||||
if (rollupExpression == null) {
|
||||
return false;
|
||||
}
|
||||
if (rollupExpression instanceof Function && !(rollupExpression instanceof AggregateFunction)) {
|
||||
return false;
|
||||
}
|
||||
if (rollupExpression instanceof AggregateFunction) {
|
||||
AggregateFunction aggregateFunction = (AggregateFunction) rollupExpression;
|
||||
return !aggregateFunction.isDistinct() && aggregateFunction instanceof RollUpTrait;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
private Pair<Set<? extends Expression>, Set<? extends Expression>> topPlanSplitToGroupAndFunction(
|
||||
Pair<Plan, LogicalAggregate<Plan>> topPlanAndAggPair, StructInfo queryStructInfo) {
|
||||
LogicalAggregate<Plan> bottomQueryAggregate = topPlanAndAggPair.value();
|
||||
@ -405,11 +427,40 @@ public abstract class AbstractMaterializedViewAggregateRule extends AbstractMate
|
||||
return rollupAggregateFunction;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Expression visitGroupingScalarFunction(GroupingScalarFunction groupingScalarFunction,
|
||||
AggregateExpressionRewriteContext context) {
|
||||
List<Expression> children = groupingScalarFunction.children();
|
||||
List<Expression> rewrittenChildren = new ArrayList<>();
|
||||
for (Expression child : children) {
|
||||
Expression rewrittenChild = child.accept(this, context);
|
||||
if (!context.isValid()) {
|
||||
return groupingScalarFunction;
|
||||
}
|
||||
rewrittenChildren.add(rewrittenChild);
|
||||
}
|
||||
return groupingScalarFunction.withChildren(rewrittenChildren);
|
||||
}
|
||||
|
||||
@Override
|
||||
public Expression visitSlot(Slot slot, AggregateExpressionRewriteContext rewriteContext) {
|
||||
if (!rewriteContext.isValid()) {
|
||||
return slot;
|
||||
}
|
||||
if (slot instanceof VirtualSlotReference) {
|
||||
Optional<GroupingScalarFunction> originExpression = ((VirtualSlotReference) slot).getOriginExpression();
|
||||
if (!originExpression.isPresent()) {
|
||||
return Repeat.generateVirtualGroupingIdSlot();
|
||||
} else {
|
||||
GroupingScalarFunction groupingScalarFunction = originExpression.get();
|
||||
groupingScalarFunction =
|
||||
(GroupingScalarFunction) groupingScalarFunction.accept(this, rewriteContext);
|
||||
if (!rewriteContext.isValid()) {
|
||||
return slot;
|
||||
}
|
||||
return Repeat.generateVirtualSlotByFunction(groupingScalarFunction);
|
||||
}
|
||||
}
|
||||
if (rewriteContext.getMvExprToMvScanExprQueryBasedMapping().containsKey(slot)) {
|
||||
return rewriteContext.getMvExprToMvScanExprQueryBasedMapping().get(slot);
|
||||
}
|
||||
|
||||
@ -43,6 +43,7 @@ import org.apache.doris.nereids.trees.expressions.NamedExpression;
|
||||
import org.apache.doris.nereids.trees.expressions.Not;
|
||||
import org.apache.doris.nereids.trees.expressions.Slot;
|
||||
import org.apache.doris.nereids.trees.expressions.SlotReference;
|
||||
import org.apache.doris.nereids.trees.expressions.VirtualSlotReference;
|
||||
import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
|
||||
import org.apache.doris.nereids.trees.expressions.functions.scalar.NonNullable;
|
||||
import org.apache.doris.nereids.trees.expressions.functions.scalar.Nullable;
|
||||
@ -306,7 +307,18 @@ public abstract class AbstractMaterializedViewRule implements ExplorationRuleFac
|
||||
}
|
||||
}
|
||||
}
|
||||
List<Slot> rewrittenPlanOutput = rewrittenPlan.getOutput();
|
||||
rewrittenPlan = normalizeExpressions(rewrittenPlan, queryPlan);
|
||||
if (rewrittenPlan == null) {
|
||||
// maybe virtual slot reference added automatically
|
||||
materializationContext.recordFailReason(queryStructInfo,
|
||||
"RewrittenPlan output logical properties is different with target group",
|
||||
() -> String.format("materialized view rule normalizeExpressions, output size between "
|
||||
+ "origin and rewritten plan is different, rewritten output is %s, "
|
||||
+ "origin output is %s",
|
||||
rewrittenPlanOutput, queryPlan.getOutput()));
|
||||
continue;
|
||||
}
|
||||
if (!isOutputValid(queryPlan, rewrittenPlan)) {
|
||||
LogicalProperties logicalProperties = rewrittenPlan.getLogicalProperties();
|
||||
materializationContext.recordFailReason(queryStructInfo,
|
||||
@ -343,6 +355,9 @@ public abstract class AbstractMaterializedViewRule implements ExplorationRuleFac
|
||||
|
||||
// Normalize expression such as nullable property and output slot id
|
||||
protected Plan normalizeExpressions(Plan rewrittenPlan, Plan originPlan) {
|
||||
if (rewrittenPlan.getOutput().size() != originPlan.getOutput().size()) {
|
||||
return null;
|
||||
}
|
||||
// normalize nullable
|
||||
List<NamedExpression> normalizeProjects = new ArrayList<>();
|
||||
for (int i = 0; i < originPlan.getOutput().size(); i++) {
|
||||
@ -693,6 +708,11 @@ public abstract class AbstractMaterializedViewRule implements ExplorationRuleFac
|
||||
&& context.alreadyRewrite(plan.getGroupExpression().get().getOwnerGroup().getGroupId());
|
||||
}
|
||||
|
||||
protected boolean isEmptyVirtualSlot(Expression expression) {
|
||||
return expression instanceof VirtualSlotReference
|
||||
&& ((VirtualSlotReference) expression).getRealExpressions().isEmpty();
|
||||
}
|
||||
|
||||
// check mv plan is valid or not, this can use cache for performance
|
||||
private boolean isMaterializationValid(CascadesContext cascadesContext, MaterializationContext context) {
|
||||
long materializationId = context.getMaterializationQualifier().hashCode();
|
||||
|
||||
@ -204,7 +204,7 @@ public class MaterializedViewUtils {
|
||||
* should be different
|
||||
*/
|
||||
public static Plan generateMvScanPlan(MTMV materializedView, CascadesContext cascadesContext) {
|
||||
LogicalOlapScan mvScan = new LogicalOlapScan(
|
||||
return new LogicalOlapScan(
|
||||
cascadesContext.getStatementContext().getNextRelationId(),
|
||||
materializedView,
|
||||
materializedView.getFullQualifiers(),
|
||||
@ -216,9 +216,6 @@ public class MaterializedViewUtils {
|
||||
// this must be empty, or it will be used to sample
|
||||
ImmutableList.of(),
|
||||
Optional.empty());
|
||||
List<NamedExpression> mvProjects = mvScan.getOutput().stream().map(NamedExpression.class::cast)
|
||||
.collect(Collectors.toList());
|
||||
return new LogicalProject<Plan>(mvProjects, mvScan);
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@ -54,6 +54,7 @@ import org.apache.doris.nereids.trees.plans.logical.LogicalJoin;
|
||||
import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
|
||||
import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
|
||||
import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
|
||||
import org.apache.doris.nereids.trees.plans.logical.LogicalRepeat;
|
||||
import org.apache.doris.nereids.trees.plans.logical.LogicalSort;
|
||||
import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
|
||||
import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanVisitor;
|
||||
@ -607,9 +608,6 @@ public class StructInfo {
|
||||
checkContext.setContainsTopAggregate(true);
|
||||
checkContext.plusTopAggregateNum();
|
||||
}
|
||||
if (aggregate.getSourceRepeat().isPresent()) {
|
||||
return false;
|
||||
}
|
||||
return visit(aggregate, checkContext);
|
||||
}
|
||||
|
||||
@ -627,7 +625,8 @@ public class StructInfo {
|
||||
|| plan instanceof Join
|
||||
|| plan instanceof LogicalSort
|
||||
|| plan instanceof LogicalAggregate
|
||||
|| plan instanceof GroupPlan) {
|
||||
|| plan instanceof GroupPlan
|
||||
|| plan instanceof LogicalRepeat) {
|
||||
return doVisit(plan, checkContext);
|
||||
}
|
||||
return false;
|
||||
@ -660,7 +659,8 @@ public class StructInfo {
|
||||
if (plan instanceof Filter
|
||||
|| plan instanceof Project
|
||||
|| plan instanceof CatalogRelation
|
||||
|| plan instanceof GroupPlan) {
|
||||
|| plan instanceof GroupPlan
|
||||
|| plan instanceof LogicalRepeat) {
|
||||
return doVisit(plan, checkContext);
|
||||
}
|
||||
return false;
|
||||
|
||||
Reference in New Issue
Block a user