You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tajo.apache.org by hy...@apache.org on 2014/10/25 20:18:00 UTC
[11/28] TAJO-1125: Separate logical plan and optimizer into a maven
module.
http://git-wip-us.apache.org/repos/asf/tajo/blob/b143f991/tajo-plan/src/main/java/org/apache/tajo/plan/LogicalOptimizer.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/LogicalOptimizer.java b/tajo-plan/src/main/java/org/apache/tajo/plan/LogicalOptimizer.java
new file mode 100644
index 0000000..bcc3bce
--- /dev/null
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/LogicalOptimizer.java
@@ -0,0 +1,316 @@
+/**
+ * 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.tajo.plan;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.tajo.OverridableConf;
+import org.apache.tajo.SessionVars;
+import org.apache.tajo.algebra.JoinType;
+import org.apache.tajo.conf.TajoConf;
+import org.apache.tajo.conf.TajoConf.ConfVars;
+import org.apache.tajo.util.graph.DirectedGraphCursor;
+import org.apache.tajo.plan.expr.AlgebraicUtil;
+import org.apache.tajo.plan.expr.EvalNode;
+import org.apache.tajo.plan.joinorder.FoundJoinOrder;
+import org.apache.tajo.plan.joinorder.GreedyHeuristicJoinOrderAlgorithm;
+import org.apache.tajo.plan.joinorder.JoinGraph;
+import org.apache.tajo.plan.joinorder.JoinOrderAlgorithm;
+import org.apache.tajo.plan.logical.*;
+import org.apache.tajo.plan.rewrite.*;
+import org.apache.tajo.plan.rewrite.rules.FilterPushDownRule;
+import org.apache.tajo.plan.rewrite.rules.PartitionedTableRewriter;
+import org.apache.tajo.plan.rewrite.rules.ProjectionPushDownRule;
+import org.apache.tajo.plan.util.PlannerUtil;
+import org.apache.tajo.plan.visitor.BasicLogicalPlanVisitor;
+
+import java.util.LinkedHashSet;
+import java.util.Set;
+import java.util.Stack;
+
+import static org.apache.tajo.plan.LogicalPlan.BlockEdge;
+import static org.apache.tajo.plan.joinorder.GreedyHeuristicJoinOrderAlgorithm.getCost;
+
+/**
+ * This class optimizes a logical plan.
+ */
+@InterfaceStability.Evolving
+public class LogicalOptimizer {
+ private static final Log LOG = LogFactory.getLog(LogicalOptimizer.class.getName());
+
+ private BasicQueryRewriteEngine rulesBeforeJoinOpt;
+ private BasicQueryRewriteEngine rulesAfterToJoinOpt;
+ private JoinOrderAlgorithm joinOrderAlgorithm = new GreedyHeuristicJoinOrderAlgorithm();
+
+ public LogicalOptimizer(TajoConf systemConf) {
+ rulesBeforeJoinOpt = new BasicQueryRewriteEngine();
+ if (systemConf.getBoolVar(ConfVars.$TEST_FILTER_PUSHDOWN_ENABLED)) {
+ rulesBeforeJoinOpt.addRewriteRule(new FilterPushDownRule());
+ }
+
+ rulesAfterToJoinOpt = new BasicQueryRewriteEngine();
+ rulesAfterToJoinOpt.addRewriteRule(new ProjectionPushDownRule());
+ rulesAfterToJoinOpt.addRewriteRule(new PartitionedTableRewriter(systemConf));
+
+ // Currently, it is only used for some test cases to inject exception manually.
+ String userDefinedRewriterClass = systemConf.get("tajo.plan.rewriter.classes");
+ if (userDefinedRewriterClass != null && !userDefinedRewriterClass.isEmpty()) {
+ for (String eachRewriterClass : userDefinedRewriterClass.split(",")) {
+ try {
+ RewriteRule rule = (RewriteRule) Class.forName(eachRewriterClass).newInstance();
+ rulesAfterToJoinOpt.addRewriteRule(rule);
+ } catch (Exception e) {
+ LOG.error("Can't initiate a Rewriter object: " + eachRewriterClass, e);
+ continue;
+ }
+ }
+ }
+ }
+
+ public LogicalNode optimize(LogicalPlan plan) throws PlanningException {
+ return optimize(null, plan);
+ }
+
+ public LogicalNode optimize(OverridableConf context, LogicalPlan plan) throws PlanningException {
+ rulesBeforeJoinOpt.rewrite(plan);
+
+ DirectedGraphCursor<String, BlockEdge> blockCursor =
+ new DirectedGraphCursor<String, BlockEdge>(plan.getQueryBlockGraph(), plan.getRootBlock().getName());
+
+ if (context == null || context.getBool(SessionVars.TEST_JOIN_OPT_ENABLED)) {
+ // default is true
+ while (blockCursor.hasNext()) {
+ optimizeJoinOrder(plan, blockCursor.nextBlock());
+ }
+ } else {
+ LOG.info("Skip Join Optimized.");
+ }
+ rulesAfterToJoinOpt.rewrite(plan);
+ return plan.getRootBlock().getRoot();
+ }
+
+ private void optimizeJoinOrder(LogicalPlan plan, String blockName) throws PlanningException {
+ LogicalPlan.QueryBlock block = plan.getBlock(blockName);
+
+ if (block.hasNode(NodeType.JOIN)) {
+ String originalOrder = JoinOrderStringBuilder.buildJoinOrderString(plan, block);
+ double nonOptimizedJoinCost = JoinCostComputer.computeCost(plan, block);
+
+ // finding relations and filter expressions
+ JoinGraphContext joinGraphContext = JoinGraphBuilder.buildJoinGraph(plan, block);
+
+ // finding join order and restore remain filter order
+ FoundJoinOrder order = joinOrderAlgorithm.findBestOrder(plan, block,
+ joinGraphContext.joinGraph, joinGraphContext.relationsForProduct);
+
+ // replace join node with FoundJoinOrder.
+ JoinNode newJoinNode = order.getOrderedJoin();
+ JoinNode old = PlannerUtil.findTopNode(block.getRoot(), NodeType.JOIN);
+
+ JoinTargetCollector collector = new JoinTargetCollector();
+ Set<Target> targets = new LinkedHashSet<Target>();
+ collector.visitJoin(targets, plan, block, old, new Stack<LogicalNode>());
+
+ if (targets.size() == 0) {
+ newJoinNode.setTargets(PlannerUtil.schemaToTargets(old.getOutSchema()));
+ } else {
+ newJoinNode.setTargets(targets.toArray(new Target[targets.size()]));
+ }
+ PlannerUtil.replaceNode(plan, block.getRoot(), old, newJoinNode);
+ // End of replacement logic
+
+ String optimizedOrder = JoinOrderStringBuilder.buildJoinOrderString(plan, block);
+ block.addPlanHistory("Non-optimized join order: " + originalOrder + " (cost: " + nonOptimizedJoinCost + ")");
+ block.addPlanHistory("Optimized join order : " + optimizedOrder + " (cost: " + order.getCost() + ")");
+ }
+ }
+
+ private static class JoinTargetCollector extends BasicLogicalPlanVisitor<Set<Target>, LogicalNode> {
+ @Override
+ public LogicalNode visitJoin(Set<Target> ctx, LogicalPlan plan, LogicalPlan.QueryBlock block, JoinNode node,
+ Stack<LogicalNode> stack)
+ throws PlanningException {
+ super.visitJoin(ctx, plan, block, node, stack);
+
+ if (node.hasTargets()) {
+ for (Target target : node.getTargets()) {
+ ctx.add(target);
+ }
+ }
+ return node;
+ }
+ }
+
+ private static class JoinGraphContext {
+ JoinGraph joinGraph = new JoinGraph();
+ Set<EvalNode> quals = Sets.newHashSet();
+ Set<String> relationsForProduct = Sets.newHashSet();
+ }
+
+ private static class JoinGraphBuilder extends BasicLogicalPlanVisitor<JoinGraphContext, LogicalNode> {
+ private final static JoinGraphBuilder instance;
+
+ static {
+ instance = new JoinGraphBuilder();
+ }
+
+ /**
+ * This is based on the assumtion that all join and filter conditions are placed on the right join and
+ * scan operators. In other words, filter push down must be performed before this method.
+ * Otherwise, this method may build incorrectly a join graph.
+ */
+ public static JoinGraphContext buildJoinGraph(LogicalPlan plan, LogicalPlan.QueryBlock block)
+ throws PlanningException {
+ JoinGraphContext joinGraphContext = new JoinGraphContext();
+ instance.visit(joinGraphContext, plan, block);
+ return joinGraphContext;
+ }
+
+ public LogicalNode visitFilter(JoinGraphContext context, LogicalPlan plan, LogicalPlan.QueryBlock block,
+ SelectionNode node, Stack<LogicalNode> stack) throws PlanningException {
+ super.visitFilter(context, plan, block, node, stack);
+ context.quals.addAll(Lists.newArrayList(AlgebraicUtil.toConjunctiveNormalFormArray(node.getQual())));
+ return node;
+ }
+
+ @Override
+ public LogicalNode visitJoin(JoinGraphContext joinGraphContext, LogicalPlan plan, LogicalPlan.QueryBlock block,
+ JoinNode joinNode, Stack<LogicalNode> stack)
+ throws PlanningException {
+ super.visitJoin(joinGraphContext, plan, block, joinNode, stack);
+ if (joinNode.hasJoinQual()) {
+ joinGraphContext.joinGraph.addJoin(plan, block, joinNode);
+ } else {
+ LogicalNode leftChild = joinNode.getLeftChild();
+ LogicalNode rightChild = joinNode.getRightChild();
+ if (leftChild instanceof RelationNode) {
+ RelationNode rel = (RelationNode) leftChild;
+ joinGraphContext.relationsForProduct.add(rel.getCanonicalName());
+ }
+ if (rightChild instanceof RelationNode) {
+ RelationNode rel = (RelationNode) rightChild;
+ joinGraphContext.relationsForProduct.add(rel.getCanonicalName());
+ }
+ }
+ return joinNode;
+ }
+ }
+
+ public static class JoinOrderStringBuilder extends BasicLogicalPlanVisitor<StringBuilder, LogicalNode> {
+ private static final JoinOrderStringBuilder instance;
+ static {
+ instance = new JoinOrderStringBuilder();
+ }
+
+ public static JoinOrderStringBuilder getInstance() {
+ return instance;
+ }
+
+ public static String buildJoinOrderString(LogicalPlan plan, LogicalPlan.QueryBlock block) throws PlanningException {
+ StringBuilder originalOrder = new StringBuilder();
+ instance.visit(originalOrder, plan, block);
+ return originalOrder.toString();
+ }
+
+ @Override
+ public LogicalNode visitJoin(StringBuilder sb, LogicalPlan plan, LogicalPlan.QueryBlock block, JoinNode joinNode,
+ Stack<LogicalNode> stack)
+ throws PlanningException {
+ stack.push(joinNode);
+ sb.append("(");
+ visit(sb, plan, block, joinNode.getLeftChild(), stack);
+ sb.append(" ").append(getJoinNotation(joinNode.getJoinType())).append(" ");
+ visit(sb, plan, block, joinNode.getRightChild(), stack);
+ sb.append(")");
+ stack.pop();
+ return joinNode;
+ }
+
+ private static String getJoinNotation(JoinType joinType) {
+ switch (joinType) {
+ case CROSS: return "⋈";
+ case INNER: return "⋈θ";
+ case LEFT_OUTER: return "⟕";
+ case RIGHT_OUTER: return "⟖";
+ case FULL_OUTER: return "⟗";
+ case LEFT_SEMI: return "⋉";
+ case RIGHT_SEMI: return "⋊";
+ case LEFT_ANTI: return "▷";
+ }
+ return ",";
+ }
+
+ @Override
+ public LogicalNode visitTableSubQuery(StringBuilder sb, LogicalPlan plan, LogicalPlan.QueryBlock block,
+ TableSubQueryNode node, Stack<LogicalNode> stack) {
+ sb.append(node.getTableName());
+ return node;
+ }
+
+ public LogicalNode visitScan(StringBuilder sb, LogicalPlan plan, LogicalPlan.QueryBlock block, ScanNode node,
+ Stack<LogicalNode> stack) {
+ sb.append(node.getTableName());
+ return node;
+ }
+ }
+
+ private static class CostContext {
+ double accumulatedCost = 0;
+ }
+
+ public static class JoinCostComputer extends BasicLogicalPlanVisitor<CostContext, LogicalNode> {
+ private static final JoinCostComputer instance;
+
+ static {
+ instance = new JoinCostComputer();
+ }
+
+ public static double computeCost(LogicalPlan plan, LogicalPlan.QueryBlock block) throws PlanningException {
+ CostContext costContext = new CostContext();
+ instance.visit(costContext, plan, block);
+ return costContext.accumulatedCost;
+ }
+
+ @Override
+ public LogicalNode visitJoin(CostContext joinGraphContext, LogicalPlan plan, LogicalPlan.QueryBlock block,
+ JoinNode joinNode, Stack<LogicalNode> stack)
+ throws PlanningException {
+ super.visitJoin(joinGraphContext, plan, block, joinNode, stack);
+
+ double filterFactor = 1;
+ if (joinNode.hasJoinQual()) {
+ EvalNode [] quals = AlgebraicUtil.toConjunctiveNormalFormArray(joinNode.getJoinQual());
+ filterFactor = Math.pow(GreedyHeuristicJoinOrderAlgorithm.DEFAULT_SELECTION_FACTOR, quals.length);
+ }
+
+ if (joinNode.getLeftChild() instanceof RelationNode) {
+ joinGraphContext.accumulatedCost = getCost(joinNode.getLeftChild()) * getCost(joinNode.getRightChild())
+ * filterFactor;
+ } else {
+ joinGraphContext.accumulatedCost = joinGraphContext.accumulatedCost +
+ (joinGraphContext.accumulatedCost * getCost(joinNode.getRightChild()) * filterFactor);
+ }
+
+ return joinNode;
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/b143f991/tajo-plan/src/main/java/org/apache/tajo/plan/LogicalPlan.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/LogicalPlan.java b/tajo-plan/src/main/java/org/apache/tajo/plan/LogicalPlan.java
new file mode 100644
index 0000000..3baf61d
--- /dev/null
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/LogicalPlan.java
@@ -0,0 +1,683 @@
+/*
+ * 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.tajo.plan;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import org.apache.commons.lang.ObjectUtils;
+import org.apache.tajo.algebra.*;
+import org.apache.tajo.annotation.NotThreadSafe;
+import org.apache.tajo.catalog.Column;
+import org.apache.tajo.catalog.Schema;
+import org.apache.tajo.util.graph.DirectedGraphCursor;
+import org.apache.tajo.util.graph.SimpleDirectedGraph;
+import org.apache.tajo.plan.expr.ConstEval;
+import org.apache.tajo.plan.expr.EvalNode;
+import org.apache.tajo.plan.logical.LogicalNode;
+import org.apache.tajo.plan.logical.LogicalRootNode;
+import org.apache.tajo.plan.logical.NodeType;
+import org.apache.tajo.plan.logical.RelationNode;
+import org.apache.tajo.plan.nameresolver.NameResolver;
+import org.apache.tajo.plan.nameresolver.NameResolvingMode;
+import org.apache.tajo.plan.visitor.ExplainLogicalPlanVisitor;
+import org.apache.tajo.util.TUtil;
+
+import java.lang.reflect.Constructor;
+import java.util.*;
+
+/**
+ * This represents and keeps every information about a query plan for a query.
+ */
+@NotThreadSafe
+public class LogicalPlan {
+ /** the prefix character for virtual tables */
+ public static final char VIRTUAL_TABLE_PREFIX='#';
+ public static final char NONAMED_COLUMN_PREFIX='?';
+
+ /** it indicates the root block */
+ public static final String ROOT_BLOCK = VIRTUAL_TABLE_PREFIX + "ROOT";
+ public static final String NONAME_BLOCK_PREFIX = VIRTUAL_TABLE_PREFIX + "QB_";
+ private static final int NO_SEQUENCE_PID = -1;
+ private int nextPid = 0;
+ private Integer noNameBlockId = 0;
+ private Integer noNameColumnId = 0;
+
+ /** a map from between a block name to a block plan */
+ private Map<String, QueryBlock> queryBlocks = new LinkedHashMap<String, QueryBlock>();
+ private Map<Integer, LogicalNode> nodeMap = new HashMap<Integer, LogicalNode>();
+ private Map<Integer, QueryBlock> queryBlockByPID = new HashMap<Integer, QueryBlock>();
+ private Map<String, String> exprToBlockNameMap = TUtil.newHashMap();
+ private SimpleDirectedGraph<String, BlockEdge> queryBlockGraph = new SimpleDirectedGraph<String, BlockEdge>();
+
+ /** planning and optimization log */
+ private List<String> planingHistory = Lists.newArrayList();
+ LogicalPlanner planner;
+
+ private boolean isExplain;
+
+ public LogicalPlan(LogicalPlanner planner) {
+ this.planner = planner;
+ }
+
+ /**
+ * Create a LogicalNode instance for a type. Each a LogicalNode instance is given an unique plan node id (PID).
+ *
+ * @param theClass The class to be created
+ * @return a LogicalNode instance identified by an unique plan node id (PID).
+ */
+ public <T extends LogicalNode> T createNode(Class<T> theClass) {
+ try {
+ Constructor<T> ctor = theClass.getConstructor(int.class);
+ return ctor.newInstance(newPID());
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ /**
+ * Create a LogicalNode instance for a type. Each a LogicalNode instance is not given an unique plan node id (PID).
+ * This method must be only used after all query planning and optimization phases.
+ *
+ * @param theClass The class to be created
+ * @return a LogicalNode instance
+ */
+ public static <T extends LogicalNode> T createNodeWithoutPID(Class<T> theClass) {
+ try {
+ Constructor<T> ctor = theClass.getConstructor(int.class);
+ return ctor.newInstance(NO_SEQUENCE_PID);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ public void setExplain() {
+ isExplain = true;
+ }
+
+ public boolean isExplain() {
+ return isExplain;
+ }
+
+ /**
+ * Create a new {@link QueryBlock} and Get
+ *
+ * @param blockName the query block name
+ * @return a created query block
+ */
+ public QueryBlock newAndGetBlock(String blockName) {
+ QueryBlock block = new QueryBlock(blockName);
+ queryBlocks.put(blockName, block);
+ return block;
+ }
+
+ public int newPID() {
+ return nextPid++;
+ }
+
+ public QueryBlock newQueryBlock() {
+ return newAndGetBlock(NONAME_BLOCK_PREFIX + (noNameBlockId++));
+ }
+
+ public void resetGeneratedId() {
+ noNameColumnId = 0;
+ }
+
+ /**
+ * It generates an unique column name from EvalNode. It is usually used for an expression or predicate without
+ * a specified name (i.e., alias).
+ */
+ public String generateUniqueColumnName(EvalNode evalNode) {
+ String prefix = evalNode.getName();
+ return attachSeqIdToGeneratedColumnName(prefix).toLowerCase();
+ }
+
+ /**
+ * It generates an unique column name from Expr. It is usually used for an expression or predicate without
+ * a specified name (i.e., alias).
+ */
+ public String generateUniqueColumnName(Expr expr) {
+ String generatedName;
+ if (expr.getType() == OpType.Column) {
+ generatedName = ((ColumnReferenceExpr) expr).getCanonicalName();
+ } else { // if a generated column name
+ generatedName = attachSeqIdToGeneratedColumnName(getGeneratedPrefixFromExpr(expr));
+ }
+ return generatedName;
+ }
+
+ /**
+ * It attaches a generated column name with a sequence id. It always keeps generated names unique.
+ */
+ private String attachSeqIdToGeneratedColumnName(String prefix) {
+ int sequence = noNameColumnId++;
+ return NONAMED_COLUMN_PREFIX + prefix.toLowerCase() + (sequence > 0 ? "_" + sequence : "");
+ }
+
+ /**
+ * It generates a column reference prefix name. It is usually used for an expression or predicate without
+ * a specified name (i.e., alias). For example, a predicate in WHERE does not have any alias name.
+ * It just returns a prefix name. In other words, it always returns the same prefix for the same type of expressions.
+ * So, you must add some suffix to the returned name in order to distinguish reference names.
+ */
+ private static String getGeneratedPrefixFromExpr(Expr expr) {
+ String prefix;
+
+ switch (expr.getType()) {
+ case Column:
+ prefix = ((ColumnReferenceExpr) expr).getCanonicalName();
+ break;
+ case CountRowsFunction:
+ prefix = "count";
+ break;
+ case GeneralSetFunction:
+ GeneralSetFunctionExpr setFunction = (GeneralSetFunctionExpr) expr;
+ prefix = setFunction.getSignature();
+ break;
+ case Function:
+ FunctionExpr function = (FunctionExpr) expr;
+ prefix = function.getSignature();
+ break;
+ case Literal:
+ LiteralValue literal = (LiteralValue) expr;
+ switch (literal.getValueType()) {
+ case Boolean:
+ prefix = "bool";
+ break;
+ case String:
+ prefix = "text";
+ break;
+ case Unsigned_Integer:
+ case Unsigned_Large_Integer:
+ prefix = "number";
+ break;
+ case Unsigned_Float:
+ prefix = "real";
+ break;
+ default:
+ throw new IllegalStateException(literal.getValueType() + " is not implemented");
+ }
+ break;
+ case DateLiteral:
+ prefix = "date";
+ break;
+ case TimeLiteral:
+ prefix = "time";
+ break;
+ case TimestampLiteral:
+ prefix = "timestamp";
+ break;
+ default:
+ prefix = expr.getType().name();
+ }
+ return prefix;
+ }
+
+ public QueryBlock getRootBlock() {
+ return queryBlocks.get(ROOT_BLOCK);
+ }
+
+ public QueryBlock getBlock(String blockName) {
+ return queryBlocks.get(blockName);
+ }
+
+ public QueryBlock getBlock(LogicalNode node) {
+ return queryBlockByPID.get(node.getPID());
+ }
+
+ public void removeBlock(QueryBlock block) {
+ queryBlocks.remove(block.getName());
+ List<Integer> tobeRemoved = new ArrayList<Integer>();
+ for (Map.Entry<Integer, QueryBlock> entry : queryBlockByPID.entrySet()) {
+ tobeRemoved.add(entry.getKey());
+ }
+ for (Integer rn : tobeRemoved) {
+ queryBlockByPID.remove(rn);
+ }
+ }
+
+ public void disconnectBlocks(QueryBlock srcBlock, QueryBlock targetBlock) {
+ queryBlockGraph.removeEdge(srcBlock.getName(), targetBlock.getName());
+ }
+
+ public void connectBlocks(QueryBlock srcBlock, QueryBlock targetBlock, BlockType type) {
+ queryBlockGraph.addEdge(srcBlock.getName(), targetBlock.getName(), new BlockEdge(srcBlock, targetBlock, type));
+ }
+
+ public QueryBlock getParentBlock(QueryBlock block) {
+ return queryBlocks.get(queryBlockGraph.getParent(block.getName(), 0));
+ }
+
+ public List<QueryBlock> getChildBlocks(QueryBlock block) {
+ List<QueryBlock> childBlocks = TUtil.newList();
+ for (String blockName : queryBlockGraph.getChilds(block.getName())) {
+ childBlocks.add(queryBlocks.get(blockName));
+ }
+ return childBlocks;
+ }
+
+ public void mapExprToBlock(Expr expr, String blockName) {
+ exprToBlockNameMap.put(ObjectUtils.identityToString(expr), blockName);
+ }
+
+ public QueryBlock getBlockByExpr(Expr expr) {
+ return getBlock(exprToBlockNameMap.get(ObjectUtils.identityToString(expr)));
+ }
+
+ public String getBlockNameByExpr(Expr expr) {
+ return exprToBlockNameMap.get(ObjectUtils.identityToString(expr));
+ }
+
+ public Collection<QueryBlock> getQueryBlocks() {
+ return queryBlocks.values();
+ }
+
+ public SimpleDirectedGraph<String, BlockEdge> getQueryBlockGraph() {
+ return queryBlockGraph;
+ }
+
+ public Column resolveColumn(QueryBlock block, ColumnReferenceExpr columnRef) throws PlanningException {
+ return NameResolver.resolve(this, block, columnRef, NameResolvingMode.LEGACY);
+ }
+
+ public String getQueryGraphAsString() {
+ StringBuilder sb = new StringBuilder();
+
+ sb.append("\n-----------------------------\n");
+ sb.append("Query Block Graph\n");
+ sb.append("-----------------------------\n");
+ sb.append(queryBlockGraph.toStringGraph(getRootBlock().getName()));
+ sb.append("-----------------------------\n");
+ sb.append("Optimization Log:\n");
+ if (!planingHistory.isEmpty()) {
+ sb.append("[LogicalPlan]\n");
+ for (String eachHistory: planingHistory) {
+ sb.append("\t> ").append(eachHistory).append("\n");
+ }
+ }
+ DirectedGraphCursor<String, BlockEdge> cursor =
+ new DirectedGraphCursor<String, BlockEdge>(queryBlockGraph, getRootBlock().getName());
+ while(cursor.hasNext()) {
+ QueryBlock block = getBlock(cursor.nextBlock());
+ if (block.getPlanHistory().size() > 0) {
+ sb.append("[").append(block.getName()).append("]\n");
+ for (String log : block.getPlanHistory()) {
+ sb.append("\t> ").append(log).append("\n");
+ }
+ }
+ }
+ sb.append("-----------------------------\n");
+ sb.append("\n");
+
+ sb.append(getLogicalPlanAsString());
+
+ return sb.toString();
+ }
+
+ public String getLogicalPlanAsString() {
+ ExplainLogicalPlanVisitor explain = new ExplainLogicalPlanVisitor();
+
+ StringBuilder explains = new StringBuilder();
+ try {
+ ExplainLogicalPlanVisitor.Context explainContext = explain.getBlockPlanStrings(this, getRootBlock().getRoot());
+ while(!explainContext.explains.empty()) {
+ explains.append(
+ ExplainLogicalPlanVisitor.printDepthString(explainContext.getMaxDepth(), explainContext.explains.pop()));
+ }
+ } catch (PlanningException e) {
+ throw new RuntimeException(e);
+ }
+
+ return explains.toString();
+ }
+
+ public void addHistory(String string) {
+ planingHistory.add(string);
+ }
+
+ public List<String> getHistory() {
+ return planingHistory;
+ }
+
+ @Override
+ public String toString() {
+ return getQueryGraphAsString();
+ }
+
+ ///////////////////////////////////////////////////////////////////////////
+ // Query Block
+ ///////////////////////////////////////////////////////////////////////////
+
+ public static enum BlockType {
+ TableSubQuery,
+ ScalarSubQuery
+ }
+
+ public static class BlockEdge {
+ private String childName;
+ private String parentName;
+ private BlockType blockType;
+
+
+ public BlockEdge(String childName, String parentName, BlockType blockType) {
+ this.childName = childName;
+ this.parentName = parentName;
+ this.blockType = blockType;
+ }
+
+ public BlockEdge(QueryBlock child, QueryBlock parent, BlockType blockType) {
+ this(child.getName(), parent.getName(), blockType);
+ }
+
+ public String getParentName() {
+ return parentName;
+ }
+
+ public String getChildName() {
+ return childName;
+ }
+
+ public BlockType getBlockType() {
+ return blockType;
+ }
+ }
+
+ public class QueryBlock {
+ private final String blockName;
+ private LogicalNode rootNode;
+ private NodeType rootType;
+
+ // transient states
+ private final Map<String, RelationNode> canonicalNameToRelationMap = TUtil.newHashMap();
+ private final Map<String, List<String>> relationAliasMap = TUtil.newHashMap();
+ private final Map<String, String> columnAliasMap = TUtil.newHashMap();
+ private final Map<OpType, List<Expr>> operatorToExprMap = TUtil.newHashMap();
+ private final List<RelationNode> relationList = TUtil.newList();
+ private boolean hasWindowFunction = false;
+ private final Map<String, ConstEval> constantPoolByRef = Maps.newHashMap();
+ private final Map<Expr, String> constantPool = Maps.newHashMap();
+
+ /**
+ * It's a map between nodetype and node. node types can be duplicated. So, latest node type is only kept.
+ */
+ private final Map<NodeType, LogicalNode> nodeTypeToNodeMap = TUtil.newHashMap();
+ private final Map<String, LogicalNode> exprToNodeMap = TUtil.newHashMap();
+ final NamedExprsManager namedExprsMgr;
+
+ private LogicalNode currentNode;
+ private LogicalNode latestNode;
+ private final Set<JoinType> includedJoinTypes = TUtil.newHashSet();
+ /**
+ * Set true value if this query block has either implicit or explicit aggregation.
+ */
+ private boolean aggregationRequired = false;
+ private Schema schema;
+
+ /** It contains a planning log for this block */
+ private final List<String> planingHistory = Lists.newArrayList();
+ /** It is for debugging or unit tests */
+ private Target[] rawTargets;
+
+ public QueryBlock(String blockName) {
+ this.blockName = blockName;
+ this.namedExprsMgr = new NamedExprsManager(LogicalPlan.this, this);
+ }
+
+ public String getName() {
+ return blockName;
+ }
+
+ public void refresh() {
+ setRoot(rootNode);
+ }
+
+ public void setRoot(LogicalNode blockRoot) {
+ this.rootNode = blockRoot;
+ if (blockRoot instanceof LogicalRootNode) {
+ LogicalRootNode rootNode = (LogicalRootNode) blockRoot;
+ rootType = rootNode.getChild().getType();
+ }
+ }
+
+ public <NODE extends LogicalNode> NODE getRoot() {
+ return (NODE) rootNode;
+ }
+
+ public NodeType getRootType() {
+ return rootType;
+ }
+
+ public Target [] getRawTargets() {
+ return rawTargets;
+ }
+
+ public void setRawTargets(Target[] rawTargets) {
+ this.rawTargets = rawTargets;
+ }
+
+ public boolean existsRelation(String name) {
+ return canonicalNameToRelationMap.containsKey(name);
+ }
+
+ public boolean isAlreadyRenamedTableName(String name) {
+ return relationAliasMap.containsKey(name);
+ }
+
+ public RelationNode getRelation(String name) {
+ if (canonicalNameToRelationMap.containsKey(name)) {
+ return canonicalNameToRelationMap.get(name);
+ }
+
+ if (relationAliasMap.containsKey(name)) {
+ return canonicalNameToRelationMap.get(relationAliasMap.get(name).get(0));
+ }
+
+ return null;
+ }
+
+ public void addRelation(RelationNode relation) {
+ if (relation.hasAlias()) {
+ TUtil.putToNestedList(relationAliasMap, relation.getTableName(), relation.getCanonicalName());
+ }
+ canonicalNameToRelationMap.put(relation.getCanonicalName(), relation);
+ relationList.add(relation);
+ }
+
+ public Collection<RelationNode> getRelations() {
+ return Collections.unmodifiableList(relationList);
+ }
+
+ public boolean hasTableExpression() {
+ return this.canonicalNameToRelationMap.size() > 0;
+ }
+
+ public void addConstReference(String refName, Expr expr, ConstEval value) {
+ constantPoolByRef.put(refName, value);
+ constantPool.put(expr, refName);
+ }
+
+ public boolean isConstReference(String refName) {
+ return constantPoolByRef.containsKey(refName);
+ }
+
+ public boolean isRegisteredConst(Expr expr) {
+ return constantPool.containsKey(expr);
+ }
+
+ public String getConstReference(Expr expr) {
+ return constantPool.get(expr);
+ }
+
+ public ConstEval getConstByReference(String refName) {
+ return constantPoolByRef.get(refName);
+ }
+
+ public void addColumnAlias(String original, String alias) {
+ columnAliasMap.put(alias, original);
+ }
+
+ public boolean isAliasedName(String alias) {
+ return columnAliasMap.containsKey(alias);
+ }
+
+ public String getOriginalName(String alias) {
+ return columnAliasMap.get(alias);
+ }
+
+ public void setSchema(Schema schema) {
+ this.schema = schema;
+ }
+
+ public Schema getSchema() {
+ return schema;
+ }
+
+ public NamedExprsManager getNamedExprsManager() {
+ return namedExprsMgr;
+ }
+
+ public void updateCurrentNode(Expr expr) throws PlanningException {
+
+ if (expr.getType() != OpType.RelationList) { // skip relation list because it is a virtual expr.
+ this.currentNode = exprToNodeMap.get(ObjectUtils.identityToString(expr));
+ if (currentNode == null) {
+ throw new PlanningException("Unregistered Algebra Expression: " + expr.getType());
+ }
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ public <T extends LogicalNode> T getCurrentNode() {
+ return (T) this.currentNode;
+ }
+
+ public void updateLatestNode(LogicalNode node) {
+ this.latestNode = node;
+ }
+
+ public <T extends LogicalNode> T getLatestNode() {
+ return (T) this.latestNode;
+ }
+
+ public void setAlgebraicExpr(Expr expr) {
+ TUtil.putToNestedList(operatorToExprMap, expr.getType(), expr);
+ }
+
+ public boolean hasAlgebraicExpr(OpType opType) {
+ return operatorToExprMap.containsKey(opType);
+ }
+
+ public <T extends Expr> List<T> getAlgebraicExpr(OpType opType) {
+ return (List<T>) operatorToExprMap.get(opType);
+ }
+
+ public <T extends Expr> T getSingletonExpr(OpType opType) {
+ if (hasAlgebraicExpr(opType)) {
+ return (T) operatorToExprMap.get(opType).get(0);
+ } else {
+ return null;
+ }
+ }
+
+ public boolean hasNode(NodeType nodeType) {
+ return nodeTypeToNodeMap.containsKey(nodeType);
+ }
+
+ public void registerNode(LogicalNode node) {
+ // id -> node
+ nodeMap.put(node.getPID(), node);
+
+ // So, this is only for filter, groupby, sort, limit, projection, which exists once at a query block.
+ nodeTypeToNodeMap.put(node.getType(), node);
+
+ queryBlockByPID.put(node.getPID(), this);
+ }
+
+ public void unregisterNode(LogicalNode node) {
+ nodeMap.remove(node.getPID());
+ nodeTypeToNodeMap.remove(node.getType());
+ queryBlockByPID.remove(node.getPID());
+ }
+
+ @SuppressWarnings("unchecked")
+ public <T extends LogicalNode> T getNode(NodeType nodeType) {
+ return (T) nodeTypeToNodeMap.get(nodeType);
+ }
+
+ // expr -> node
+ public void registerExprWithNode(Expr expr, LogicalNode node) {
+ exprToNodeMap.put(ObjectUtils.identityToString(expr), node);
+ }
+
+ public <T extends LogicalNode> T getNodeFromExpr(Expr expr) {
+ return (T) exprToNodeMap.get(ObjectUtils.identityToString(expr));
+ }
+
+ public void setHasWindowFunction() {
+ hasWindowFunction = true;
+ }
+
+ public boolean hasWindowSpecs() {
+ return hasWindowFunction;
+ }
+
+ /**
+ * This flag can be changed as a plan is generated.
+ *
+ * True value means that this query should have aggregation phase. If aggregation plan is added to this block,
+ * it becomes false because it doesn't need aggregation phase anymore. It is usually used to add aggregation
+ * phase from SELECT statement without group-by clause.
+ *
+ * @return True if aggregation is needed but this query hasn't had aggregation phase.
+ */
+ public boolean isAggregationRequired() {
+ return this.aggregationRequired;
+ }
+
+ /**
+ * Unset aggregation required flag. It has to be called after an aggregation phase is added to this block.
+ */
+ public void unsetAggregationRequire() {
+ this.aggregationRequired = false;
+ }
+
+ public void setAggregationRequire() {
+ aggregationRequired = true;
+ }
+
+ public boolean containsJoinType(JoinType joinType) {
+ return includedJoinTypes.contains(joinType);
+ }
+
+ public void addJoinType(JoinType joinType) {
+ includedJoinTypes.add(joinType);
+ }
+
+ public List<String> getPlanHistory() {
+ return planingHistory;
+ }
+
+ public void addPlanHistory(String history) {
+ this.planingHistory.add(history);
+ }
+
+ public String toString() {
+ return blockName;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/b143f991/tajo-plan/src/main/java/org/apache/tajo/plan/LogicalPlanPreprocessor.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/LogicalPlanPreprocessor.java b/tajo-plan/src/main/java/org/apache/tajo/plan/LogicalPlanPreprocessor.java
new file mode 100644
index 0000000..6bdfb9d
--- /dev/null
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/LogicalPlanPreprocessor.java
@@ -0,0 +1,507 @@
+/**
+ * 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.tajo.plan;
+
+import org.apache.tajo.SessionVars;
+import org.apache.tajo.algebra.*;
+import org.apache.tajo.catalog.*;
+import org.apache.tajo.catalog.exception.NoSuchColumnException;
+import org.apache.tajo.common.TajoDataTypes;
+import org.apache.tajo.plan.LogicalPlan.QueryBlock;
+import org.apache.tajo.plan.algebra.BaseAlgebraVisitor;
+import org.apache.tajo.plan.expr.ConstEval;
+import org.apache.tajo.plan.expr.EvalNode;
+import org.apache.tajo.plan.expr.FieldEval;
+import org.apache.tajo.plan.logical.*;
+import org.apache.tajo.plan.nameresolver.NameResolver;
+import org.apache.tajo.plan.nameresolver.NameResolvingMode;
+import org.apache.tajo.plan.util.PlannerUtil;
+import org.apache.tajo.plan.util.SchemaUtil;
+import org.apache.tajo.plan.visitor.SimpleAlgebraVisitor;
+import org.apache.tajo.util.TUtil;
+
+import java.util.*;
+
+/**
+ * It finds all relations for each block and builds base schema information.
+ */
+public class LogicalPlanPreprocessor extends BaseAlgebraVisitor<LogicalPlanner.PlanContext, LogicalNode> {
+ private TypeDeterminant typeDeterminant;
+ private ExprAnnotator annotator;
+
+ /** Catalog service */
+ private CatalogService catalog;
+
+ LogicalPlanPreprocessor(CatalogService catalog, ExprAnnotator annotator) {
+ this.catalog = catalog;
+ this.annotator = annotator;
+ this.typeDeterminant = new TypeDeterminant(catalog);
+ }
+
+ @Override
+ public void preHook(LogicalPlanner.PlanContext ctx, Stack<Expr> stack, Expr expr) throws PlanningException {
+ ctx.queryBlock.setAlgebraicExpr(expr);
+ ctx.plan.mapExprToBlock(expr, ctx.queryBlock.getName());
+ }
+
+ @Override
+ public LogicalNode postHook(LogicalPlanner.PlanContext ctx, Stack<Expr> stack, Expr expr, LogicalNode result)
+ throws PlanningException {
+ // If non-from statement, result can be null. It avoids that case.
+ if (result != null) {
+ // setNode method registers each node to corresponding block and plan.
+ ctx.queryBlock.registerNode(result);
+ // It makes a map between an expr and a logical node.
+ ctx.queryBlock.registerExprWithNode(expr, result);
+ }
+ return result;
+ }
+
+ /**
+ * Get all columns of the relations correspondent to the asterisk expression.
+ * @param ctx
+ * @param asteriskExpr
+ * @return array of columns
+ * @throws PlanningException
+ */
+ public static Column[] getColumns(LogicalPlanner.PlanContext ctx, QualifiedAsteriskExpr asteriskExpr)
+ throws PlanningException {
+ RelationNode relationOp = null;
+ QueryBlock block = ctx.queryBlock;
+ Collection<QueryBlock> queryBlocks = ctx.plan.getQueryBlocks();
+ if (asteriskExpr.hasQualifier()) {
+ String qualifier;
+
+ if (CatalogUtil.isFQTableName(asteriskExpr.getQualifier())) {
+ qualifier = asteriskExpr.getQualifier();
+ } else {
+ qualifier = CatalogUtil.buildFQName(
+ ctx.queryContext.get(SessionVars.CURRENT_DATABASE), asteriskExpr.getQualifier());
+ }
+
+ relationOp = block.getRelation(qualifier);
+
+ // if a column name is outside of this query block
+ if (relationOp == null) {
+ // TODO - nested query can only refer outer query block? or not?
+ for (QueryBlock eachBlock : queryBlocks) {
+ if (eachBlock.existsRelation(qualifier)) {
+ relationOp = eachBlock.getRelation(qualifier);
+ }
+ }
+ }
+
+ // If we cannot find any relation against a qualified column name
+ if (relationOp == null) {
+ throw new NoSuchColumnException(CatalogUtil.buildFQName(qualifier, "*"));
+ }
+
+ Schema schema = relationOp.getTableSchema();
+ Column[] resolvedColumns = new Column[schema.size()];
+ return schema.getColumns().toArray(resolvedColumns);
+ } else { // if a column reference is not qualified
+ // columns of every relation should be resolved.
+ Iterator<RelationNode> iterator = block.getRelations().iterator();
+ Schema schema;
+ List<Column> resolvedColumns = TUtil.newList();
+
+ while (iterator.hasNext()) {
+ relationOp = iterator.next();
+ schema = relationOp.getTableSchema();
+ resolvedColumns.addAll(schema.getColumns());
+ }
+
+ if (resolvedColumns.size() == 0) {
+ throw new NoSuchColumnException(asteriskExpr.toString());
+ }
+
+ return resolvedColumns.toArray(new Column[resolvedColumns.size()]);
+ }
+ }
+
+ /**
+ * Resolve an asterisk expression to the real column reference expressions.
+ * @param ctx context
+ * @param asteriskExpr asterisk expression
+ * @return a list of NamedExpr each of which has ColumnReferenceExprs as its child
+ * @throws PlanningException
+ */
+ private static List<NamedExpr> resolveAsterisk(LogicalPlanner.PlanContext ctx, QualifiedAsteriskExpr asteriskExpr)
+ throws PlanningException {
+ Column[] columns = getColumns(ctx, asteriskExpr);
+ List<NamedExpr> newTargetExprs = new ArrayList<NamedExpr>(columns.length);
+ int i;
+ for (i = 0; i < columns.length; i++) {
+ newTargetExprs.add(new NamedExpr(new ColumnReferenceExpr(columns[i].getQualifier(), columns[i].getSimpleName())));
+ }
+ return newTargetExprs;
+ }
+
+ private static boolean hasAsterisk(NamedExpr [] namedExprs) {
+ for (NamedExpr eachTarget : namedExprs) {
+ if (eachTarget.getExpr().getType() == OpType.Asterisk) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private static NamedExpr [] voidResolveAsteriskNamedExpr(LogicalPlanner.PlanContext context,
+ NamedExpr [] namedExprs) throws PlanningException {
+ List<NamedExpr> rewrittenTargets = TUtil.newList();
+ for (NamedExpr originTarget : namedExprs) {
+ if (originTarget.getExpr().getType() == OpType.Asterisk) {
+ // rewrite targets
+ rewrittenTargets.addAll(resolveAsterisk(context, (QualifiedAsteriskExpr) originTarget.getExpr()));
+ } else {
+ rewrittenTargets.add(originTarget);
+ }
+ }
+ return rewrittenTargets.toArray(new NamedExpr[rewrittenTargets.size()]);
+ }
+
+ @Override
+ public LogicalNode visitProjection(LogicalPlanner.PlanContext ctx, Stack<Expr> stack, Projection expr)
+ throws PlanningException {
+ // If Non-from statement, it immediately returns.
+ if (!expr.hasChild()) {
+ EvalExprNode exprNode = ctx.plan.createNode(EvalExprNode.class);
+ exprNode.setTargets(buildTargets(ctx, expr.getNamedExprs()));
+ return exprNode;
+ }
+
+ stack.push(expr); // <--- push
+ LogicalNode child = visit(ctx, stack, expr.getChild());
+
+ // Resolve the asterisk expression
+ if (hasAsterisk(expr.getNamedExprs())) {
+ expr.setNamedExprs(voidResolveAsteriskNamedExpr(ctx, expr.getNamedExprs()));
+ }
+
+ NamedExpr[] projectTargetExprs = expr.getNamedExprs();
+ for (int i = 0; i < expr.getNamedExprs().length; i++) {
+ NamedExpr namedExpr = projectTargetExprs[i];
+
+ // 1) Normalize all field names occured in each expr into full qualified names
+ NameRefInSelectListNormalizer.normalize(ctx, namedExpr.getExpr());
+
+ // 2) Register explicit column aliases to block
+ if (namedExpr.getExpr().getType() == OpType.Column && namedExpr.hasAlias()) {
+ ctx.queryBlock.addColumnAlias(((ColumnReferenceExpr)namedExpr.getExpr()).getCanonicalName(),
+ namedExpr.getAlias());
+ } else if (OpType.isLiteralType(namedExpr.getExpr().getType()) && namedExpr.hasAlias()) {
+ Expr constExpr = namedExpr.getExpr();
+ ConstEval constEval = (ConstEval) annotator.createEvalNode(ctx, constExpr, NameResolvingMode.RELS_ONLY);
+ ctx.queryBlock.addConstReference(namedExpr.getAlias(), constExpr, constEval);
+ }
+ }
+
+ Target[] targets = buildTargets(ctx, expr.getNamedExprs());
+
+ stack.pop(); // <--- Pop
+
+ ProjectionNode projectionNode = ctx.plan.createNode(ProjectionNode.class);
+ projectionNode.setInSchema(child.getOutSchema());
+ projectionNode.setOutSchema(PlannerUtil.targetToSchema(targets));
+
+ ctx.queryBlock.setSchema(projectionNode.getOutSchema());
+ return projectionNode;
+ }
+
+ private Target [] buildTargets(LogicalPlanner.PlanContext context, NamedExpr [] exprs) throws PlanningException {
+ Target [] targets = new Target[exprs.length];
+ for (int i = 0; i < exprs.length; i++) {
+ NamedExpr namedExpr = exprs[i];
+ TajoDataTypes.DataType dataType = typeDeterminant.determineDataType(context, namedExpr.getExpr());
+
+ if (namedExpr.hasAlias()) {
+ targets[i] = new Target(new FieldEval(new Column(namedExpr.getAlias(), dataType)));
+ } else {
+ String generatedName = context.plan.generateUniqueColumnName(namedExpr.getExpr());
+ targets[i] = new Target(new FieldEval(new Column(generatedName, dataType)));
+ }
+ }
+ return targets;
+ }
+
+ @Override
+ public LogicalNode visitLimit(LogicalPlanner.PlanContext ctx, Stack<Expr> stack, Limit expr)
+ throws PlanningException {
+ stack.push(expr);
+ LogicalNode child = visit(ctx, stack, expr.getChild());
+ stack.pop();
+
+ LimitNode limitNode = ctx.plan.createNode(LimitNode.class);
+ limitNode.setInSchema(child.getOutSchema());
+ limitNode.setOutSchema(child.getOutSchema());
+ return limitNode;
+ }
+
+ @Override
+ public LogicalNode visitSort(LogicalPlanner.PlanContext ctx, Stack<Expr> stack, Sort expr) throws PlanningException {
+ stack.push(expr);
+ LogicalNode child = visit(ctx, stack, expr.getChild());
+ stack.pop();
+
+ SortNode sortNode = ctx.plan.createNode(SortNode.class);
+ sortNode.setInSchema(child.getOutSchema());
+ sortNode.setOutSchema(child.getOutSchema());
+ return sortNode;
+ }
+
+ @Override
+ public LogicalNode visitHaving(LogicalPlanner.PlanContext ctx, Stack<Expr> stack, Having expr)
+ throws PlanningException {
+ stack.push(expr);
+ LogicalNode child = visit(ctx, stack, expr.getChild());
+ stack.pop();
+
+ HavingNode havingNode = ctx.plan.createNode(HavingNode.class);
+ havingNode.setInSchema(child.getOutSchema());
+ havingNode.setOutSchema(child.getOutSchema());
+ return havingNode;
+ }
+
+ @Override
+ public LogicalNode visitGroupBy(LogicalPlanner.PlanContext ctx, Stack<Expr> stack, Aggregation expr)
+ throws PlanningException {
+ stack.push(expr); // <--- push
+ LogicalNode child = visit(ctx, stack, expr.getChild());
+
+ Projection projection = ctx.queryBlock.getSingletonExpr(OpType.Projection);
+ int finalTargetNum = projection.getNamedExprs().length;
+ Target [] targets = new Target[finalTargetNum];
+
+ if (hasAsterisk(projection.getNamedExprs())) {
+ projection.setNamedExprs(voidResolveAsteriskNamedExpr(ctx, projection.getNamedExprs()));
+ }
+
+ for (int i = 0; i < finalTargetNum; i++) {
+ NamedExpr namedExpr = projection.getNamedExprs()[i];
+ EvalNode evalNode = annotator.createEvalNode(ctx, namedExpr.getExpr(), NameResolvingMode.SUBEXPRS_AND_RELS);
+
+ if (namedExpr.hasAlias()) {
+ targets[i] = new Target(evalNode, namedExpr.getAlias());
+ } else {
+ targets[i] = new Target(evalNode, "?name_" + i);
+ }
+ }
+ stack.pop();
+
+ GroupbyNode groupByNode = ctx.plan.createNode(GroupbyNode.class);
+ groupByNode.setInSchema(child.getOutSchema());
+ groupByNode.setOutSchema(PlannerUtil.targetToSchema(targets));
+ return groupByNode;
+ }
+
+ @Override
+ public LogicalNode visitUnion(LogicalPlanner.PlanContext ctx, Stack<Expr> stack, SetOperation expr)
+ throws PlanningException {
+ LogicalPlan.QueryBlock leftBlock = ctx.plan.newQueryBlock();
+ LogicalPlanner.PlanContext leftContext = new LogicalPlanner.PlanContext(ctx, leftBlock);
+ LogicalNode leftChild = visit(leftContext, new Stack<Expr>(), expr.getLeft());
+ leftBlock.setRoot(leftChild);
+ ctx.queryBlock.registerExprWithNode(expr.getLeft(), leftChild);
+
+ LogicalPlan.QueryBlock rightBlock = ctx.plan.newQueryBlock();
+ LogicalPlanner.PlanContext rightContext = new LogicalPlanner.PlanContext(ctx, rightBlock);
+ LogicalNode rightChild = visit(rightContext, new Stack<Expr>(), expr.getRight());
+ rightBlock.setRoot(rightChild);
+ ctx.queryBlock.registerExprWithNode(expr.getRight(), rightChild);
+
+ UnionNode unionNode = new UnionNode(ctx.plan.newPID());
+ unionNode.setLeftChild(leftChild);
+ unionNode.setRightChild(rightChild);
+ unionNode.setInSchema(leftChild.getOutSchema());
+ unionNode.setOutSchema(leftChild.getOutSchema());
+
+ return unionNode;
+ }
+
+ public LogicalNode visitFilter(LogicalPlanner.PlanContext ctx, Stack<Expr> stack, Selection expr)
+ throws PlanningException {
+ stack.push(expr);
+ LogicalNode child = visit(ctx, stack, expr.getChild());
+ stack.pop();
+
+ SelectionNode selectionNode = ctx.plan.createNode(SelectionNode.class);
+ selectionNode.setInSchema(child.getOutSchema());
+ selectionNode.setOutSchema(child.getOutSchema());
+ return selectionNode;
+ }
+
+ @Override
+ public LogicalNode visitJoin(LogicalPlanner.PlanContext ctx, Stack<Expr> stack, Join expr) throws PlanningException {
+ stack.push(expr);
+ LogicalNode left = visit(ctx, stack, expr.getLeft());
+ LogicalNode right = visit(ctx, stack, expr.getRight());
+ stack.pop();
+ JoinNode joinNode = ctx.plan.createNode(JoinNode.class);
+ joinNode.setJoinType(expr.getJoinType());
+ Schema merged = SchemaUtil.merge(left.getOutSchema(), right.getOutSchema());
+ joinNode.setInSchema(merged);
+ joinNode.setOutSchema(merged);
+
+ ctx.queryBlock.addJoinType(expr.getJoinType());
+ return joinNode;
+ }
+
+ @Override
+ public LogicalNode visitRelation(LogicalPlanner.PlanContext ctx, Stack<Expr> stack, Relation expr)
+ throws PlanningException {
+ Relation relation = expr;
+
+ String actualRelationName;
+ if (CatalogUtil.isFQTableName(expr.getName())) {
+ actualRelationName = relation.getName();
+ } else {
+ actualRelationName =
+ CatalogUtil.buildFQName(ctx.queryContext.get(SessionVars.CURRENT_DATABASE), relation.getName());
+ }
+
+ TableDesc desc = catalog.getTableDesc(actualRelationName);
+ ScanNode scanNode = ctx.plan.createNode(ScanNode.class);
+ if (relation.hasAlias()) {
+ scanNode.init(desc, relation.getAlias());
+ } else {
+ scanNode.init(desc);
+ }
+ ctx.queryBlock.addRelation(scanNode);
+
+ return scanNode;
+ }
+
+ @Override
+ public LogicalNode visitTableSubQuery(LogicalPlanner.PlanContext ctx, Stack<Expr> stack, TablePrimarySubQuery expr)
+ throws PlanningException {
+
+ LogicalPlanner.PlanContext newContext;
+ // Note: TableSubQuery always has a table name.
+ // SELECT .... FROM (SELECT ...) TB_NAME <-
+ QueryBlock queryBlock = ctx.plan.newQueryBlock();
+ newContext = new LogicalPlanner.PlanContext(ctx, queryBlock);
+ LogicalNode child = super.visitTableSubQuery(newContext, stack, expr);
+ queryBlock.setRoot(child);
+
+ // a table subquery should be dealt as a relation.
+ TableSubQueryNode node = ctx.plan.createNode(TableSubQueryNode.class);
+ node.init(CatalogUtil.buildFQName(ctx.queryContext.get(SessionVars.CURRENT_DATABASE), expr.getName()), child);
+ ctx.queryBlock.addRelation(node);
+ return node;
+ }
+
+ ///////////////////////////////////////////////////////////////////////////////////////////////////////////
+ // Data Definition Language Section
+ ///////////////////////////////////////////////////////////////////////////////////////////////////////////
+
+ @Override
+ public LogicalNode visitCreateDatabase(LogicalPlanner.PlanContext ctx, Stack<Expr> stack, CreateDatabase expr)
+ throws PlanningException {
+ CreateDatabaseNode createDatabaseNode = ctx.plan.createNode(CreateDatabaseNode.class);
+ return createDatabaseNode;
+ }
+
+ @Override
+ public LogicalNode visitDropDatabase(LogicalPlanner.PlanContext ctx, Stack<Expr> stack, DropDatabase expr)
+ throws PlanningException {
+ DropDatabaseNode dropDatabaseNode = ctx.plan.createNode(DropDatabaseNode.class);
+ return dropDatabaseNode;
+ }
+
+ @Override
+ public LogicalNode visitCreateTable(LogicalPlanner.PlanContext ctx, Stack<Expr> stack, CreateTable expr)
+ throws PlanningException {
+
+ CreateTableNode createTableNode = ctx.plan.createNode(CreateTableNode.class);
+
+ if (expr.hasSubQuery()) {
+ stack.push(expr);
+ visit(ctx, stack, expr.getSubQuery());
+ stack.pop();
+ }
+
+ return createTableNode;
+ }
+
+ @Override
+ public LogicalNode visitDropTable(LogicalPlanner.PlanContext ctx, Stack<Expr> stack, DropTable expr)
+ throws PlanningException {
+ DropTableNode dropTable = ctx.plan.createNode(DropTableNode.class);
+ return dropTable;
+ }
+
+ @Override
+ public LogicalNode visitAlterTablespace(LogicalPlanner.PlanContext ctx, Stack<Expr> stack, AlterTablespace expr)
+ throws PlanningException {
+ AlterTablespaceNode alterTablespace = ctx.plan.createNode(AlterTablespaceNode.class);
+ return alterTablespace;
+ }
+
+ @Override
+ public LogicalNode visitAlterTable(LogicalPlanner.PlanContext ctx, Stack<Expr> stack, AlterTable expr)
+ throws PlanningException {
+ AlterTableNode alterTableNode = ctx.plan.createNode(AlterTableNode.class);
+ return alterTableNode;
+ }
+
+ @Override
+ public LogicalNode visitTruncateTable(LogicalPlanner.PlanContext ctx, Stack<Expr> stack, TruncateTable expr)
+ throws PlanningException {
+ TruncateTableNode truncateTableNode = ctx.plan.createNode(TruncateTableNode.class);
+ return truncateTableNode;
+ }
+
+ ///////////////////////////////////////////////////////////////////////////////////////////////////////////
+ // Insert or Update Section
+ ///////////////////////////////////////////////////////////////////////////////////////////////////////////
+
+ public LogicalNode visitInsert(LogicalPlanner.PlanContext ctx, Stack<Expr> stack, Insert expr)
+ throws PlanningException {
+ LogicalNode child = super.visitInsert(ctx, stack, expr);
+
+ InsertNode insertNode = new InsertNode(ctx.plan.newPID());
+ insertNode.setInSchema(child.getOutSchema());
+ insertNode.setOutSchema(child.getOutSchema());
+ return insertNode;
+ }
+
+ static class NameRefInSelectListNormalizer extends SimpleAlgebraVisitor<LogicalPlanner.PlanContext, Object> {
+ private static final NameRefInSelectListNormalizer instance;
+
+ static {
+ instance = new NameRefInSelectListNormalizer();
+ }
+
+ public static void normalize(LogicalPlanner.PlanContext context, Expr expr) throws PlanningException {
+ NameRefInSelectListNormalizer normalizer = new NameRefInSelectListNormalizer();
+ normalizer.visit(context,new Stack<Expr>(), expr);
+ }
+
+ @Override
+ public Expr visitColumnReference(LogicalPlanner.PlanContext ctx, Stack<Expr> stack, ColumnReferenceExpr expr)
+ throws PlanningException {
+
+ String normalized = NameResolver.resolve(ctx.plan, ctx.queryBlock, expr,
+ NameResolvingMode.RELS_ONLY).getQualifiedName();
+ expr.setName(normalized);
+
+ return expr;
+ }
+ }
+}