Skip to content

Commit

Permalink
demo
Browse files Browse the repository at this point in the history
  • Loading branch information
@englefly
englefly committed Dec 26, 2024
1 parent 33b1f69 commit 486c9a4
Show file tree
Hide file tree
Showing 28 changed files with 1,666 additions and 27 deletions.
13 changes: 12 additions & 1 deletion 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 @@ -53,6 +53,7 @@
import org.apache.doris.statistics.Statistics;
import org.apache.doris.statistics.StatisticsBuilder;

import com.clearspring.analytics.util.Lists;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Maps;
import com.google.common.collect.Multimap;
Expand Down Expand Up @@ -80,7 +81,7 @@ public class CascadesContext implements ScheduleContext {
private static final Logger LOG = LogManager.getLogger(CascadesContext.class);

// in analyze/rewrite stage, the plan will storage in this field
private Plan plan;
public Plan plan;
private Optional<RootRewriteJobContext> currentRootRewriteJobContext;
// in optimize stage, the plan will storage in the memo
private Memo memo;
Expand Down Expand Up @@ -122,6 +123,8 @@ public class CascadesContext implements ScheduleContext {
private int distinctAggLevel;
private final boolean isEnableExprTrace;

public Plan cteplan;
public Map<Plan, Memo> memos = new HashMap<>();
/**
* Constructor of OptimizerContext.
*
Expand Down Expand Up @@ -222,6 +225,14 @@ public synchronized boolean isTimeout() {

public void toMemo() {
this.memo = new Memo(getConnectContext(), plan);
// if (cteplan != null) {
// List<Plan> plans = new ArrayList<>();
// plans.add(cteplan);
// plans.add(plan);
// this.memo = new Memo(getConnectContext(), plans);
// } else {
// this.memo = new Memo(getConnectContext(), plan);
// }
}

public TableCollector newTableCollector() {
Expand Down
29 changes: 26 additions & 3 deletions fe/fe-core/src/main/java/org/apache/doris/nereids/NereidsPlanner.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,6 @@
import org.apache.doris.common.profile.SummaryProfile;
import org.apache.doris.common.util.DebugUtil;
import org.apache.doris.mysql.FieldInfo;
import org.apache.doris.nereids.CascadesContext.Lock;
import org.apache.doris.nereids.commonCTE.CteExtractor;
import org.apache.doris.nereids.exceptions.AnalysisException;
import org.apache.doris.nereids.glue.LogicalPlanAdapter;
Expand Down Expand Up @@ -119,6 +118,8 @@ public class NereidsPlanner extends Planner {
private double cost = 0;
private LogicalPlanAdapter logicalPlanAdapter;

private CteExtractor cteExtractor;

public NereidsPlanner(StatementContext statementContext) {
this.statementContext = statementContext;
}
Expand Down Expand Up @@ -269,6 +270,7 @@ private Plan planWithoutLock(
}
}
extractCommonCTE();

optimize();
// print memo before choose plan.
// if chooseNthPlan failed, we could get memo to debug
Expand All @@ -278,6 +280,17 @@ private Plan planWithoutLock(
}
int nth = cascadesContext.getConnectContext().getSessionVariable().getNthOptimizedPlan();
PhysicalPlan physicalPlan = chooseNthPlan(getRoot(), requireProperties, nth);
double cost1 = cost;


cascadesContext.plan = cascadesContext.cteplan;
optimize();
PhysicalPlan physicalPlan2 = chooseNthPlan(getRoot(), requireProperties, nth);
double cost2 = cost;

if (cost2 < cost1) {
physicalPlan = physicalPlan2;
}

physicalPlan = postProcess(physicalPlan);
if (cascadesContext.getConnectContext().getSessionVariable().dumpNereidsMemo) {
Expand All @@ -301,8 +314,17 @@ protected LogicalPlan preprocess(LogicalPlan logicalPlan) {
}

private void extractCommonCTE() {
CteExtractor commonCTE = new CteExtractor((AbstractLogicalPlan) cascadesContext.getRewritePlan());
commonCTE.execute();
if (ConnectContext.get() != null && ConnectContext.get().getSessionVariable().enableCommonCte) {
CteExtractor commonCTE = new CteExtractor(
(AbstractLogicalPlan) cascadesContext.getRewritePlan(),
cascadesContext);
List<Plan> plans = commonCTE.execute();
//cascadesContext.setRewritePlan(plans.get(0));
if (!plans.isEmpty()) {
cascadesContext.cteplan = plans.get(0);
}
cteExtractor = commonCTE;
}
}

/**
Expand Down Expand Up @@ -682,6 +704,7 @@ public String getExplainString(ExplainOptions explainOptions) {
break;
case REWRITTEN_PLAN:
plan = rewrittenPlan.treeString();
plan += "\n common cte info \n" + cteExtractor;
break;
case OPTIMIZED_PLAN:
plan = "cost = " + cost + "\n" + optimizedPlan.treeString() + mvSummary;
Expand Down
37 changes: 37 additions & 0 deletions fe/fe-core/src/main/java/org/apache/doris/nereids/commonCTE/ConsumerRewriter.java
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,37 @@
package org.apache.doris.nereids.commonCTE;

import org.apache.doris.nereids.trees.expressions.CTEId;
import org.apache.doris.nereids.trees.expressions.Slot;
import org.apache.doris.nereids.trees.expressions.StatementScopeIdGenerator;
import org.apache.doris.nereids.trees.plans.Plan;
import org.apache.doris.nereids.trees.plans.logical.AbstractLogicalPlan;
import org.apache.doris.nereids.trees.plans.logical.LogicalCTEConsumer;
import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;

import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableMultimap;

import java.util.ArrayList;
import java.util.List;

public class ConsumerRewriter extends DefaultPlanRewriter<ConsumerRewriterContext> {
public static final ConsumerRewriter INSTANCE = new ConsumerRewriter();
@Override
public Plan visit(Plan plan, ConsumerRewriterContext context) {
if (context.isTarget(plan)) {
context.producer.getOutput();
plan.getOutput();
ImmutableMap.Builder<Slot, Slot> cToPBuilder = ImmutableMap.builder();
ImmutableMultimap.Builder<Slot, Slot> pToCBuilder = ImmutableMultimap.builder();
for (int i = 0; i< context.producer.getOutput().size(); i++) {
cToPBuilder.put(plan.getOutput().get(i), context.producer.getOutput().get(i));
pToCBuilder.put(context.producer.getOutput().get(i), plan.getOutput().get(i));
}
return new LogicalCTEConsumer(StatementScopeIdGenerator.newRelationId(), context.cteId,
"comm_cte_" + context.cteId.asInt(), cToPBuilder.build(), pToCBuilder.build());
}
return visitChildren(this, plan, context);
}
}
22 changes: 22 additions & 0 deletions fe/fe-core/src/main/java/org/apache/doris/nereids/commonCTE/ConsumerRewriterContext.java
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,22 @@
package org.apache.doris.nereids.commonCTE;

import org.apache.doris.nereids.trees.expressions.CTEId;
import org.apache.doris.nereids.trees.plans.Plan;
import org.apache.doris.nereids.trees.plans.logical.AbstractLogicalPlan;

import java.util.List;

public class ConsumerRewriterContext {
private final List<Plan> targets;
public Plan producer;
public final CTEId cteId;

public ConsumerRewriterContext(List<Plan> targets, Plan producer, CTEId cteId) {
this.targets = targets;
this.producer = producer;
this.cteId = cteId;
}
public boolean isTarget(Plan plan) {
return targets.contains(plan);
}
}
102 changes: 88 additions & 14 deletions fe/fe-core/src/main/java/org/apache/doris/nereids/commonCTE/CteExtractor.java
View file
Original file line number Diff line number Diff line change
@@ -1,43 +1,117 @@
package org.apache.doris.nereids.commonCTE;

import org.apache.doris.nereids.CascadesContext;
import org.apache.doris.nereids.rules.analysis.CheckAfterRewrite;
import org.apache.doris.nereids.rules.exploration.mv.ComparisonResult;
import org.apache.doris.nereids.rules.exploration.mv.LogicalCompatibilityContext;
import org.apache.doris.nereids.rules.exploration.mv.MaterializationContext;
import org.apache.doris.nereids.rules.exploration.mv.StructInfo;
import org.apache.doris.nereids.rules.exploration.mv.mapping.RelationMapping;
import org.apache.doris.nereids.rules.exploration.mv.mapping.SlotMapping;
import org.apache.doris.nereids.trees.copier.DeepCopierContext;
import org.apache.doris.nereids.trees.copier.LogicalPlanDeepCopier;
import org.apache.doris.nereids.trees.expressions.CTEId;
import org.apache.doris.nereids.trees.expressions.StatementScopeIdGenerator;
import org.apache.doris.nereids.trees.plans.AbstractPlan;
import org.apache.doris.nereids.trees.plans.Plan;
import org.apache.doris.nereids.trees.plans.logical.AbstractLogicalPlan;
import org.apache.doris.nereids.trees.plans.logical.LogicalCTEAnchor;
import org.apache.doris.nereids.trees.plans.logical.LogicalCTEProducer;
import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;

import com.google.common.collect.Lists;

import java.util.ArrayList;
import java.util.BitSet;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.stream.Collectors;

public class CteExtractor {
private AbstractLogicalPlan plan;
private Map<Plan, TableSignature> anchorToSignature = new HashMap<>();
private Map<TableSignature, List<Plan>> signatureToAnchorList = new HashMap<>();
private final CascadesContext cascadesContext;
private final AbstractLogicalPlan plan;
private final Map<AbstractLogicalPlan, TableSignature> planToSignature = new HashMap<>();
private final Map<TableSignature, List<AbstractLogicalPlan>> signatureToPlanList = new HashMap<>();

public CteExtractor(AbstractLogicalPlan plan) {
public CteExtractor(AbstractLogicalPlan plan, CascadesContext cascadesContext) {
this.plan = plan;
this.cascadesContext = cascadesContext;
}

public AbstractLogicalPlan execute() {
public List<Plan> execute() {
sign();
return plan;
return extract();
}

private void sign() {
SignatureVisitor visitor = new SignatureVisitor();
visitor.visit(plan, anchorToSignature);
extract();
visitor.visit(plan, planToSignature);
}

private void extract() {
List<TableSignature> a = anchorToSignature.values().stream().collect(Collectors.toList());
public List<Plan> extract() {
for (AbstractLogicalPlan plan : planToSignature.keySet()) {
TableSignature signature = planToSignature.get(plan);
List<AbstractLogicalPlan> plans = signatureToPlanList.computeIfAbsent(signature, key -> new ArrayList<>());
plans.add(plan);
}
TableSignature targetSignature = null;
for (TableSignature signature : signatureToPlanList.keySet()) {
if (signature.isContainsAggregation()) {
targetSignature = signature;
}
}
if (targetSignature != null) {
List<AbstractLogicalPlan> targetPlans = signatureToPlanList.get(targetSignature);
Plan target0 = targetPlans.get(0);
Optional<StructInfo> optStruct0 = MaterializationContext.constructStructInfo(target0, target0, cascadesContext, new BitSet());
StructInfo viewStructInfo = optStruct0.get();
Plan target1 = targetPlans.get(1);
Optional<StructInfo> optStruct1 = MaterializationContext.constructStructInfo(target1, target1, cascadesContext, new BitSet());
StructInfo queryStructInfo = optStruct1.get();
List<RelationMapping> queryToViewTableMappings = RelationMapping.generate(optStruct0.get().getRelations(),
optStruct1.get().getRelations());
RelationMapping queryToViewTableMapping = queryToViewTableMappings.get(0);
SlotMapping queryToViewSlotMapping = SlotMapping.generate(queryToViewTableMapping);
SlotMapping viewToQuerySlotMapping = queryToViewSlotMapping.inverse();
LogicalCompatibilityContext compatibilityContext = LogicalCompatibilityContext.from(
queryToViewTableMapping, viewToQuerySlotMapping, optStruct0.get(), optStruct1.get());
ComparisonResult comparisonResult = StructInfo.isGraphLogicalEquals(queryStructInfo, viewStructInfo,
compatibilityContext);
if (!comparisonResult.isInvalid()) {
CTEId cteId = StatementScopeIdGenerator.newCTEId();
Plan producerBody = LogicalPlanDeepCopier.INSTANCE
.deepCopy((LogicalPlan) target0, new DeepCopierContext());
LogicalCTEProducer producer = new LogicalCTEProducer(cteId, producerBody);
ConsumerRewriterContext context = new ConsumerRewriterContext(Lists.newArrayList(target0, target1), producer, cteId);
Plan rewritten = plan.accept(ConsumerRewriter.INSTANCE, context);
AbstractLogicalPlan root = new LogicalCTEAnchor(cteId, producer, rewritten);
CheckAfterRewrite checker = new CheckAfterRewrite();
checker.checkValidation(root);
return Lists.newArrayList(root);
}
System.out.println(queryToViewTableMappings);
}
return Lists.newArrayList();
}

private void checkPlanValidation(Plan plan) {

for (Plan plan : anchorToSignature.keySet()) {
TableSignature signature = anchorToSignature.get(plan);
List<Plan> plans = signatureToAnchorList.computeIfAbsent(signature, key -> new ArrayList<>());
plans.add(plan);
}

@Override
public String toString() {
StringBuilder sb = new StringBuilder();
for (TableSignature signature : signatureToPlanList.keySet()) {
sb.append(String.format("%s\n", signature.toString()));
sb.append(" ");
for (AbstractLogicalPlan plan : signatureToPlanList.get(signature)) {
sb.append(plan.getId()).append(" ");
}
sb.append("\n");
}
return sb.toString();
}


Expand Down
15 changes: 8 additions & 7 deletions fe/fe-core/src/main/java/org/apache/doris/nereids/commonCTE/SignatureVisitor.java
View file
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
package org.apache.doris.nereids.commonCTE;

import org.apache.doris.nereids.trees.plans.Plan;
import org.apache.doris.nereids.trees.plans.logical.AbstractLogicalPlan;
import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
import org.apache.doris.nereids.trees.plans.logical.LogicalCatalogRelation;
import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
Expand All @@ -12,11 +13,11 @@

import java.util.Map;

public class SignatureVisitor extends DefaultPlanVisitor <TableSignature, Map<Plan, TableSignature>>{
public class SignatureVisitor extends DefaultPlanVisitor <TableSignature, Map<AbstractLogicalPlan, TableSignature>>{

@Override
public TableSignature visitLogicalCatalogRelation(LogicalCatalogRelation relation,
Map<Plan, TableSignature> signatureMap) {
Map<AbstractLogicalPlan, TableSignature> signatureMap) {
TableSignature signature = new TableSignature(true, false,
ImmutableSet.of(relation.getTable().getId()));
signatureMap.put(relation, signature);
Expand All @@ -25,7 +26,7 @@ public TableSignature visitLogicalCatalogRelation(LogicalCatalogRelation relatio

@Override
public TableSignature visitLogicalFilter(LogicalFilter<? extends Plan> filter,
Map<Plan, TableSignature> signatureMap) {
Map<AbstractLogicalPlan, TableSignature> signatureMap) {
TableSignature childSignature = filter.child().accept(this, signatureMap);
if (filter.child() instanceof LogicalAggregate) {
return TableSignature.EMPTY;
Expand All @@ -36,7 +37,7 @@ public TableSignature visitLogicalFilter(LogicalFilter<? extends Plan> filter,

@Override
public TableSignature visitLogicalJoin(LogicalJoin<? extends Plan, ? extends Plan> join,
Map<Plan, TableSignature> signatureMap) {
Map<AbstractLogicalPlan, TableSignature> signatureMap) {
TableSignature signature = TableSignature.EMPTY;
TableSignature leftSignature = join.left().accept(this, signatureMap);

Expand All @@ -56,7 +57,7 @@ public TableSignature visitLogicalJoin(LogicalJoin<? extends Plan, ? extends Pla

@Override
public TableSignature visitLogicalAggregate(LogicalAggregate<? extends Plan> aggregate,
Map<Plan, TableSignature> signatureMap) {
Map<AbstractLogicalPlan, TableSignature> signatureMap) {
TableSignature signature = TableSignature.EMPTY;
TableSignature childSignature = aggregate.child().accept(this, signatureMap);
if (childSignature != TableSignature.EMPTY) {
Expand All @@ -68,7 +69,7 @@ public TableSignature visitLogicalAggregate(LogicalAggregate<? extends Plan> agg

@Override
public TableSignature visitLogicalProject(LogicalProject<? extends Plan> project,
Map<Plan, TableSignature> signatureMap) {
Map<AbstractLogicalPlan, TableSignature> signatureMap) {
TableSignature childSignature = project.child().accept(this, signatureMap);
if (childSignature != TableSignature.EMPTY) {
signatureMap.put(project, childSignature);
Expand All @@ -77,7 +78,7 @@ public TableSignature visitLogicalProject(LogicalProject<? extends Plan> project
}

@Override
public TableSignature visit(Plan plan, Map<Plan, TableSignature> signatureMap) {
public TableSignature visit(Plan plan, Map<AbstractLogicalPlan, TableSignature> signatureMap) {
for (Plan child : plan.children()) {
child.accept(this, signatureMap);
}
Expand Down
4 changes: 4 additions & 0 deletions fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PlanTranslatorContext.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -212,6 +212,10 @@ public void addPlanFragment(PlanFragment planFragment) {
}

public void addExprIdSlotRefPair(ExprId exprId, SlotRef slotRef) {

if (exprIdToSlotRef.get(exprId) != null) {
System.out.println("eeee");
}
exprIdToSlotRef.put(exprId, slotRef);
slotIdToExprId.put(slotRef.getDesc().getId(), exprId);
}
Expand Down
4 changes: 4 additions & 0 deletions fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/executor/Optimizer.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,10 @@ public Optimizer(CascadesContext cascadesContext) {
public void execute() {
// init memo
cascadesContext.toMemo();
cascadesContext.memos.put(cascadesContext.plan, cascadesContext.getMemo());
// if (cascadesContext.cteplan != null) {
// cascadesContext.getMemo().copyIn(cascadesContext.cteplan, cascadesContext.getMemo().getRoot(), false);
// }
// stats derive
cascadesContext.pushJob(new DeriveStatsJob(cascadesContext.getMemo().getRoot().getLogicalExpression(),
cascadesContext.getCurrentJobContext()));
Expand Down
Loading

0 comments on commit 486c9a4

Please sign in to comment.