Skip to content

Commit

Permalink
refactor analyze view
Browse files Browse the repository at this point in the history
  • Loading branch information
@924060929
924060929 committed Jul 1, 2024
1 parent 86defed commit baa0524
Show file tree
Hide file tree
Showing 8 changed files with 158 additions and 124 deletions.
12 changes: 2 additions & 10 deletions fe/fe-core/src/main/java/org/apache/doris/nereids/CascadesContext.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -240,19 +240,11 @@ public void toMemo() {
}

public Analyzer newAnalyzer() {
return newAnalyzer(false);
}

public Analyzer newAnalyzer(boolean analyzeView) {
return new Analyzer(this, analyzeView);
}

public Analyzer newAnalyzer(boolean analyzeView, Optional<CustomTableResolver> customTableResolver) {
return new Analyzer(this, analyzeView, customTableResolver);
return newAnalyzer(Optional.empty());
}

public Analyzer newAnalyzer(Optional<CustomTableResolver> customTableResolver) {
return newAnalyzer(false, customTableResolver);
return new Analyzer(this, customTableResolver);
}

@Override
Expand Down
39 changes: 37 additions & 2 deletions ...e-core/src/main/java/org/apache/doris/nereids/jobs/executor/AbstractBatchJobExecutor.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -29,13 +29,16 @@
import org.apache.doris.nereids.rules.Rule;
import org.apache.doris.nereids.rules.RuleFactory;
import org.apache.doris.nereids.rules.RuleType;
import org.apache.doris.nereids.trees.plans.Plan;
import org.apache.doris.nereids.trees.plans.visitor.CustomRewriter;

import com.google.common.collect.ImmutableList;

import java.util.Arrays;
import java.util.List;
import java.util.Objects;
import java.util.Set;
import java.util.function.Predicate;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import java.util.stream.Stream;
Expand All @@ -46,6 +49,8 @@
* Each batch of rules will be uniformly executed.
*/
public abstract class AbstractBatchJobExecutor {
private static final ThreadLocal<Set<Class<Plan>>> NOT_TRAVERSE_CHILDREN = new ThreadLocal();
private static final Predicate<Plan> TRAVERSE_ALL_PLANS = plan -> true;

protected CascadesContext cascadesContext;

Expand All @@ -65,6 +70,16 @@ public static List<RewriteJob> jobs(RewriteJob... jobs) {
).collect(ImmutableList.toImmutableList());
}

public static List<RewriteJob> notTraverseChildrenOf(
Set<Class<? extends Plan>> notTraverseClasses, Supplier<List<RewriteJob>> jobs) {
try {
NOT_TRAVERSE_CHILDREN.set((Set) notTraverseClasses);
return jobs.get();
} finally {
NOT_TRAVERSE_CHILDREN.remove();
}
}

public static TopicRewriteJob topic(String topicName, RewriteJob... jobs) {
return new TopicRewriteJob(topicName, Arrays.asList(jobs));
}
Expand All @@ -82,7 +97,7 @@ public static RewriteJob bottomUp(List<RuleFactory> ruleFactories) {
.map(RuleFactory::buildRules)
.flatMap(List::stream)
.collect(ImmutableList.toImmutableList());
return new RootPlanTreeRewriteJob(rules, PlanTreeRewriteBottomUpJob::new, true);
return new RootPlanTreeRewriteJob(rules, PlanTreeRewriteBottomUpJob::new, getTraversePredicate(), true);
}

public static RewriteJob topDown(RuleFactory... ruleFactories) {
Expand All @@ -98,7 +113,7 @@ public static RewriteJob topDown(List<RuleFactory> ruleFactories, boolean once)
.map(RuleFactory::buildRules)
.flatMap(List::stream)
.collect(ImmutableList.toImmutableList());
return new RootPlanTreeRewriteJob(rules, PlanTreeRewriteTopDownJob::new, once);
return new RootPlanTreeRewriteJob(rules, PlanTreeRewriteTopDownJob::new, getTraversePredicate(), once);
}

public static RewriteJob custom(RuleType ruleType, Supplier<CustomRewriter> planRewriter) {
Expand Down Expand Up @@ -126,4 +141,24 @@ public void execute() {
}

public abstract List<RewriteJob> getJobs();

private static Predicate<Plan> getTraversePredicate() {
Set<Class<Plan>> notTraverseChildren = NOT_TRAVERSE_CHILDREN.get();
return notTraverseChildren == null
? TRAVERSE_ALL_PLANS
: new NotTraverseChildren(notTraverseChildren);
}

private static class NotTraverseChildren implements Predicate<Plan> {
private final Set<Class<Plan>> notTraverseChildren;

public NotTraverseChildren(Set<Class<Plan>> notTraverseChildren) {
this.notTraverseChildren = Objects.requireNonNull(notTraverseChildren, "notTraversePlans can not be null");
}

@Override
public boolean test(Plan plan) {
return !notTraverseChildren.contains(plan.getClass());
}
}
}
165 changes: 73 additions & 92 deletions fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/executor/Analyzer.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,9 @@
import org.apache.doris.nereids.rules.rewrite.MergeProjects;
import org.apache.doris.nereids.rules.rewrite.SemiJoinCommute;
import org.apache.doris.nereids.rules.rewrite.SimplifyAggGroupBy;
import org.apache.doris.nereids.trees.plans.logical.LogicalView;

import com.google.common.collect.ImmutableSet;

import java.util.List;
import java.util.Objects;
Expand All @@ -59,8 +62,7 @@
*/
public class Analyzer extends AbstractBatchJobExecutor {

public static final List<RewriteJob> DEFAULT_ANALYZE_JOBS = buildAnalyzeJobs(Optional.empty());
public static final List<RewriteJob> DEFAULT_ANALYZE_VIEW_JOBS = buildAnalyzeViewJobs(Optional.empty());
public static final List<RewriteJob> ANALYZE_JOBS = buildAnalyzeJobs(Optional.empty());

private final List<RewriteJob> jobs;

Expand All @@ -69,36 +71,23 @@ public class Analyzer extends AbstractBatchJobExecutor {
* @param cascadesContext planner context for execute job
*/
public Analyzer(CascadesContext cascadesContext) {
this(cascadesContext, false);
}

public Analyzer(CascadesContext cascadesContext, boolean analyzeView) {
this(cascadesContext, analyzeView, Optional.empty());
this(cascadesContext, Optional.empty());
}

/**
* constructor of Analyzer. For view, we only do bind relation since other analyze step will do by outer Analyzer.
*
* @param cascadesContext current context for analyzer
* @param analyzeView analyze view or user sql. If true, analyzer is used for view.
* @param customTableResolver custom resolver for outer catalog.
*/
public Analyzer(CascadesContext cascadesContext, boolean analyzeView,
Optional<CustomTableResolver> customTableResolver) {
public Analyzer(CascadesContext cascadesContext, Optional<CustomTableResolver> customTableResolver) {
super(cascadesContext);
Objects.requireNonNull(customTableResolver, "customTableResolver cannot be null");
if (analyzeView) {
if (customTableResolver.isPresent()) {
this.jobs = buildAnalyzeViewJobs(customTableResolver);
} else {
this.jobs = DEFAULT_ANALYZE_VIEW_JOBS;
}

if (customTableResolver.isPresent()) {
this.jobs = buildAnalyzeJobs(customTableResolver);
} else {
if (customTableResolver.isPresent()) {
this.jobs = buildAnalyzeJobs(customTableResolver);
} else {
this.jobs = DEFAULT_ANALYZE_JOBS;
}
this.jobs = ANALYZE_JOBS;
}
}

Expand All @@ -114,79 +103,71 @@ public void analyze() {
execute();
}

private static List<RewriteJob> buildAnalyzeViewJobs(Optional<CustomTableResolver> customTableResolver) {
return jobs(
topDown(new AnalyzeCTE()),
topDown(new EliminateLogicalSelectHint()),
bottomUp(
new BindRelation(customTableResolver),
new CheckPolicy()
)
);
}

private static List<RewriteJob> buildAnalyzeJobs(Optional<CustomTableResolver> customTableResolver) {
return jobs(
// we should eliminate hint before "Subquery unnesting".
topDown(new AnalyzeCTE()),
topDown(new EliminateLogicalSelectHint()),
bottomUp(
new BindRelation(customTableResolver),
new CheckPolicy()
),
bottomUp(new BindExpression()),
topDown(new BindSink()),
bottomUp(new CheckAfterBind()),
bottomUp(
new ProjectToGlobalAggregate(),
// this rule check's the logicalProject node's isDistinct property
// and replace the logicalProject node with a LogicalAggregate node
// so any rule before this, if create a new logicalProject node
// should make sure isDistinct property is correctly passed around.
// please see rule BindSlotReference or BindFunction for example
new EliminateDistinctConstant(),
new ProjectWithDistinctToAggregate(),
new ReplaceExpressionByChildOutput(),
new OneRowRelationExtractAggregate()
),
topDown(
new FillUpMissingSlots(),
// We should use NormalizeRepeat to compute nullable properties for LogicalRepeat in the analysis
// stage. NormalizeRepeat will compute nullable property, add virtual slot, LogicalAggregate and
// LogicalProject for normalize. This rule depends on FillUpMissingSlots to fill up slots.
new NormalizeRepeat()
),
bottomUp(new AdjustAggregateNullableForEmptySet()),
// consider sql with user defined var @t_zone
// set @t_zone='GMT';
// SELECT
// DATE_FORMAT(convert_tz(dt, time_zone, @t_zone),'%Y-%m-%d') day
// FROM
// t
// GROUP BY
// 1;
// @t_zone must be replaced as 'GMT' before EliminateGroupByConstant and NormalizeAggregate rule.
// So need run VariableToLiteral rule before the two rules.
topDown(new VariableToLiteral()),
// run CheckAnalysis before EliminateGroupByConstant in order to report error message correctly like bellow
// select SUM(lo_tax) FROM lineorder group by 1;
// errCode = 2, detailMessage = GROUP BY expression must not contain aggregate functions: sum(lo_tax)
bottomUp(new CheckAnalysis()),
topDown(new EliminateGroupByConstant()),
return notTraverseChildrenOf(
ImmutableSet.of(LogicalView.class),
() -> jobs(
// we should eliminate hint before "Subquery unnesting".
topDown(new AnalyzeCTE()),
topDown(new EliminateLogicalSelectHint()),
bottomUp(
new BindRelation(customTableResolver),
new CheckPolicy()
),
bottomUp(new BindExpression()),
topDown(new BindSink()),
bottomUp(new CheckAfterBind()),
bottomUp(
new ProjectToGlobalAggregate(),
// this rule check's the logicalProject node's isDistinct property
// and replace the logicalProject node with a LogicalAggregate node
// so any rule before this, if create a new logicalProject node
// should make sure isDistinct property is correctly passed around.
// please see rule BindSlotReference or BindFunction for example
new EliminateDistinctConstant(),
new ProjectWithDistinctToAggregate(),
new ReplaceExpressionByChildOutput(),
new OneRowRelationExtractAggregate()
),
topDown(
new FillUpMissingSlots(),
// We should use NormalizeRepeat to compute nullable properties for LogicalRepeat in the analysis
// stage. NormalizeRepeat will compute nullable property, add virtual slot, LogicalAggregate and
// LogicalProject for normalize. This rule depends on FillUpMissingSlots to fill up slots.
new NormalizeRepeat()
),
bottomUp(new AdjustAggregateNullableForEmptySet()),
// consider sql with user defined var @t_zone
// set @t_zone='GMT';
// SELECT
// DATE_FORMAT(convert_tz(dt, time_zone, @t_zone),'%Y-%m-%d') day
// FROM
// t
// GROUP BY
// 1;
// @t_zone must be replaced as 'GMT' before EliminateGroupByConstant and NormalizeAggregate rule.
// So need run VariableToLiteral rule before the two rules.
topDown(new VariableToLiteral()),
// run CheckAnalysis before EliminateGroupByConstant in order to report error message correctly like bellow
// select SUM(lo_tax) FROM lineorder group by 1;
// errCode = 2, detailMessage = GROUP BY expression must not contain aggregate functions: sum(lo_tax)
bottomUp(new CheckAnalysis()),
topDown(new EliminateGroupByConstant()),

topDown(new SimplifyAggGroupBy()),
// run BuildAggForRandomDistributedTable before NormalizeAggregate in order to optimize the agg plan
topDown(new BuildAggForRandomDistributedTable()),
topDown(new NormalizeAggregate()),
topDown(new HavingToFilter()),
bottomUp(new SemiJoinCommute()),
bottomUp(
new CollectSubQueryAlias(),
new CollectJoinConstraint()
),
topDown(new LeadingJoin()),
bottomUp(new SubqueryToApply()),
topDown(new MergeProjects())
topDown(new SimplifyAggGroupBy()),
// run BuildAggForRandomDistributedTable before NormalizeAggregate in order to optimize the agg plan
topDown(new BuildAggForRandomDistributedTable()),
topDown(new NormalizeAggregate()),
topDown(new HavingToFilter()),
bottomUp(new SemiJoinCommute()),
bottomUp(
new CollectSubQueryAlias(),
new CollectJoinConstraint()
),
topDown(new LeadingJoin()),
bottomUp(new SubqueryToApply()),
topDown(new MergeProjects())
)
);
}
}
22 changes: 14 additions & 8 deletions ...-core/src/main/java/org/apache/doris/nereids/jobs/rewrite/PlanTreeRewriteBottomUpJob.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.function.Predicate;

/**
* PlanTreeRewriteBottomUpJob
Expand Down Expand Up @@ -55,8 +56,10 @@ enum RewriteState {
ENSURE_CHILDREN_REWRITTEN
}

public PlanTreeRewriteBottomUpJob(RewriteJobContext rewriteJobContext, JobContext context, List<Rule> rules) {
super(JobType.BOTTOM_UP_REWRITE, context);
public PlanTreeRewriteBottomUpJob(
RewriteJobContext rewriteJobContext, JobContext context,
Predicate<Plan> isTraverseChildren, List<Rule> rules) {
super(JobType.BOTTOM_UP_REWRITE, context, isTraverseChildren);
this.rewriteJobContext = Objects.requireNonNull(rewriteJobContext, "rewriteContext cannot be null");
this.rules = Objects.requireNonNull(rules, "rules cannot be null");
this.batchId = rewriteJobContext.batchId;
Expand Down Expand Up @@ -97,7 +100,7 @@ private void rewriteThis() {
return;
}
// After the rewrite take effect, we should handle the children part again.
pushJob(new PlanTreeRewriteBottomUpJob(newJobContext, context, rules));
pushJob(new PlanTreeRewriteBottomUpJob(newJobContext, context, isTraverseChildren, rules));
setState(rewriteResult.plan, RewriteState.ENSURE_CHILDREN_REWRITTEN, batchId);
} else {
// No new plan is generated, so just set the state of the current plan to 'REWRITTEN'.
Expand All @@ -110,7 +113,7 @@ private void ensureChildrenRewritten() {
Plan plan = rewriteJobContext.plan;
int batchId = rewriteJobContext.batchId;
setState(plan, RewriteState.REWRITE_THIS, batchId);
pushJob(new PlanTreeRewriteBottomUpJob(rewriteJobContext, context, rules));
pushJob(new PlanTreeRewriteBottomUpJob(rewriteJobContext, context, isTraverseChildren, rules));

// some rule return new plan tree, which the number of new plan node > 1,
// we should transform this new plan nodes too.
Expand All @@ -122,31 +125,34 @@ private void ensureChildrenRewritten() {

private void pushChildrenJobs(Plan plan) {
List<Plan> children = plan.children();
if (!isTraverseChildren.test(plan)) {
return;
}
switch (children.size()) {
case 0: return;
case 1:
Plan child = children.get(0);
RewriteJobContext childRewriteJobContext = new RewriteJobContext(
child, rewriteJobContext, 0, false, batchId);
pushJob(new PlanTreeRewriteBottomUpJob(childRewriteJobContext, context, rules));
pushJob(new PlanTreeRewriteBottomUpJob(childRewriteJobContext, context, isTraverseChildren, rules));
return;
case 2:
Plan right = children.get(1);
RewriteJobContext rightRewriteJobContext = new RewriteJobContext(
right, rewriteJobContext, 1, false, batchId);
pushJob(new PlanTreeRewriteBottomUpJob(rightRewriteJobContext, context, rules));
pushJob(new PlanTreeRewriteBottomUpJob(rightRewriteJobContext, context, isTraverseChildren, rules));

Plan left = children.get(0);
RewriteJobContext leftRewriteJobContext = new RewriteJobContext(
left, rewriteJobContext, 0, false, batchId);
pushJob(new PlanTreeRewriteBottomUpJob(leftRewriteJobContext, context, rules));
pushJob(new PlanTreeRewriteBottomUpJob(leftRewriteJobContext, context, isTraverseChildren, rules));
return;
default:
for (int i = children.size() - 1; i >= 0; i--) {
child = children.get(i);
childRewriteJobContext = new RewriteJobContext(
child, rewriteJobContext, i, false, batchId);
pushJob(new PlanTreeRewriteBottomUpJob(childRewriteJobContext, context, rules));
pushJob(new PlanTreeRewriteBottomUpJob(childRewriteJobContext, context, isTraverseChildren, rules));
}
}
}
Expand Down
Loading

0 comments on commit baa0524

Please sign in to comment.