FatLittle commented on a change in pull request #1991: URL: https://github.com/apache/calcite/pull/1991#discussion_r452102742
########## File path: core/src/main/java/org/apache/calcite/plan/volcano/TopDownRuleDriver.java ########## @@ -0,0 +1,928 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.calcite.plan.volcano; + +import org.apache.calcite.plan.ConventionTraitDef; +import org.apache.calcite.plan.DeriveMode; +import org.apache.calcite.plan.RelOptCost; +import org.apache.calcite.plan.RelTrait; +import org.apache.calcite.plan.RelTraitSet; +import org.apache.calcite.rel.PhysicalNode; +import org.apache.calcite.rel.RelNode; +import org.apache.calcite.rel.convert.ConverterRule; +import org.apache.calcite.util.Pair; +import org.apache.calcite.util.trace.CalciteTrace; + +import org.slf4j.Logger; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.HashSet; +import java.util.List; +import java.util.Set; +import java.util.Stack; +import java.util.function.Predicate; + +class TopDownRuleDriver implements RuleDriver { + + /** + * the Logger + */ + private static final Logger LOGGER = CalciteTrace.getPlannerTaskTracer(); + + /** + * The planner + */ + private final VolcanoPlanner planner; + + /** + * the rule queue designed for top-down rule applying + */ + private final CascadesRuleQueue ruleQueue; + + /** + * all tasks waiting for execution + */ + private Stack<Task> tasks = new Stack<>(); + + /** + * A task that is currently applying and may generate new RelNode. + * It provides a callback to schedule tasks for new RelNodes that + * are registered during task performing + */ + private GeneratorTask applying = null; + + /** + * relnodes that are generated by passThrough or derive + * these nodes will not takes part in another passThrough or derive + */ + private Set<RelNode> passThroughCache = new HashSet<>(); + + //~ Constructors ----------------------------------------------------------- + + TopDownRuleDriver(VolcanoPlanner planner) { + this.planner = planner; + ruleQueue = new CascadesRuleQueue(planner); + } + + //~ Methods ---------------------------------------------------------------- + + @Override public void drive() { + TaskDescriptor description = new TaskDescriptor(); + tasks.push(new OptimizeGroup(planner.root, planner.infCost)); + exploreMaterializationRoots(); + while (!tasks.isEmpty()) { + Task task = tasks.pop(); + description.log(task); + task.perform(); + } + } + + private void exploreMaterializationRoots() { + for (RelSubset extraRoot : planner.explorationRoots) { + RelSet rootSet = VolcanoPlanner.equivRoot(extraRoot.set); + if (rootSet == planner.root.set) { + continue; + } + for (RelNode rel : extraRoot.set.rels) { + if (planner.isLogical(rel)) { + tasks.push(new OptimizeMExpr(rel, extraRoot, true)); + } + } + } + } + + @Override public CascadesRuleQueue getRuleQueue() { + return ruleQueue; + } + + @Override public void clear() { + ruleQueue.clear(); + tasks.clear(); + passThroughCache.clear(); + applying = null; + } + + private interface Procedure { + void exec(); + } + + private void applyGenerator(GeneratorTask task, Procedure proc) { + GeneratorTask applying = this.applying; + this.applying = task; + try { + proc.exec(); + } finally { + this.applying = applying; + } + } + + public void onSetMerged(RelSet set) { + applyGenerator(null, () -> clearProcessed(set)); + } + + private void clearProcessed(RelSet set) { + boolean explored = set.exploringState != null; + set.exploringState = null; + + for (RelSubset subset : set.subsets) { + if (subset.resetOptimizing() || explored) { + Collection<RelNode> parentRels = subset.getParentRels(); + for (RelNode parentRel : parentRels) { + clearProcessed(planner.getSet(parentRel)); + } + if (subset == planner.root) { + tasks.push(new OptimizeGroup(subset, planner.infCost)); + } + } + } + } + + public void onProduce(RelNode node, RelSubset subset) { + if (applying == null || subset.set + != VolcanoPlanner.equivRoot(applying.group().set)) { + return; + } + + if (!applying.onProduce(node)) { + return; + } + + if (!planner.isLogical(node)) { + RelSubset optimizingGroup = null; + boolean canPassThrough = node instanceof PhysicalNode + && !passThroughCache.contains(node); + if (!canPassThrough && subset.taskState != null) { + optimizingGroup = subset; + } else { + RelOptCost upperBound = planner.zeroCost; + RelSet set = subset.getSet(); + List<RelSubset> subsetsToPassThrough = new ArrayList<>(); + for (RelSubset otherSubset : set.subsets) { + if (!otherSubset.isRequired() || otherSubset != planner.root + && otherSubset.taskState != RelSubset.OptimizeState.OPTIMIZING) { + continue; + } + if (node.getTraitSet().satisfies(otherSubset.getTraitSet())) { + if (upperBound.isLt(otherSubset.upperBound)) { + upperBound = otherSubset.upperBound; + optimizingGroup = otherSubset; + } + } else if (canPassThrough) { + subsetsToPassThrough.add(otherSubset); + } + } + for (RelSubset otherSubset : subsetsToPassThrough) { + Task task = getOptimizeInputTask(node, otherSubset); + if (task != null) { + tasks.push(task); + } + } + } + if (optimizingGroup == null) { + return; + } + Task task = getOptimizeInputTask(node, optimizingGroup); + if (task != null) { + tasks.push(task); + } + } else { + boolean optimizing = subset.set.subsets.stream() + .anyMatch(s -> s.taskState == RelSubset.OptimizeState.OPTIMIZING); + tasks.push( + new OptimizeMExpr(node, applying.group(), + applying.exploring() && !optimizing)); + } + } + + //~ Inner Classes ---------------------------------------------------------- + + /** + * Base class for planner task + */ + private interface Task { + void perform(); + void describe(TaskDescriptor desc); + } + + /** + * A class for task logging + */ + private static class TaskDescriptor { + private boolean first = true; + private StringBuilder builder = new StringBuilder(); + + void log(Task task) { + if (!LOGGER.isDebugEnabled()) { + return; + } + first = true; + builder.setLength(0); + builder.append("Execute task: ").append(task.getClass().getSimpleName()); + task.describe(this); + if (!first) { + builder.append(")"); + } + + LOGGER.info(builder.toString()); + } + + TaskDescriptor item(String name, Object value) { + if (first) { + first = false; + builder.append("("); + } else { + builder.append(", "); + } + builder.append(name).append("=").append(value); + return this; + } + } + + private interface GeneratorTask extends Task { + RelSubset group(); + boolean exploring(); + default boolean onProduce(RelNode node) { + return true; + } + } + + /** + * O_GROUP + */ + private class OptimizeGroup implements Task { + private final RelSubset group; + private RelOptCost upperBound; + + OptimizeGroup(RelSubset group, RelOptCost upperBound) { + this.group = group; + this.upperBound = upperBound; + } + + @Override public void perform() { + RelOptCost winner = group.getWinnerCost(); + if (winner != null) { + return; + } + + if (group.taskState != null && upperBound.isLe(group.upperBound)) { + // this group failed to optimize before or it is a ring + return; + } + + group.startOptimize(upperBound); + + // cannot decide an actual lower bound before MExpr are fully explored + // so delay the lower bound checking + + // a gate keeper to update context + tasks.push(new GroupOptimized(group)); + + // optimize mExprs in group + List<RelNode> physicals = new ArrayList<>(); + for (RelNode rel : group.set.rels) { + if (planner.isLogical(rel)) { + tasks.push(new OptimizeMExpr(rel, group, false)); + } else if (rel.isEnforcer()) { + // Enforcers have lower priority than other physical nodes + physicals.add(0, rel); + } else { + physicals.add(rel); + } + } + // always apply O_INPUTS first so as to get an valid upper bound + for (RelNode rel : physicals) { + Task task = getOptimizeInputTask(rel, group); + if (task != null) { + tasks.add(task); + } + } + } + + @Override public void describe(TaskDescriptor desc) { + desc.item("group", group).item("upperBound", upperBound); + } + } + + /** + * Mark the group optimized + */ + private static class GroupOptimized implements Task { + private final RelSubset group; + + GroupOptimized(RelSubset group) { + this.group = group; + } + + @Override public void perform() { + group.optimized(); + } + + @Override public void describe(TaskDescriptor desc) { + desc.item("group", group); + } + } + + /** + * O_EXPR + */ + private class OptimizeMExpr implements Task { + private final RelNode mExpr; + private final RelSubset group; + private final boolean explore; + + OptimizeMExpr(RelNode mExpr, + RelSubset group, boolean explore) { + this.mExpr = mExpr; + this.group = group; + this.explore = explore; + } + + @Override public void perform() { + if (explore && group.isExplored()) { + return; + } + // 1. explode input + // 2. apply other rules + tasks.push(new ApplyRules(mExpr, group, explore)); + for (int i = mExpr.getInputs().size() - 1; i >= 0; --i) { + tasks.push(new ExploreInput(mExpr, i)); + } + } + + @Override public void describe(TaskDescriptor desc) { + desc.item("mExpr", mExpr).item("explore", explore); + } + } + + /** + * ensure ExploreInput are working on the correct input group since calcite + * may merge sets + */ + private class EnsureGroupExplored implements Task { + + private final RelSubset input; + private final RelNode parent; + private final int inputOrdinal; + + EnsureGroupExplored(RelSubset input, RelNode parent, int inputOrdinal) { + this.input = input; + this.parent = parent; + this.inputOrdinal = inputOrdinal; + } + + @Override public void perform() { + if (parent.getInput(inputOrdinal) != input) { + tasks.push(new ExploreInput(parent, inputOrdinal)); + return; + } + input.setExplored(); + for (RelSubset subset : input.getSet().subsets) { + // clear the LB cache as exploring state have changed + input.getCluster().getMetadataQuery().clearCache(subset); + } + } + + @Override public void describe(TaskDescriptor desc) { + desc.item("mExpr", parent).item("i", inputOrdinal); + } + } + + /** + * E_GROUP + */ + private class ExploreInput implements Task { + private final RelSubset group; + private final RelNode parent; + private final int inputOrdinal; + + ExploreInput(RelNode parent, int inputOrdinal) { + this.group = (RelSubset) parent.getInput(inputOrdinal); + this.parent = parent; + this.inputOrdinal = inputOrdinal; + } + + @Override public void perform() { + if (!group.explore()) { + return; + } + tasks.push(new EnsureGroupExplored(group, parent, inputOrdinal)); + for (RelNode rel : group.set.rels) { + if (planner.isLogical(rel)) { + tasks.push(new OptimizeMExpr(rel, group, true)); + } + } + } + + @Override public void describe(TaskDescriptor desc) { + desc.item("group", group); + } + } + + /** + * extract rule matches from rule queue and add them to task stack + */ + private class ApplyRules implements Task { + private final RelNode mExpr; + private final RelSubset group; + private final boolean exploring; + + ApplyRules(RelNode mExpr, RelSubset group, boolean exploring) { + this.mExpr = mExpr; + this.group = group; + this.exploring = exploring; + } + + @Override public void perform() { + Pair<RelNode, Predicate<VolcanoRuleMatch>> category = + exploring ? Pair.of(mExpr, planner::isTransformationRule) + : Pair.of(mExpr, m -> true); + VolcanoRuleMatch match = ruleQueue.popMatch(category); + while (match != null) { + tasks.push(new ApplyRule(match, group, exploring)); + match = ruleQueue.popMatch(category); + } + } + + @Override public void describe(TaskDescriptor desc) { + desc.item("mExpr", mExpr).item("exploring", exploring); + } + } + + /** + * APPLY_RULE + */ + private class ApplyRule implements GeneratorTask { + private final VolcanoRuleMatch match; + private final RelSubset group; + private final boolean exploring; + + ApplyRule(VolcanoRuleMatch match, RelSubset group, boolean exploring) { + this.match = match; + this.group = group; + this.exploring = exploring; + } + + @Override public void describe(TaskDescriptor desc) { + desc.item("match", match).item("exploring", exploring); + } + + @Override public void perform() { + if (!exploring && !planner.isLogical(match.rel(0)) && !checkLowerBound()) { + // for implementation and enforcing rules, + // we can skip rule match by checking the lower bound + // however, we cannot decide LB for logical nodes, + // so we only check enforcing rules now + if (!group.getTraitSet().satisfies(match.rels[0].getTraitSet())) { + // the match cannot be discarded directly + // because of another subset's upper bound + ruleQueue.addMatch(match); + } + return; + } + + applyGenerator(this, match::onMatch); + } + + private boolean checkLowerBound() { + RelOptCost upperBound = group.upperBound; + if (upperBound.isInfinite()) { + return true; + } + RelOptCost lb = planner.getLowerBound(match); + if (upperBound.isLe(lb)) { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "Skip because of lower bound. LB = {}, UP = {}", + lb, upperBound); + } + return false; + } + return true; + } + + @Override public RelSubset group() { + return group; + } + + @Override public boolean exploring() { + return exploring; + } + } + + private Task getOptimizeInputTask(RelNode rel, RelSubset group) { + if (!rel.getTraitSet().satisfies(group.getTraitSet())) { + RelNode passThroughRel = convert(rel, group); + if (passThroughRel == null) { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug("Skip optimizing because of traits: {}", rel); + } + return null; + } + final RelNode finalPassThroughRel = passThroughRel; + applyGenerator(null, () -> + planner.register(finalPassThroughRel, group)); + rel = passThroughRel; + } + boolean unProcess = false; + for (RelNode input : rel.getInputs()) { + RelOptCost winner = ((RelSubset) input).getWinnerCost(); + if (winner == null) { + unProcess = true; + break; + } + } + if (!unProcess) { + return new DeriveTrait(rel, group); + } + if (rel.getInputs().size() == 1) { + return new OptimizeInput1(rel, group); + } + return new OptimizeInputs(rel, group); + } + + private RelNode convert(RelNode rel, RelSubset group) { + if (!passThroughCache.contains(rel)) { + RelNode passThrough = group.passThrough(rel); + if (passThrough != null) { + passThroughCache.add(passThrough); + return passThrough; + } + } + VolcanoRuleMatch match = ruleQueue.popMatch( + Pair.of(rel, + m -> m.getRule() instanceof ConverterRule + && m.getRule().getOutTrait().satisfies(group.getTraitSet().getConvention()))); + if (match != null) { + tasks.add(new ApplyRule(match, group, false)); + } + return null; + } + + /** + * O_INPUT when there is only one input + */ + private class OptimizeInput1 implements Task { + + private final RelNode mExpr; + private final RelSubset group; + + OptimizeInput1(RelNode mExpr, RelSubset group) { + this.mExpr = mExpr; + this.group = group; + } + + + @Override public void describe(TaskDescriptor desc) { + desc.item("mExpr", mExpr).item("upperBound", group.upperBound); + } + + @Override public void perform() { + RelOptCost upperBound = group.upperBound; + RelOptCost upperForInput = planner.upperBoundForInputs(mExpr, upperBound); + if (upperForInput.isLe(planner.zeroCost)) { + if (LOGGER.isDebugEnabled()) { Review comment: resolved ########## File path: core/src/main/java/org/apache/calcite/plan/volcano/TopDownRuleDriver.java ########## @@ -0,0 +1,928 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.calcite.plan.volcano; + +import org.apache.calcite.plan.ConventionTraitDef; +import org.apache.calcite.plan.DeriveMode; +import org.apache.calcite.plan.RelOptCost; +import org.apache.calcite.plan.RelTrait; +import org.apache.calcite.plan.RelTraitSet; +import org.apache.calcite.rel.PhysicalNode; +import org.apache.calcite.rel.RelNode; +import org.apache.calcite.rel.convert.ConverterRule; +import org.apache.calcite.util.Pair; +import org.apache.calcite.util.trace.CalciteTrace; + +import org.slf4j.Logger; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.HashSet; +import java.util.List; +import java.util.Set; +import java.util.Stack; +import java.util.function.Predicate; + +class TopDownRuleDriver implements RuleDriver { + + /** + * the Logger + */ + private static final Logger LOGGER = CalciteTrace.getPlannerTaskTracer(); + + /** + * The planner + */ + private final VolcanoPlanner planner; + + /** + * the rule queue designed for top-down rule applying + */ + private final CascadesRuleQueue ruleQueue; + + /** + * all tasks waiting for execution + */ + private Stack<Task> tasks = new Stack<>(); + + /** + * A task that is currently applying and may generate new RelNode. + * It provides a callback to schedule tasks for new RelNodes that + * are registered during task performing + */ + private GeneratorTask applying = null; + + /** + * relnodes that are generated by passThrough or derive + * these nodes will not takes part in another passThrough or derive + */ + private Set<RelNode> passThroughCache = new HashSet<>(); + + //~ Constructors ----------------------------------------------------------- + + TopDownRuleDriver(VolcanoPlanner planner) { + this.planner = planner; + ruleQueue = new CascadesRuleQueue(planner); + } + + //~ Methods ---------------------------------------------------------------- + + @Override public void drive() { + TaskDescriptor description = new TaskDescriptor(); + tasks.push(new OptimizeGroup(planner.root, planner.infCost)); + exploreMaterializationRoots(); + while (!tasks.isEmpty()) { + Task task = tasks.pop(); + description.log(task); + task.perform(); + } + } + + private void exploreMaterializationRoots() { + for (RelSubset extraRoot : planner.explorationRoots) { + RelSet rootSet = VolcanoPlanner.equivRoot(extraRoot.set); + if (rootSet == planner.root.set) { + continue; + } + for (RelNode rel : extraRoot.set.rels) { + if (planner.isLogical(rel)) { + tasks.push(new OptimizeMExpr(rel, extraRoot, true)); + } + } + } + } + + @Override public CascadesRuleQueue getRuleQueue() { + return ruleQueue; + } + + @Override public void clear() { + ruleQueue.clear(); + tasks.clear(); + passThroughCache.clear(); + applying = null; + } + + private interface Procedure { + void exec(); + } + + private void applyGenerator(GeneratorTask task, Procedure proc) { + GeneratorTask applying = this.applying; + this.applying = task; + try { + proc.exec(); + } finally { + this.applying = applying; + } + } + + public void onSetMerged(RelSet set) { + applyGenerator(null, () -> clearProcessed(set)); + } + + private void clearProcessed(RelSet set) { + boolean explored = set.exploringState != null; + set.exploringState = null; + + for (RelSubset subset : set.subsets) { + if (subset.resetOptimizing() || explored) { + Collection<RelNode> parentRels = subset.getParentRels(); + for (RelNode parentRel : parentRels) { + clearProcessed(planner.getSet(parentRel)); + } + if (subset == planner.root) { + tasks.push(new OptimizeGroup(subset, planner.infCost)); + } + } + } + } + + public void onProduce(RelNode node, RelSubset subset) { + if (applying == null || subset.set + != VolcanoPlanner.equivRoot(applying.group().set)) { + return; + } + + if (!applying.onProduce(node)) { + return; + } + + if (!planner.isLogical(node)) { + RelSubset optimizingGroup = null; + boolean canPassThrough = node instanceof PhysicalNode + && !passThroughCache.contains(node); + if (!canPassThrough && subset.taskState != null) { + optimizingGroup = subset; + } else { + RelOptCost upperBound = planner.zeroCost; + RelSet set = subset.getSet(); + List<RelSubset> subsetsToPassThrough = new ArrayList<>(); + for (RelSubset otherSubset : set.subsets) { + if (!otherSubset.isRequired() || otherSubset != planner.root + && otherSubset.taskState != RelSubset.OptimizeState.OPTIMIZING) { + continue; + } + if (node.getTraitSet().satisfies(otherSubset.getTraitSet())) { + if (upperBound.isLt(otherSubset.upperBound)) { + upperBound = otherSubset.upperBound; + optimizingGroup = otherSubset; + } + } else if (canPassThrough) { + subsetsToPassThrough.add(otherSubset); + } + } + for (RelSubset otherSubset : subsetsToPassThrough) { + Task task = getOptimizeInputTask(node, otherSubset); + if (task != null) { + tasks.push(task); + } + } + } + if (optimizingGroup == null) { + return; + } + Task task = getOptimizeInputTask(node, optimizingGroup); + if (task != null) { + tasks.push(task); + } + } else { + boolean optimizing = subset.set.subsets.stream() + .anyMatch(s -> s.taskState == RelSubset.OptimizeState.OPTIMIZING); + tasks.push( + new OptimizeMExpr(node, applying.group(), + applying.exploring() && !optimizing)); + } + } + + //~ Inner Classes ---------------------------------------------------------- + + /** + * Base class for planner task + */ + private interface Task { + void perform(); + void describe(TaskDescriptor desc); + } + + /** + * A class for task logging + */ + private static class TaskDescriptor { + private boolean first = true; + private StringBuilder builder = new StringBuilder(); + + void log(Task task) { + if (!LOGGER.isDebugEnabled()) { + return; + } + first = true; + builder.setLength(0); + builder.append("Execute task: ").append(task.getClass().getSimpleName()); + task.describe(this); + if (!first) { + builder.append(")"); + } + + LOGGER.info(builder.toString()); + } + + TaskDescriptor item(String name, Object value) { + if (first) { + first = false; + builder.append("("); + } else { + builder.append(", "); + } + builder.append(name).append("=").append(value); + return this; + } + } + + private interface GeneratorTask extends Task { + RelSubset group(); + boolean exploring(); + default boolean onProduce(RelNode node) { + return true; + } + } + + /** + * O_GROUP + */ + private class OptimizeGroup implements Task { + private final RelSubset group; + private RelOptCost upperBound; + + OptimizeGroup(RelSubset group, RelOptCost upperBound) { + this.group = group; + this.upperBound = upperBound; + } + + @Override public void perform() { + RelOptCost winner = group.getWinnerCost(); + if (winner != null) { + return; + } + + if (group.taskState != null && upperBound.isLe(group.upperBound)) { + // this group failed to optimize before or it is a ring + return; + } + + group.startOptimize(upperBound); + + // cannot decide an actual lower bound before MExpr are fully explored + // so delay the lower bound checking + + // a gate keeper to update context + tasks.push(new GroupOptimized(group)); + + // optimize mExprs in group + List<RelNode> physicals = new ArrayList<>(); + for (RelNode rel : group.set.rels) { + if (planner.isLogical(rel)) { + tasks.push(new OptimizeMExpr(rel, group, false)); + } else if (rel.isEnforcer()) { + // Enforcers have lower priority than other physical nodes + physicals.add(0, rel); + } else { + physicals.add(rel); + } + } + // always apply O_INPUTS first so as to get an valid upper bound + for (RelNode rel : physicals) { + Task task = getOptimizeInputTask(rel, group); + if (task != null) { + tasks.add(task); + } + } + } + + @Override public void describe(TaskDescriptor desc) { + desc.item("group", group).item("upperBound", upperBound); + } + } + + /** + * Mark the group optimized + */ + private static class GroupOptimized implements Task { + private final RelSubset group; + + GroupOptimized(RelSubset group) { + this.group = group; + } + + @Override public void perform() { + group.optimized(); + } + + @Override public void describe(TaskDescriptor desc) { + desc.item("group", group); + } + } + + /** + * O_EXPR + */ + private class OptimizeMExpr implements Task { + private final RelNode mExpr; + private final RelSubset group; + private final boolean explore; + + OptimizeMExpr(RelNode mExpr, + RelSubset group, boolean explore) { + this.mExpr = mExpr; + this.group = group; + this.explore = explore; + } + + @Override public void perform() { + if (explore && group.isExplored()) { + return; + } + // 1. explode input + // 2. apply other rules + tasks.push(new ApplyRules(mExpr, group, explore)); + for (int i = mExpr.getInputs().size() - 1; i >= 0; --i) { + tasks.push(new ExploreInput(mExpr, i)); + } + } + + @Override public void describe(TaskDescriptor desc) { + desc.item("mExpr", mExpr).item("explore", explore); + } + } + + /** + * ensure ExploreInput are working on the correct input group since calcite + * may merge sets + */ + private class EnsureGroupExplored implements Task { + + private final RelSubset input; + private final RelNode parent; + private final int inputOrdinal; + + EnsureGroupExplored(RelSubset input, RelNode parent, int inputOrdinal) { + this.input = input; + this.parent = parent; + this.inputOrdinal = inputOrdinal; + } + + @Override public void perform() { + if (parent.getInput(inputOrdinal) != input) { + tasks.push(new ExploreInput(parent, inputOrdinal)); + return; + } + input.setExplored(); + for (RelSubset subset : input.getSet().subsets) { + // clear the LB cache as exploring state have changed + input.getCluster().getMetadataQuery().clearCache(subset); + } + } + + @Override public void describe(TaskDescriptor desc) { + desc.item("mExpr", parent).item("i", inputOrdinal); + } + } + + /** + * E_GROUP + */ + private class ExploreInput implements Task { + private final RelSubset group; + private final RelNode parent; + private final int inputOrdinal; + + ExploreInput(RelNode parent, int inputOrdinal) { + this.group = (RelSubset) parent.getInput(inputOrdinal); + this.parent = parent; + this.inputOrdinal = inputOrdinal; + } + + @Override public void perform() { + if (!group.explore()) { + return; + } + tasks.push(new EnsureGroupExplored(group, parent, inputOrdinal)); + for (RelNode rel : group.set.rels) { + if (planner.isLogical(rel)) { + tasks.push(new OptimizeMExpr(rel, group, true)); + } + } + } + + @Override public void describe(TaskDescriptor desc) { + desc.item("group", group); + } + } + + /** + * extract rule matches from rule queue and add them to task stack + */ + private class ApplyRules implements Task { + private final RelNode mExpr; + private final RelSubset group; + private final boolean exploring; + + ApplyRules(RelNode mExpr, RelSubset group, boolean exploring) { + this.mExpr = mExpr; + this.group = group; + this.exploring = exploring; + } + + @Override public void perform() { + Pair<RelNode, Predicate<VolcanoRuleMatch>> category = + exploring ? Pair.of(mExpr, planner::isTransformationRule) + : Pair.of(mExpr, m -> true); + VolcanoRuleMatch match = ruleQueue.popMatch(category); + while (match != null) { + tasks.push(new ApplyRule(match, group, exploring)); + match = ruleQueue.popMatch(category); + } + } + + @Override public void describe(TaskDescriptor desc) { + desc.item("mExpr", mExpr).item("exploring", exploring); + } + } + + /** + * APPLY_RULE + */ + private class ApplyRule implements GeneratorTask { + private final VolcanoRuleMatch match; + private final RelSubset group; + private final boolean exploring; + + ApplyRule(VolcanoRuleMatch match, RelSubset group, boolean exploring) { + this.match = match; + this.group = group; + this.exploring = exploring; + } + + @Override public void describe(TaskDescriptor desc) { + desc.item("match", match).item("exploring", exploring); + } + + @Override public void perform() { + if (!exploring && !planner.isLogical(match.rel(0)) && !checkLowerBound()) { + // for implementation and enforcing rules, + // we can skip rule match by checking the lower bound + // however, we cannot decide LB for logical nodes, + // so we only check enforcing rules now + if (!group.getTraitSet().satisfies(match.rels[0].getTraitSet())) { + // the match cannot be discarded directly + // because of another subset's upper bound + ruleQueue.addMatch(match); + } + return; + } + + applyGenerator(this, match::onMatch); + } + + private boolean checkLowerBound() { + RelOptCost upperBound = group.upperBound; + if (upperBound.isInfinite()) { + return true; + } + RelOptCost lb = planner.getLowerBound(match); + if (upperBound.isLe(lb)) { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "Skip because of lower bound. LB = {}, UP = {}", + lb, upperBound); + } + return false; + } + return true; + } + + @Override public RelSubset group() { + return group; + } + + @Override public boolean exploring() { + return exploring; + } + } + + private Task getOptimizeInputTask(RelNode rel, RelSubset group) { + if (!rel.getTraitSet().satisfies(group.getTraitSet())) { + RelNode passThroughRel = convert(rel, group); + if (passThroughRel == null) { + if (LOGGER.isDebugEnabled()) { Review comment: resolved ########## File path: core/src/main/java/org/apache/calcite/plan/volcano/VolcanoPlanner.java ########## @@ -1396,6 +1437,103 @@ public void setLocked(boolean locked) { this.locked = locked; } + /** + * decide whether a rule is logical or not + * @param rel the specific rel node + * @return true if the relnode is a logical node + */ + public boolean isLogical(RelNode rel) { + return !(rel instanceof PhysicalNode) + && rel.getConvention() != rootConvention; + } + + /** + * check whether a rule match is a substitute rule match + * @param match the rule match to check + * @return true if the rule match is a substitute rule match + */ + protected boolean isSubstituteRule(VolcanoRuleCall match) { + return match.getRule() instanceof SubstitutionRule; + } + + /** + * check whether a rule match is a transformation rule match + * @param match the rule match to check + * @return true if the rule match is a transformation rule match + */ + protected boolean isTransformationRule(VolcanoRuleCall match) { + if (match.getRule() instanceof SubstitutionRule) { + return true; + } + if (match.getRule() instanceof ConverterRule + && match.getRule().getOutTrait() == rootConvention) { + return false; + } + return match.getRule().getOperand().trait == Convention.NONE + || match.getRule().getOperand().trait == null; + } + + + /** + * gets the lower bound cost of a relational operator + * @param rel the rel node + * @return the lower bound cost of the given rel. The value is ensured NOT NULL. + */ + protected RelOptCost getLowerBound(RelNode rel) { + RelMetadataQuery mq = rel.getCluster().getMetadataQuery(); + RelOptCost lowerBound = mq.getLowerBoundCost(rel, this); + if (lowerBound == null) { + return zeroCost; + } + return lowerBound; + } + + /** + * Gets the lower bound cost of a RelOptRuleCall. + * A match match with inputs whose Sum(LB) is higher than upper bound + * needs not to be applied + */ + protected RelOptCost getLowerBound(RelOptRuleCall match) { + return getLowerBoundInternal(match, match.getOperand0()); + } + + private RelOptCost getLowerBoundInternal( + RelOptRuleCall match, RelOptRuleOperand op) { + List<RelOptRuleOperand> children = op.getChildOperands(); + if (children == null || children.isEmpty()) { + RelNode rel = match.rel(op.ordinalInRule); + RelOptCost sum = zeroCost; + for (RelNode input : rel.getInputs()) { + RelOptCost lb = getLowerBound(input); + sum = sum == zeroCost ? lb : sum.plus(lb); + } + return sum; + } + RelOptCost sum = null; + for (RelOptRuleOperand child : children) { + RelOptCost lb = getLowerBoundInternal(match, child); + sum = sum == null ? lb : sum.plus(lb); + } + return sum; + } + + /** + * Gets the upper bound of its inputs. + * Allow users to overwrite this method as some implementations may have + * different cost model on some RelNodes, like Spool. + */ + protected RelOptCost upperBoundForInputs( + RelNode mExpr, RelOptCost upperBound) { + if (!upperBound.isInfinite()) { + RelOptCost rootCost = mExpr.computeSelfCost(this, Review comment: resolved ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: [email protected]
