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 2013/07/02 16:16:10 UTC
[16/51] [partial] TAJO-22: The package prefix should be
org.apache.tajo. (DaeMyung Kang via hyunsik)
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/bc6359b8/tajo-core/tajo-core-backend/src/main/java/tajo/engine/planner/LogicalOptimizer.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/tajo/engine/planner/LogicalOptimizer.java b/tajo-core/tajo-core-backend/src/main/java/tajo/engine/planner/LogicalOptimizer.java
deleted file mode 100644
index d188dda..0000000
--- a/tajo-core/tajo-core-backend/src/main/java/tajo/engine/planner/LogicalOptimizer.java
+++ /dev/null
@@ -1,742 +0,0 @@
-/**
- * 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 tajo.engine.planner;
-
-import com.google.common.base.Preconditions;
-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.fs.Path;
-import tajo.catalog.Column;
-import tajo.catalog.Schema;
-import tajo.catalog.SortSpec;
-import tajo.catalog.TableMeta;
-import tajo.engine.eval.EvalNode;
-import tajo.engine.eval.EvalTreeUtil;
-import tajo.engine.eval.FieldEval;
-import tajo.engine.parser.CreateTableStmt;
-import tajo.engine.parser.ParseTree;
-import tajo.engine.parser.QueryBlock;
-import tajo.engine.parser.QueryBlock.Target;
-import tajo.engine.planner.logical.*;
-import tajo.engine.query.exception.InvalidQueryException;
-import tajo.engine.utils.SchemaUtil;
-import tajo.storage.StorageManager;
-import tajo.util.IndexUtil;
-
-import java.io.IOException;
-import java.util.*;
-
-/**
- * This class optimizes a logical plan corresponding to one query block.
- */
-public class LogicalOptimizer {
- private static Log LOG = LogFactory.getLog(LogicalOptimizer.class);
-
- private LogicalOptimizer() {
- }
-
- public static LogicalNode optimize(PlanningContext context, LogicalNode plan) {
- LogicalNode toBeOptimized;
-
- try {
- toBeOptimized = (LogicalNode) plan.clone();
- } catch (CloneNotSupportedException e) {
- LOG.error(e);
- throw new InvalidQueryException("Cannot clone: " + plan);
- }
-
- switch (context.getParseTree().getType()) {
- case SELECT:
- //case UNION: // TODO - to be implemented
- //case EXCEPT:
- //case INTERSECT:
- case CREATE_TABLE_AS:
- // if there are selection node
- if(PlannerUtil.findTopNode(plan, ExprType.SELECTION) != null) {
- pushSelection(context, toBeOptimized);
- }
-
- try {
- pushProjection(context, toBeOptimized);
- } catch (CloneNotSupportedException e) {
- throw new InvalidQueryException(e);
- }
-
- break;
- default:
- }
-
- return toBeOptimized;
- }
-
- public static LogicalNode pushIndex(LogicalNode plan , StorageManager sm) throws IOException {
- if(PlannerUtil.findTopNode(plan, ExprType.SCAN) == null) {
- return plan;
- }
- LogicalNode toBeOptimized;
- try {
- toBeOptimized = (LogicalNode) plan.clone();
- } catch (CloneNotSupportedException e) {
- LOG.error(e);
- throw new InvalidQueryException("Cannot clone: " + plan);
- }
-
- changeScanToIndexNode(null ,toBeOptimized , sm);
- return toBeOptimized;
-
- }
- private static void changeScanToIndexNode
- (LogicalNode parent , LogicalNode cur , StorageManager sm ) throws IOException {
- if( cur instanceof BinaryNode) {
- changeScanToIndexNode(cur , ((BinaryNode)cur).getOuterNode() , sm);
- changeScanToIndexNode(cur , ((BinaryNode)cur).getInnerNode() , sm);
- } else {
- switch(cur.getType()) {
- case CREATE_INDEX:
- return;
- case SCAN:
- ScanNode scan = (ScanNode)cur;
- EvalNode qual = scan.getQual();
- if(qual == null) {
- return;
- }else {
- String tableName = scan.getTableId();
- Path path = new Path(sm.getTablePath(tableName), "index");
-
- if(sm.getFileSystem().exists(path)) {
-
- TableMeta meta = sm.getTableMeta(path);
- IndexScanNode node;
- if ((node = IndexUtil.indexEval(scan, meta.getOptions())) == null) {
- return;
- }
- if( parent instanceof BinaryNode ) {
- if (scan.equals(((BinaryNode)parent).getOuterNode())) {
- ((BinaryNode)parent).setOuter(node);
- } else {
- ((BinaryNode)parent).setInner(node);
- }
- } else {
- ((UnaryNode)parent).setSubNode(node);
- }
- }
- return;
- }
- default:
- changeScanToIndexNode(cur , ((UnaryNode)cur).getSubNode() , sm);
- break;
- }
- }
- }
- /**
- * This method pushes down the selection into the appropriate sub
- * logical operators.
- * <br />
- *
- * There are three operators that can have search conditions.
- * Selection, Join, and GroupBy clause.
- * However, the search conditions of Join and GroupBy cannot be pushed down
- * into child operators because they can be used when the data layout change
- * caused by join and grouping relations.
- * <br />
- *
- * However, some of the search conditions of selection clause can be pushed
- * down into appropriate sub operators. Some comparison expressions on
- * multiple relations are actually join conditions, and other expression
- * on single relation can be used in a scan operator or an Index Scan
- * operator.
- *
- * @param ctx
- * @param plan
- */
- private static void pushSelection(PlanningContext ctx, LogicalNode plan) {
- SelectionNode selNode = (SelectionNode) PlannerUtil.findTopNode(plan,
- ExprType.SELECTION);
- Preconditions.checkNotNull(selNode);
-
- Stack<LogicalNode> stack = new Stack<LogicalNode>();
- EvalNode [] cnf = EvalTreeUtil.getConjNormalForm(selNode.getQual());
- pushSelectionRecursive(plan, Lists.newArrayList(cnf), stack);
- }
-
- private static void pushSelectionRecursive(LogicalNode plan,
- List<EvalNode> cnf, Stack<LogicalNode> stack) {
-
- switch(plan.getType()) {
-
- case SELECTION:
- SelectionNode selNode = (SelectionNode) plan;
- stack.push(selNode);
- pushSelectionRecursive(selNode.getSubNode(),cnf, stack);
- stack.pop();
-
- // remove the selection operator if there is no search condition
- // after selection push.
- if(cnf.size() == 0) {
- LogicalNode node = stack.peek();
- if (node instanceof UnaryNode) {
- UnaryNode unary = (UnaryNode) node;
- unary.setSubNode(selNode.getSubNode());
- } else {
- throw new InvalidQueryException("Unexpected Logical Query Plan");
- }
- }
- break;
- case JOIN:
- JoinNode join = (JoinNode) plan;
-
- LogicalNode outer = join.getOuterNode();
- LogicalNode inner = join.getInnerNode();
-
- pushSelectionRecursive(outer, cnf, stack);
- pushSelectionRecursive(inner, cnf, stack);
-
- List<EvalNode> matched = Lists.newArrayList();
- for (EvalNode eval : cnf) {
- if (canBeEvaluated(eval, plan)) {
- matched.add(eval);
- }
- }
-
- EvalNode qual = null;
- if (matched.size() > 1) {
- // merged into one eval tree
- qual = EvalTreeUtil.transformCNF2Singleton(
- matched.toArray(new EvalNode [matched.size()]));
- } else if (matched.size() == 1) {
- // if the number of matched expr is one
- qual = matched.get(0);
- }
-
- if (qual != null) {
- JoinNode joinNode = (JoinNode) plan;
- if (joinNode.hasJoinQual()) {
- EvalNode conjQual = EvalTreeUtil.
- transformCNF2Singleton(joinNode.getJoinQual(), qual);
- joinNode.setJoinQual(conjQual);
- } else {
- joinNode.setJoinQual(qual);
- }
- if (joinNode.getJoinType() == JoinType.CROSS_JOIN) {
- joinNode.setJoinType(JoinType.INNER);
- }
- cnf.removeAll(matched);
- }
-
- break;
-
- case SCAN:
- matched = Lists.newArrayList();
- for (EvalNode eval : cnf) {
- if (canBeEvaluated(eval, plan)) {
- matched.add(eval);
- }
- }
-
- qual = null;
- if (matched.size() > 1) {
- // merged into one eval tree
- qual = EvalTreeUtil.transformCNF2Singleton(
- matched.toArray(new EvalNode [matched.size()]));
- } else if (matched.size() == 1) {
- // if the number of matched expr is one
- qual = matched.get(0);
- }
-
- if (qual != null) { // if a matched qual exists
- ScanNode scanNode = (ScanNode) plan;
- scanNode.setQual(qual);
- }
-
- cnf.removeAll(matched);
- break;
-
- default:
- stack.push(plan);
- if (plan instanceof UnaryNode) {
- UnaryNode unary = (UnaryNode) plan;
- pushSelectionRecursive(unary.getSubNode(), cnf, stack);
- } else if (plan instanceof BinaryNode) {
- BinaryNode binary = (BinaryNode) plan;
- pushSelectionRecursive(binary.getOuterNode(), cnf, stack);
- pushSelectionRecursive(binary.getInnerNode(), cnf, stack);
- }
- stack.pop();
- break;
- }
- }
-
- public static boolean canBeEvaluated(EvalNode eval, LogicalNode node) {
- Set<Column> columnRefs = EvalTreeUtil.findDistinctRefColumns(eval);
-
- if (node.getType() == ExprType.JOIN) {
- JoinNode joinNode = (JoinNode) node;
- Set<String> tableIds = Sets.newHashSet();
- // getting distinct table references
- for (Column col : columnRefs) {
- if (!tableIds.contains(col.getTableName())) {
- tableIds.add(col.getTableName());
- }
- }
-
- // if the references only indicate two relation, the condition can be
- // pushed into a join operator.
- if (tableIds.size() != 2) {
- return false;
- }
-
- String [] outer = PlannerUtil.getLineage(joinNode.getOuterNode());
- String [] inner = PlannerUtil.getLineage(joinNode.getInnerNode());
-
- Set<String> o = Sets.newHashSet(outer);
- Set<String> i = Sets.newHashSet(inner);
- if (outer == null || inner == null) {
- throw new InvalidQueryException("ERROR: Unexpected logical plan");
- }
- Iterator<String> it = tableIds.iterator();
- if (o.contains(it.next()) && i.contains(it.next())) {
- return true;
- }
-
- it = tableIds.iterator();
- if (i.contains(it.next()) && o.contains(it.next())) {
- return true;
- }
-
- return false;
- } else {
- for (Column col : columnRefs) {
- if (!node.getInSchema().contains(col.getQualifiedName())) {
- return false;
- }
- }
-
- return true;
- }
- }
-
- /**
- * This method pushes down the projection list into the appropriate and
- * below logical operators.
- * @param context
- * @param plan
- */
- private static void pushProjection(PlanningContext context,
- LogicalNode plan)
- throws CloneNotSupportedException {
- Stack<LogicalNode> stack = new Stack<LogicalNode>();
-
- ParseTree tree = context.getParseTree();
- QueryBlock block;
-
- if (tree instanceof QueryBlock) {
- block = (QueryBlock) context.getParseTree();
-
- } else if (tree instanceof CreateTableStmt) {
-
- CreateTableStmt createTableStmt = (CreateTableStmt) tree;
-
- if (createTableStmt.hasQueryBlock()) {
- block = createTableStmt.getSelectStmt();
- } else {
- return;
- }
- } else {
-
- return;
- }
- OptimizationContext optCtx = new OptimizationContext(context,
- block.getTargetList());
-
- pushProjectionRecursive(context, optCtx, plan, stack, new HashSet<Column>());
- }
-
- /**
- * Groupby, Join, and Scan can project necessary columns.
- * This method has three roles:
- * 1) collect column reference necessary for sortkeys, join keys, selection conditions, grouping fields,
- * and having conditions
- * 2) shrink the output schema of each operator so that the operator reduces the output columns according to
- * the necessary columns of their parent operators
- * 3) shrink the input schema of each operator according to the shrunk output schemas of the child operators.
- *
- *
- * @param ctx
- * @param node
- * //@param necessary - columns necessary for above logical nodes, but it excepts the fields for the target list
- * //@param targetList
- * @return
- */
- private static LogicalNode pushProjectionRecursive(
- final PlanningContext ctx, final OptimizationContext optContext,
- final LogicalNode node, final Stack<LogicalNode> stack,
- final Set<Column> necessary) throws CloneNotSupportedException {
-
- LogicalNode outer;
- LogicalNode inner;
- switch (node.getType()) {
- case ROOT: // It does not support the projection
- LogicalRootNode root = (LogicalRootNode) node;
- stack.add(root);
- outer = pushProjectionRecursive(ctx, optContext,
- root.getSubNode(), stack, necessary);
- root.setInSchema(outer.getOutSchema());
- root.setOutSchema(outer.getOutSchema());
- break;
-
- case STORE:
- StoreTableNode store = (StoreTableNode) node;
- stack.add(store);
- outer = pushProjectionRecursive(ctx, optContext,
- store.getSubNode(), stack, necessary);
- store.setInSchema(outer.getOutSchema());
- store.setOutSchema(outer.getOutSchema());
- break;
-
- case PROJECTION:
- ProjectionNode projNode = (ProjectionNode) node;
-
- stack.add(projNode);
- outer = pushProjectionRecursive(ctx, optContext,
- projNode.getSubNode(), stack, necessary);
- stack.pop();
-
- LogicalNode childNode = projNode.getSubNode();
- if (optContext.getTargetListManager().isAllEvaluated() // if all exprs are evaluated
- && (childNode.getType() == ExprType.JOIN ||
- childNode.getType() == ExprType.GROUP_BY ||
- childNode.getType() == ExprType.SCAN)) { // if the child node is projectable
- projNode.getSubNode().setOutSchema(
- optContext.getTargetListManager().getUpdatedSchema());
- LogicalNode parent = stack.peek();
- ((UnaryNode)parent).setSubNode(projNode.getSubNode());
- return projNode.getSubNode();
- } else {
- // the output schema is not changed.
- projNode.setInSchema(outer.getOutSchema());
- projNode.setTargetList(
- optContext.getTargetListManager().getUpdatedTarget());
- }
- return projNode;
-
- case SELECTION: // It does not support the projection
- SelectionNode selNode = (SelectionNode) node;
-
- if (selNode.getQual() != null) {
- necessary.addAll(EvalTreeUtil.findDistinctRefColumns(selNode.getQual()));
- }
-
- stack.add(selNode);
- outer = pushProjectionRecursive(ctx, optContext, selNode.getSubNode(),
- stack, necessary);
- stack.pop();
- selNode.setInSchema(outer.getOutSchema());
- selNode.setOutSchema(outer.getOutSchema());
- break;
-
- case GROUP_BY: {
- GroupbyNode groupByNode = (GroupbyNode)node;
-
- if (groupByNode.hasHavingCondition()) {
- necessary.addAll(EvalTreeUtil.findDistinctRefColumns(groupByNode.getHavingCondition()));
- }
-
- stack.add(groupByNode);
- outer = pushProjectionRecursive(ctx, optContext,
- groupByNode.getSubNode(), stack, necessary);
- stack.pop();
- groupByNode.setInSchema(outer.getOutSchema());
- // set all targets
- groupByNode.setTargetList(optContext.getTargetListManager().getUpdatedTarget());
-
- TargetListManager targets = optContext.getTargetListManager();
- List<Target> groupbyPushable = Lists.newArrayList();
- List<Integer> groupbyPushableId = Lists.newArrayList();
-
- EvalNode expr;
- for (int i = 0; i < targets.size(); i++) {
- expr = targets.getTarget(i).getEvalTree();
- if (canBeEvaluated(expr, groupByNode) &&
- EvalTreeUtil.findDistinctAggFunction(expr).size() > 0 && expr.getType() != EvalNode.Type.FIELD) {
- targets.setEvaluated(i);
- groupbyPushable.add((Target) targets.getTarget(i).clone());
- groupbyPushableId.add(i);
- }
- }
-
- return groupByNode;
- }
-
- case SORT: // It does not support the projection
- SortNode sortNode = (SortNode) node;
-
- for (SortSpec spec : sortNode.getSortKeys()) {
- necessary.add(spec.getSortKey());
- }
-
- stack.add(sortNode);
- outer = pushProjectionRecursive(ctx, optContext,
- sortNode.getSubNode(), stack, necessary);
- stack.pop();
- sortNode.setInSchema(outer.getOutSchema());
- sortNode.setOutSchema(outer.getOutSchema());
- break;
-
-
- case JOIN: {
- JoinNode joinNode = (JoinNode) node;
- Set<Column> parentNecessary = Sets.newHashSet(necessary);
-
- if (joinNode.hasJoinQual()) {
- necessary.addAll(EvalTreeUtil.findDistinctRefColumns(joinNode.getJoinQual()));
- }
-
- stack.add(joinNode);
- outer = pushProjectionRecursive(ctx, optContext,
- joinNode.getOuterNode(), stack, necessary);
- inner = pushProjectionRecursive(ctx, optContext,
- joinNode.getInnerNode(), stack, necessary);
- stack.pop();
- Schema merged = SchemaUtil
- .merge(outer.getOutSchema(), inner.getOutSchema());
- joinNode.setInSchema(merged);
-
- TargetListManager targets = optContext.getTargetListManager();
- List<Target> joinPushable = Lists.newArrayList();
- List<Integer> joinPushableId = Lists.newArrayList();
- EvalNode expr;
- for (int i = 0; i < targets.size(); i++) {
- expr = targets.getTarget(i).getEvalTree();
- if (canBeEvaluated(expr, joinNode)
- && EvalTreeUtil.findDistinctAggFunction(expr).size() == 0
- && expr.getType() != EvalNode.Type.FIELD) {
- targets.setEvaluated(i);
- joinPushable.add(targets.getTarget(i));
- joinPushableId.add(i);
- }
- }
- if (joinPushable.size() > 0) {
- joinNode.setTargetList(targets.targets);
- }
-
- Schema outSchema = shrinkOutSchema(joinNode.getInSchema(), targets.getUpdatedSchema().getColumns());
- for (Integer t : joinPushableId) {
- outSchema.addColumn(targets.getEvaluatedColumn(t));
- }
- outSchema = SchemaUtil.mergeAllWithNoDup(outSchema.getColumns(),
- SchemaUtil.getProjectedSchema(joinNode.getInSchema(),parentNecessary).getColumns());
- joinNode.setOutSchema(outSchema);
- break;
- }
-
- case UNION: // It does not support the projection
- UnionNode unionNode = (UnionNode) node;
- stack.add(unionNode);
-
- ParseTree tree = ctx.getParseTree();
- if (tree instanceof CreateTableStmt) {
- tree = ((CreateTableStmt) tree).getSelectStmt();
- }
- QueryBlock block = (QueryBlock) tree;
-
- OptimizationContext outerCtx = new OptimizationContext(ctx,
- block.getTargetList());
- OptimizationContext innerCtx = new OptimizationContext(ctx,
- block.getTargetList());
- pushProjectionRecursive(ctx, outerCtx, unionNode.getOuterNode(),
- stack, necessary);
- pushProjectionRecursive(ctx, innerCtx, unionNode.getInnerNode(),
- stack, necessary);
- stack.pop();
-
- // if this is the final union, we assume that all targets are evalauted
- // TODO - is it always correct?
- if (stack.peek().getType() != ExprType.UNION) {
- optContext.getTargetListManager().setEvaluatedAll();
- }
- break;
-
- case SCAN: {
- ScanNode scanNode = (ScanNode) node;
- TargetListManager targets = optContext.getTargetListManager();
- List<Integer> scanPushableId = Lists.newArrayList();
- List<Target> scanPushable = Lists.newArrayList();
- EvalNode expr;
- for (int i = 0; i < targets.size(); i++) {
- expr = targets.getTarget(i).getEvalTree();
- if (!targets.isEvaluated(i) && canBeEvaluated(expr, scanNode)) {
- if (expr.getType() == EvalNode.Type.FIELD) {
- targets.setEvaluated(i);
- } else if (EvalTreeUtil.findDistinctAggFunction(expr).size() == 0) {
- targets.setEvaluated(i);
- scanPushable.add(targets.getTarget(i));
- scanPushableId.add(i);
- }
- }
- }
-
- if (scanPushable.size() > 0) {
- scanNode.setTargets(scanPushable.toArray(new Target[scanPushable.size()]));
- }
- Schema outSchema = shrinkOutSchema(scanNode.getInSchema(), targets.getUpdatedSchema().getColumns());
- for (Integer t : scanPushableId) {
- outSchema.addColumn(targets.getEvaluatedColumn(t));
- }
- outSchema = SchemaUtil.mergeAllWithNoDup(outSchema.getColumns(), SchemaUtil.getProjectedSchema(scanNode.getInSchema(),necessary).getColumns());
- scanNode.setOutSchema(outSchema);
-
- break;
- }
-
- default:
- }
-
- return node;
- }
-
- private static Schema shrinkOutSchema(Schema inSchema, Collection<Column> necessary) {
- Schema projected = new Schema();
- for(Column col : inSchema.getColumns()) {
- if(necessary.contains(col)) {
- projected.addColumn(col);
- }
- }
- return projected;
- }
-
- public static class OptimizationContext {
- PlanningContext context;
- TargetListManager targetListManager;
-
- public OptimizationContext(PlanningContext context, Target [] targets) {
- this.context = context;
- this.targetListManager = new TargetListManager(context, targets);
- }
-
- public TargetListManager getTargetListManager() {
- return this.targetListManager;
- }
- }
-
- public static class TargetListManager {
- private PlanningContext context;
- private boolean [] evaluated;
- private Target [] targets;
-
- public TargetListManager(PlanningContext context, Target [] targets) {
- this.context = context;
- if (targets == null) {
- evaluated = new boolean[0];
- } else {
- evaluated = new boolean[targets.length];
- }
- this.targets = targets;
- }
-
- public Target getTarget(int id) {
- return targets[id];
- }
-
- public Target [] getTargets() {
- return this.targets;
- }
-
- public int size() {
- return targets.length;
- }
-
- public void setEvaluated(int id) {
- evaluated[id] = true;
- }
-
- public void setEvaluatedAll() {
- for (int i = 0; i < evaluated.length; i++) {
- evaluated[i] = true;
- }
- }
-
- private boolean isEvaluated(int id) {
- return evaluated[id];
- }
-
- public Target [] getUpdatedTarget() throws CloneNotSupportedException {
- Target [] updated = new Target[evaluated.length];
- for (int i = 0; i < evaluated.length; i++) {
- if (evaluated[i]) {
- Column col = getEvaluatedColumn(i);
- updated[i] = new Target(new FieldEval(col), i);
- } else {
- updated[i] = (Target) targets[i].clone();
- }
- }
- return updated;
- }
-
- public Schema getUpdatedSchema() {
- Schema schema = new Schema();
- for (int i = 0; i < evaluated.length; i++) {
- if (evaluated[i]) {
- Column col = getEvaluatedColumn(i);
- schema.addColumn(col);
- } else {
- Collection<Column> cols = getColumnRefs(i);
- for (Column col : cols) {
- if (!schema.contains(col.getQualifiedName())) {
- schema.addColumn(col);
- }
- }
- }
- }
-
- return schema;
- }
-
- public Collection<Column> getColumnRefs(int id) {
- return EvalTreeUtil.findDistinctRefColumns(targets[id].getEvalTree());
- }
-
- public Column getEvaluatedColumn(int id) {
- Target t = targets[id];
- String name;
- if (t.hasAlias()) {
- name = t.getAlias();
- } else if (t.getEvalTree().getName().equals("?")) {
- name = context.getGeneratedColumnName();
- } else {
- name = t.getEvalTree().getName();
- }
- return new Column(name, t.getEvalTree().getValueType()[0]);
- }
-
- public boolean isAllEvaluated() {
- for (boolean isEval : evaluated) {
- if (!isEval) {
- return false;
- }
- }
-
- return true;
- }
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/bc6359b8/tajo-core/tajo-core-backend/src/main/java/tajo/engine/planner/LogicalPlanner.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/tajo/engine/planner/LogicalPlanner.java b/tajo-core/tajo-core-backend/src/main/java/tajo/engine/planner/LogicalPlanner.java
deleted file mode 100644
index 91e5cdc..0000000
--- a/tajo-core/tajo-core-backend/src/main/java/tajo/engine/planner/LogicalPlanner.java
+++ /dev/null
@@ -1,687 +0,0 @@
-/**
- * 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 tajo.engine.planner;
-
-import com.google.common.collect.Lists;
-import com.google.common.collect.Maps;
-import com.google.common.collect.Sets;
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import tajo.catalog.CatalogService;
-import tajo.catalog.Column;
-import tajo.catalog.Schema;
-import tajo.catalog.proto.CatalogProtos;
-import tajo.common.TajoDataTypes.DataType;
-import tajo.engine.eval.*;
-import tajo.engine.parser.*;
-import tajo.engine.parser.QueryBlock.*;
-import tajo.engine.planner.logical.*;
-import tajo.engine.planner.logical.join.Edge;
-import tajo.engine.planner.logical.join.JoinTree;
-import tajo.engine.query.exception.InvalidQueryException;
-import tajo.engine.query.exception.NotSupportQueryException;
-import tajo.engine.utils.SchemaUtil;
-
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.Stack;
-
-/**
- * This class creates a logical plan from a parse tree ({@link tajo.engine.parser.QueryBlock})
- * generated by {@link tajo.engine.parser.QueryAnalyzer}.
- *
- * @see tajo.engine.parser.QueryBlock
- */
-public class LogicalPlanner {
- private static Log LOG = LogFactory.getLog(LogicalPlanner.class);
- private final CatalogService catalog;
-
- public LogicalPlanner(CatalogService catalog) {
- this.catalog = catalog;
- }
-
- /**
- * This generates a logical plan.
- *
- * @param context
- * @return a initial logical plan
- */
- public LogicalNode createPlan(PlanningContext context) {
- LogicalNode plan;
-
- try {
- plan = createPlanInternal(context, context.getParseTree());
- } catch (CloneNotSupportedException e) {
- throw new InvalidQueryException(e);
- }
-
- LogicalRootNode root = new LogicalRootNode();
- root.setInSchema(plan.getOutSchema());
- root.setOutSchema(plan.getOutSchema());
- root.setSubNode(plan);
-
- return root;
- }
-
- private LogicalNode createPlanInternal(PlanningContext ctx,
- ParseTree query) throws CloneNotSupportedException {
- LogicalNode plan;
-
- switch(query.getType()) {
- case SELECT:
- LOG.info("Planning select statement");
- QueryBlock select = (QueryBlock) query;
- plan = buildSelectPlan(ctx, select);
- break;
-
- case UNION:
- case EXCEPT:
- case INTERSECT:
- SetStmt set = (SetStmt) query;
- plan = buildSetPlan(ctx, set);
- break;
-
- case CREATE_INDEX:
- LOG.info("Planning create index statement");
- CreateIndexStmt createIndex = (CreateIndexStmt) query;
- plan = buildCreateIndexPlan(createIndex);
- break;
-
- case CREATE_TABLE:
- case CREATE_TABLE_AS:
- LOG.info("Planning store statement");
- CreateTableStmt createTable = (CreateTableStmt) query;
- plan = buildCreateTablePlan(ctx, createTable);
- break;
-
- default:
- throw new NotSupportQueryException(query.toString());
- }
-
- return plan;
- }
-
- private LogicalNode buildSetPlan(PlanningContext ctx, SetStmt stmt)
- throws CloneNotSupportedException {
- BinaryNode bin;
- switch (stmt.getType()) {
- case UNION:
- bin = new UnionNode();
- break;
- case EXCEPT:
- bin = new ExceptNode();
- break;
- case INTERSECT:
- bin = new IntersectNode();
- break;
- default:
- throw new IllegalStateException("the statement cannot be matched to any set operation type");
- }
-
- bin.setOuter(createPlanInternal(ctx, stmt.getLeftTree()));
- bin.setInner(createPlanInternal(ctx, stmt.getRightTree()));
- bin.setInSchema(bin.getOuterNode().getOutSchema());
- bin.setOutSchema(bin.getOuterNode().getOutSchema());
- return bin;
- }
-
- private LogicalNode buildCreateIndexPlan(CreateIndexStmt stmt) {
- FromTable table = new FromTable(catalog.getTableDesc(stmt.getTableName()));
- ScanNode scan = new ScanNode(table);
- scan.setInSchema(table.getSchema());
- scan.setOutSchema(table.getSchema());
- IndexWriteNode indexWrite = new IndexWriteNode(stmt);
- indexWrite.setSubNode(scan);
- indexWrite.setInSchema(scan.getOutSchema());
- indexWrite.setOutSchema(scan.getOutSchema());
-
- return indexWrite;
- }
-
- private static LogicalNode buildCreateTablePlan(final PlanningContext ctx,
- final CreateTableStmt query)
- throws CloneNotSupportedException {
- LogicalNode node;
-
- if (query.hasQueryBlock()) {
- LogicalNode selectPlan = buildSelectPlan(ctx, query.getSelectStmt());
- StoreTableNode storeNode = new StoreTableNode(query.getTableName());
-
- storeNode.setSubNode(selectPlan);
-
- if (query.hasDefinition()) {
- storeNode.setOutSchema(query.getTableDef());
- } else {
- // TODO - strip qualified name
- storeNode.setOutSchema(selectPlan.getOutSchema());
- }
- storeNode.setInSchema(selectPlan.getOutSchema());
-
- if (query.hasStoreType()) {
- storeNode.setStorageType(query.getStoreType());
- } else {
- // default type
- // TODO - it should be configurable.
- storeNode.setStorageType(CatalogProtos.StoreType.CSV);
- }
- if (query.hasOptions()) {
- storeNode.setOptions(query.getOptions());
- }
-
- node = storeNode;
- } else {
- CreateTableNode createTable =
- new CreateTableNode(query.getTableName(), query.getTableDef());
-
- if (query.hasStoreType()) {
- createTable.setStorageType(query.getStoreType());
- } else {
- // default type
- // TODO - it should be configurable.
- createTable.setStorageType(CatalogProtos.StoreType.CSV);
- }
- if (query.hasOptions()) {
- createTable.setOptions(query.getOptions());
- }
-
- if (query.hasPath()) {
- createTable.setPath(query.getPath());
- }
-
- node = createTable;
- }
-
- return node;
- }
-
- /**
- * ^(SELECT from_clause? where_clause? groupby_clause? selectList)
- *
- * @param query
- * @return the planed logical plan
- */
- private static LogicalNode buildSelectPlan(PlanningContext ctx,
- QueryBlock query)
- throws CloneNotSupportedException {
- LogicalNode subroot;
- EvalNode whereCondition = null;
- EvalNode [] cnf = null;
- if(query.hasWhereClause()) {
- whereCondition = query.getWhereCondition();
- whereCondition = AlgebraicUtil.simplify(whereCondition);
- cnf = EvalTreeUtil.getConjNormalForm(whereCondition);
- }
-
- if(query.hasFromClause()) {
- if (query.hasExplicitJoinClause()) {
- subroot = createExplicitJoinTree(query);
- } else {
- subroot = createImplicitJoinTree(query.getFromTables(), cnf);
- }
- } else {
- subroot = new EvalExprNode(query.getTargetList());
- subroot.setOutSchema(getProjectedSchema(ctx, query.getTargetList()));
- return subroot;
- }
-
- if(whereCondition != null) {
- SelectionNode selNode =
- new SelectionNode(query.getWhereCondition());
- selNode.setSubNode(subroot);
- selNode.setInSchema(subroot.getOutSchema());
- selNode.setOutSchema(selNode.getInSchema());
- subroot = selNode;
- }
-
- if(query.hasAggregation()) {
- if (query.isDistinct()) {
- throw new InvalidQueryException("Cannot support GROUP BY queries with distinct keyword");
- }
-
- GroupbyNode groupbyNode = null;
- if (query.hasGroupbyClause()) {
- if (query.getGroupByClause().getGroupSet().get(0).getType() == GroupType.GROUPBY) {
- groupbyNode = new GroupbyNode(query.getGroupByClause().getGroupSet().get(0).getColumns());
- groupbyNode.setTargetList(query.getTargetList());
- groupbyNode.setSubNode(subroot);
- groupbyNode.setInSchema(subroot.getOutSchema());
- Schema outSchema = getProjectedSchema(ctx, query.getTargetList());
- groupbyNode.setOutSchema(outSchema);
- subroot = groupbyNode;
- } else if (query.getGroupByClause().getGroupSet().get(0).getType() == GroupType.CUBE) {
- LogicalNode union = createGroupByUnionByCube(ctx, query,
- subroot, query.getGroupByClause());
- Schema outSchema = getProjectedSchema(ctx, query.getTargetList());
- union.setOutSchema(outSchema);
- subroot = union;
- }
- if(query.hasHavingCond())
- groupbyNode.setHavingCondition(query.getHavingCond());
- } else {
- // when aggregation functions are used without grouping fields
- groupbyNode = new GroupbyNode(new Column[] {});
- groupbyNode.setTargetList(query.getTargetList());
- groupbyNode.setSubNode(subroot);
- groupbyNode.setInSchema(subroot.getOutSchema());
- Schema outSchema = getProjectedSchema(ctx, query.getTargetList());
- groupbyNode.setOutSchema(outSchema);
- subroot = groupbyNode;
- }
- }
-
- if(query.hasOrderByClause()) {
- SortNode sortNode = new SortNode(query.getSortKeys());
- sortNode.setSubNode(subroot);
- sortNode.setInSchema(subroot.getOutSchema());
- sortNode.setOutSchema(sortNode.getInSchema());
- subroot = sortNode;
- }
-
- ProjectionNode prjNode;
- if (query.getProjectAll()) {
- Schema merged = SchemaUtil.merge(query.getFromTables());
- Target [] allTargets = PlannerUtil.schemaToTargets(merged);
- prjNode = new ProjectionNode(allTargets);
- prjNode.setSubNode(subroot);
- prjNode.setInSchema(merged);
- prjNode.setOutSchema(merged);
- subroot = prjNode;
- query.setTargetList(allTargets);
- } else {
- prjNode = new ProjectionNode(query.getTargetList());
- if (subroot != null) { // false if 'no from' statement
- prjNode.setSubNode(subroot);
- }
- prjNode.setInSchema(subroot.getOutSchema());
-
- // All aggregate functions are evaluated before the projection.
- // So, the targets for aggregate functions should be updated.
- LogicalOptimizer.TargetListManager tlm = new LogicalOptimizer.
- TargetListManager(ctx, query.getTargetList());
- for (int i = 0; i < tlm.getTargets().length; i++) {
- if (EvalTreeUtil.findDistinctAggFunction(tlm.getTarget(i).getEvalTree()).size() > 0) {
- tlm.setEvaluated(i);
- }
- }
- prjNode.setTargetList(tlm.getUpdatedTarget());
- Schema projected = getProjectedSchema(ctx, tlm.getUpdatedTarget());
- prjNode.setOutSchema(projected);
- subroot = prjNode;
- }
-
- GroupbyNode dupRemoval;
- if (query.isDistinct()) {
- dupRemoval = new GroupbyNode(subroot.getOutSchema().toArray());
- dupRemoval.setTargetList(query.getTargetList());
- dupRemoval.setSubNode(subroot);
- dupRemoval.setInSchema(subroot.getOutSchema());
- Schema outSchema = getProjectedSchema(ctx, query.getTargetList());
- dupRemoval.setOutSchema(outSchema);
- subroot = dupRemoval;
- }
-
- if (query.hasLimitClause()) {
- LimitNode limitNode = new LimitNode(query.getLimitClause());
- limitNode.setSubNode(subroot);
- limitNode.setInSchema(subroot.getOutSchema());
- limitNode.setOutSchema(limitNode.getInSchema());
- subroot = limitNode;
- }
-
- return subroot;
- }
-
- public static LogicalNode createGroupByUnionByCube(
- final PlanningContext context,
- final QueryBlock queryBlock,
- final LogicalNode subNode,
- final GroupByClause clause) {
-
- GroupElement element = clause.getGroupSet().get(0);
-
- List<Column []> cuboids = generateCuboids(element.getColumns());
-
- return createGroupByUnion(context, queryBlock, subNode, cuboids, 0);
- }
-
- private static Target [] cloneTargets(Target [] srcs)
- throws CloneNotSupportedException {
- Target [] clone = new Target[srcs.length];
- for (int i = 0; i < srcs.length; i++) {
- clone[i] = (Target) srcs[i].clone();
- }
-
- return clone;
- }
-
- private static UnionNode createGroupByUnion(final PlanningContext context,
- final QueryBlock queryBlock,
- final LogicalNode subNode,
- final List<Column []> cuboids,
- final int idx) {
- UnionNode union;
- try {
- if ((cuboids.size() - idx) > 2) {
- GroupbyNode g1 = new GroupbyNode(cuboids.get(idx));
- Target [] clone = cloneTargets(queryBlock.getTargetList());
-
- g1.setTargetList(clone);
- g1.setSubNode((LogicalNode) subNode.clone());
- g1.setInSchema(g1.getSubNode().getOutSchema());
- Schema outSchema = getProjectedSchema(context, queryBlock.getTargetList());
- g1.setOutSchema(outSchema);
-
- union = new UnionNode(g1, createGroupByUnion(context, queryBlock,
- subNode, cuboids, idx+1));
- union.setInSchema(g1.getOutSchema());
- union.setOutSchema(g1.getOutSchema());
- return union;
- } else {
- GroupbyNode g1 = new GroupbyNode(cuboids.get(idx));
- Target [] clone = cloneTargets(queryBlock.getTargetList());
- g1.setTargetList(clone);
- g1.setSubNode((LogicalNode) subNode.clone());
- g1.setInSchema(g1.getSubNode().getOutSchema());
- Schema outSchema = getProjectedSchema(context, queryBlock.getTargetList());
- g1.setOutSchema(outSchema);
-
- GroupbyNode g2 = new GroupbyNode(cuboids.get(idx+1));
- clone = cloneTargets(queryBlock.getTargetList());
- g2.setTargetList(clone);
- g2.setSubNode((LogicalNode) subNode.clone());
- g2.setInSchema(g1.getSubNode().getOutSchema());
- outSchema = getProjectedSchema(context, queryBlock.getTargetList());
- g2.setOutSchema(outSchema);
- union = new UnionNode(g1, g2);
- union.setInSchema(g1.getOutSchema());
- union.setOutSchema(g1.getOutSchema());
- return union;
- }
- } catch (CloneNotSupportedException cnse) {
- LOG.error(cnse);
- throw new InvalidQueryException(cnse);
- }
- }
-
- public static final Column [] ALL
- = Lists.newArrayList().toArray(new Column[0]);
-
- public static List<Column []> generateCuboids(Column [] columns) {
- int numCuboids = (int) Math.pow(2, columns.length);
- int maxBits = columns.length;
-
- List<Column []> cube = Lists.newArrayList();
- List<Column> cuboidCols;
-
- cube.add(ALL);
- for (int cuboidId = 1; cuboidId < numCuboids; cuboidId++) {
- cuboidCols = Lists.newArrayList();
- for (int j = 0; j < maxBits; j++) {
- int bit = 1 << j;
- if ((cuboidId & bit) == bit) {
- cuboidCols.add(columns[j]);
- }
- }
- cube.add(cuboidCols.toArray(new Column[cuboidCols.size()]));
- }
- return cube;
- }
-
- private static LogicalNode createExplicitJoinTree(QueryBlock block) {
- return createExplicitJoinTree_(block.getJoinClause());
- }
-
- private static LogicalNode createExplicitJoinTree_(JoinClause joinClause) {
-
- JoinNode join;
- if (joinClause.hasLeftJoin()) {
- LogicalNode outer = createExplicitJoinTree_(joinClause.getLeftJoin());
- join = new JoinNode(joinClause.getJoinType(), outer);
- join.setInner(new ScanNode(joinClause.getRight()));
- } else {
- join = new JoinNode(joinClause.getJoinType(), new ScanNode(joinClause.getLeft()),
- new ScanNode(joinClause.getRight()));
- }
- if (joinClause.hasJoinQual()) {
- join.setJoinQual(joinClause.getJoinQual());
- } else if (joinClause.hasJoinColumns()) {
- // for using clause of explicit join
- // TODO - to be implemented. Now, tajo only support 'ON' join clause.
- }
-
- // Determine Join Schemas
- Schema merged;
- if (joinClause.isNatural()) {
- merged = getNaturalJoin(join.getOuterNode(), join.getInnerNode());
- } else {
- merged = SchemaUtil.merge(join.getOuterNode().getOutSchema(),
- join.getInnerNode().getOutSchema());
- }
-
- join.setInSchema(merged);
- join.setOutSchema(merged);
-
- // Determine join quals
- // if natural join, should have the equi join conditions on common columns
- if (joinClause.isNatural()) {
- Schema leftSchema = join.getOuterNode().getOutSchema();
- Schema rightSchema = join.getInnerNode().getOutSchema();
- Schema commons = SchemaUtil.getCommons(
- leftSchema, rightSchema);
- EvalNode njCond = getNaturalJoinCondition(leftSchema, rightSchema, commons);
- join.setJoinQual(njCond);
- } else if (joinClause.hasJoinQual()) {
- // otherwise, the given join conditions are set
- join.setJoinQual(joinClause.getJoinQual());
- }
-
- return join;
- }
-
- private static EvalNode getNaturalJoinCondition(Schema outer, Schema inner, Schema commons) {
- EvalNode njQual = null;
- EvalNode equiQual;
-
- Column leftJoinKey;
- Column rightJoinKey;
- for (Column common : commons.getColumns()) {
- leftJoinKey = outer.getColumnByName(common.getColumnName());
- rightJoinKey = inner.getColumnByName(common.getColumnName());
- equiQual = new BinaryEval(EvalNode.Type.EQUAL,
- new FieldEval(leftJoinKey), new FieldEval(rightJoinKey));
- if (njQual == null) {
- njQual = equiQual;
- } else {
- njQual = new BinaryEval(EvalNode.Type.AND,
- njQual, equiQual);
- }
- }
-
- return njQual;
- }
-
- private static LogicalNode createImplicitJoinTree(FromTable [] tables,
- EvalNode [] cnf) {
- if (cnf == null) {
- return createCatasianProduct(tables);
- } else {
- return createCrossJoinFromJoinCondition(tables, cnf);
- }
- }
-
- private static LogicalNode createCrossJoinFromJoinCondition(
- FromTable [] tables, EvalNode [] cnf) {
- Map<String, FromTable> fromTableMap = Maps.newHashMap();
- for (FromTable f : tables) {
- // TODO - to consider alias and self-join
- fromTableMap.put(f.getTableName(), f);
- }
-
- JoinTree joinTree = new JoinTree(); // to infer join order
- for (EvalNode expr : cnf) {
- if (PlannerUtil.isJoinQual(expr)) {
- joinTree.addJoin(expr);
- }
- }
-
- List<String> remain = Lists.newArrayList(fromTableMap.keySet());
- remain.removeAll(joinTree.getTables()); // only remain joins not matched to any join condition
- List<Edge> joinOrder = null;
- LogicalNode subroot = null;
- JoinNode join;
- Schema joinSchema;
-
- // if there are at least one join matched to the one of join conditions,
- // we try to traverse the join tree in the depth-first manner and
- // determine the initial join order. Here, we do not consider the join cost.
- // The optimized join order will be considered in the optimizer.
- if (joinTree.getJoinNum() > 0) {
- Stack<String> stack = new Stack<String>();
- Set<String> visited = Sets.newHashSet();
-
-
- // initially, one table is pushed into the stack
- String seed = joinTree.getTables().iterator().next();
- stack.add(seed);
-
- joinOrder = Lists.newArrayList();
-
- while (!stack.empty()) {
- String table = stack.pop();
- if (visited.contains(table)) {
- continue;
- }
- visited.add(table);
-
- // 'joinOrder' will contain all tables corresponding to the given join conditions.
- for (Edge edge : joinTree.getEdges(table)) {
- if (!visited.contains(edge.getTarget()) && !edge.getTarget().equals(table)) {
- stack.add(edge.getTarget());
- joinOrder.add(edge);
- }
- }
- }
-
- subroot = new ScanNode(fromTableMap.get(joinOrder.get(0).getSrc()));
- LogicalNode inner;
- for (Edge edge : joinOrder) {
- inner = new ScanNode(fromTableMap.get(edge.getTarget()));
- join = new JoinNode(JoinType.CROSS_JOIN, subroot, inner);
- subroot = join;
-
- joinSchema = SchemaUtil.merge(
- join.getOuterNode().getOutSchema(),
- join.getInnerNode().getOutSchema());
- join.setInSchema(joinSchema);
- join.setOutSchema(joinSchema);
- }
- }
-
- // Here, there are two cases:
- // 1) there already exists the join plan.
- // 2) there are no join plan.
- if (joinOrder != null) { // case 1)
- // if there are join tables corresponding to any join condition,
- // the join plan is placed as the outer plan of the product.
- remain.remove(joinOrder.get(0).getSrc());
- remain.remove(joinOrder.get(0).getTarget());
- } else { // case 2)
- // if there are no inferred joins, the one of the remain join tables is placed as the left table
- subroot = new ScanNode(fromTableMap.get(remain.get(0)));
- remain.remove(remain.get(0));
- }
-
- // Here, the variable 'remain' contains join tables which are not matched to any join conditions.
- // Thus, they will be joined by catasian product
- for (String table : remain) {
- join = new JoinNode(JoinType.CROSS_JOIN,
- subroot, new ScanNode(fromTableMap.get(table)));
- joinSchema = SchemaUtil.merge(
- join.getOuterNode().getOutSchema(),
- join.getInnerNode().getOutSchema());
- join.setInSchema(joinSchema);
- join.setOutSchema(joinSchema);
- subroot = join;
- }
-
- return subroot;
- }
-
- // TODO - this method is somewhat duplicated to createCrossJoinFromJoinCondition. Later, it should be removed.
- private static LogicalNode createCatasianProduct(FromTable [] tables) {
- LogicalNode subroot = new ScanNode(tables[0]);
- Schema joinSchema;
- if(tables.length > 1) {
- for(int i=1; i < tables.length; i++) {
- JoinNode join = new JoinNode(JoinType.CROSS_JOIN,
- subroot, new ScanNode(tables[i]));
- joinSchema = SchemaUtil.merge(
- join.getOuterNode().getOutSchema(),
- join.getInnerNode().getOutSchema());
- join.setInSchema(joinSchema);
- join.setOutSchema(joinSchema);
- subroot = join;
- }
- }
-
- return subroot;
- }
-
- public static Schema getProjectedSchema(PlanningContext context, Target [] targets) {
- Schema projected = new Schema();
- for(Target t : targets) {
- DataType type = t.getEvalTree().getValueType()[0];
- String name;
- if (t.hasAlias()) {
- name = t.getAlias();
- } else if (t.getEvalTree().getName().equals("?")) {
- name = context.getGeneratedColumnName();
- } else {
- name = t.getEvalTree().getName();
- }
- projected.addColumn(name,type);
- }
-
- return projected;
- }
-
- private static Schema getNaturalJoin(LogicalNode outer, LogicalNode inner) {
- Schema joinSchema = new Schema();
- Schema commons = SchemaUtil.getCommons(outer.getOutSchema(),
- inner.getOutSchema());
- joinSchema.addColumns(commons);
- for (Column c : outer.getOutSchema().getColumns()) {
- for (Column common : commons.getColumns()) {
- if (!common.getColumnName().equals(c.getColumnName())) {
- joinSchema.addColumn(c);
- }
- }
- }
-
- for (Column c : inner.getOutSchema().getColumns()) {
- for (Column common : commons.getColumns()) {
- if (!common.getColumnName().equals(c.getColumnName())) {
- joinSchema.addColumn(c);
- }
- }
- }
- return joinSchema;
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/bc6359b8/tajo-core/tajo-core-backend/src/main/java/tajo/engine/planner/PhysicalPlanner.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/tajo/engine/planner/PhysicalPlanner.java b/tajo-core/tajo-core-backend/src/main/java/tajo/engine/planner/PhysicalPlanner.java
deleted file mode 100644
index 109468b..0000000
--- a/tajo-core/tajo-core-backend/src/main/java/tajo/engine/planner/PhysicalPlanner.java
+++ /dev/null
@@ -1,36 +0,0 @@
-/**
- * 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 tajo.engine.planner;
-
-import tajo.TaskAttemptContext;
-import tajo.engine.planner.logical.LogicalNode;
-import tajo.engine.planner.physical.PhysicalExec;
-import tajo.exception.InternalException;
-
-/**
- * This class generates a physical execution plan.
- */
-public interface PhysicalPlanner {
- public PhysicalExec createPlan(TaskAttemptContext context,
- LogicalNode logicalPlan)
- throws InternalException;
-}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/bc6359b8/tajo-core/tajo-core-backend/src/main/java/tajo/engine/planner/PhysicalPlannerImpl.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/tajo/engine/planner/PhysicalPlannerImpl.java b/tajo-core/tajo-core-backend/src/main/java/tajo/engine/planner/PhysicalPlannerImpl.java
deleted file mode 100644
index 8fe317b..0000000
--- a/tajo-core/tajo-core-backend/src/main/java/tajo/engine/planner/PhysicalPlannerImpl.java
+++ /dev/null
@@ -1,321 +0,0 @@
-/**
- * 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 tajo.engine.planner;
-
-import com.google.common.base.Preconditions;
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.hadoop.fs.Path;
-import tajo.TaskAttemptContext;
-import tajo.catalog.Column;
-import tajo.catalog.SortSpec;
-import tajo.conf.TajoConf;
-import tajo.engine.planner.logical.*;
-import tajo.engine.planner.physical.*;
-import tajo.exception.InternalException;
-import tajo.storage.Fragment;
-import tajo.storage.StorageManager;
-import tajo.storage.TupleComparator;
-import tajo.util.IndexUtil;
-
-import java.io.IOException;
-
-public class PhysicalPlannerImpl implements PhysicalPlanner {
- private static final Log LOG = LogFactory.getLog(PhysicalPlannerImpl.class);
- protected final TajoConf conf;
- protected final StorageManager sm;
-
- public PhysicalPlannerImpl(final TajoConf conf, final StorageManager sm) {
- this.conf = conf;
- this.sm = sm;
- }
-
- public PhysicalExec createPlan(final TaskAttemptContext context,
- final LogicalNode logicalPlan) throws InternalException {
-
- PhysicalExec plan;
-
- try {
- plan = createPlanRecursive(context, logicalPlan);
-
- } catch (IOException ioe) {
- throw new InternalException(ioe);
- }
-
- return plan;
- }
-
- private PhysicalExec createPlanRecursive(TaskAttemptContext ctx, LogicalNode logicalNode) throws IOException {
- PhysicalExec outer;
- PhysicalExec inner;
-
- switch (logicalNode.getType()) {
-
- case ROOT:
- LogicalRootNode rootNode = (LogicalRootNode) logicalNode;
- return createPlanRecursive(ctx, rootNode.getSubNode());
-
- case EXPRS:
- EvalExprNode evalExpr = (EvalExprNode) logicalNode;
- return new EvalExprExec(ctx, evalExpr);
-
- case STORE:
- StoreTableNode storeNode = (StoreTableNode) logicalNode;
- outer = createPlanRecursive(ctx, storeNode.getSubNode());
- return createStorePlan(ctx, storeNode, outer);
-
- case SELECTION:
- SelectionNode selNode = (SelectionNode) logicalNode;
- outer = createPlanRecursive(ctx, selNode.getSubNode());
- return new SelectionExec(ctx, selNode, outer);
-
- case PROJECTION:
- ProjectionNode prjNode = (ProjectionNode) logicalNode;
- outer = createPlanRecursive(ctx, prjNode.getSubNode());
- return new ProjectionExec(ctx, prjNode, outer);
-
- case SCAN:
- outer = createScanPlan(ctx, (ScanNode) logicalNode);
- return outer;
-
- case GROUP_BY:
- GroupbyNode grpNode = (GroupbyNode) logicalNode;
- outer = createPlanRecursive(ctx, grpNode.getSubNode());
- return createGroupByPlan(ctx, grpNode, outer);
-
- case SORT:
- SortNode sortNode = (SortNode) logicalNode;
- outer = createPlanRecursive(ctx, sortNode.getSubNode());
- return createSortPlan(ctx, sortNode, outer);
-
- case JOIN:
- JoinNode joinNode = (JoinNode) logicalNode;
- outer = createPlanRecursive(ctx, joinNode.getOuterNode());
- inner = createPlanRecursive(ctx, joinNode.getInnerNode());
- return createJoinPlan(ctx, joinNode, outer, inner);
-
- case UNION:
- UnionNode unionNode = (UnionNode) logicalNode;
- outer = createPlanRecursive(ctx, unionNode.getOuterNode());
- inner = createPlanRecursive(ctx, unionNode.getInnerNode());
- return new UnionExec(ctx, outer, inner);
-
- case LIMIT:
- LimitNode limitNode = (LimitNode) logicalNode;
- outer = createPlanRecursive(ctx, limitNode.getSubNode());
- return new LimitExec(ctx, limitNode.getInSchema(),
- limitNode.getOutSchema(), outer, limitNode);
-
- case CREATE_INDEX:
- IndexWriteNode createIndexNode = (IndexWriteNode) logicalNode;
- outer = createPlanRecursive(ctx, createIndexNode.getSubNode());
- return createIndexWritePlan(sm, ctx, createIndexNode, outer);
-
- case BST_INDEX_SCAN:
- IndexScanNode indexScanNode = (IndexScanNode) logicalNode;
- outer = createIndexScanExec(ctx, indexScanNode);
- return outer;
-
- case RENAME:
- case SET_UNION:
- case SET_DIFF:
- case SET_INTERSECT:
- case INSERT_INTO:
- case SHOW_TABLE:
- case DESC_TABLE:
- case SHOW_FUNCTION:
- default:
- return null;
- }
- }
-
- private long estimateSizeRecursive(TaskAttemptContext ctx, String [] tableIds) {
- long size = 0;
- for (String tableId : tableIds) {
- Fragment[] fragments = ctx.getTables(tableId);
- for (Fragment frag : fragments) {
- size += frag.getLength();
- }
- }
- return size;
- }
-
- public PhysicalExec createJoinPlan(TaskAttemptContext ctx, JoinNode joinNode,
- PhysicalExec outer, PhysicalExec inner)
- throws IOException {
- switch (joinNode.getJoinType()) {
- case CROSS_JOIN:
- LOG.info("The planner chooses NLJoinExec");
- return new NLJoinExec(ctx, joinNode, outer, inner);
-
- case INNER:
- String [] outerLineage = PlannerUtil.getLineage(joinNode.getOuterNode());
- String [] innerLineage = PlannerUtil.getLineage(joinNode.getInnerNode());
- long outerSize = estimateSizeRecursive(ctx, outerLineage);
- long innerSize = estimateSizeRecursive(ctx, innerLineage);
-
- final long threshold = 1048576 * 128; // 64MB
-
- boolean hashJoin = false;
- if (outerSize < threshold || innerSize < threshold) {
- hashJoin = true;
- }
-
- if (hashJoin) {
- PhysicalExec selectedOuter;
- PhysicalExec selectedInner;
-
- // HashJoinExec loads the inner relation to memory.
- if (outerSize <= innerSize) {
- selectedInner = outer;
- selectedOuter = inner;
- } else {
- selectedInner = inner;
- selectedOuter = outer;
- }
-
- LOG.info("The planner chooses HashJoinExec");
- return new HashJoinExec(ctx, joinNode, selectedOuter, selectedInner);
- }
-
- default:
- SortSpec[][] sortSpecs = PlannerUtil.getSortKeysFromJoinQual(
- joinNode.getJoinQual(), outer.getSchema(), inner.getSchema());
- ExternalSortExec outerSort = new ExternalSortExec(ctx, sm,
- new SortNode(sortSpecs[0], outer.getSchema(), outer.getSchema()),
- outer);
- ExternalSortExec innerSort = new ExternalSortExec(ctx, sm,
- new SortNode(sortSpecs[1], inner.getSchema(), inner.getSchema()),
- inner);
-
- LOG.info("The planner chooses MergeJoinExec");
- return new MergeJoinExec(ctx, joinNode, outerSort, innerSort,
- sortSpecs[0], sortSpecs[1]);
- }
- }
-
- public PhysicalExec createStorePlan(TaskAttemptContext ctx,
- StoreTableNode plan, PhysicalExec subOp) throws IOException {
- if (plan.hasPartitionKey()) {
- switch (plan.getPartitionType()) {
- case HASH:
- return new PartitionedStoreExec(ctx, sm, plan, subOp);
-
- case RANGE:
- SortSpec [] sortSpecs = null;
- if (subOp instanceof SortExec) {
- sortSpecs = ((SortExec)subOp).getSortSpecs();
- } else {
- Column[] columns = plan.getPartitionKeys();
- SortSpec specs[] = new SortSpec[columns.length];
- for (int i = 0; i < columns.length; i++) {
- specs[i] = new SortSpec(columns[i]);
- }
- }
-
- return new IndexedStoreExec(ctx, sm, subOp,
- plan.getInSchema(), plan.getInSchema(), sortSpecs);
- }
- }
- if (plan instanceof StoreIndexNode) {
- return new TunnelExec(ctx, plan.getOutSchema(), subOp);
- }
-
- return new StoreTableExec(ctx, sm, plan, subOp);
- }
-
- public PhysicalExec createScanPlan(TaskAttemptContext ctx, ScanNode scanNode)
- throws IOException {
- Preconditions.checkNotNull(ctx.getTable(scanNode.getTableId()),
- "Error: There is no table matched to %s", scanNode.getTableId());
-
- Fragment[] fragments = ctx.getTables(scanNode.getTableId());
- return new SeqScanExec(ctx, sm, scanNode, fragments);
- }
-
- public PhysicalExec createGroupByPlan(TaskAttemptContext ctx,
- GroupbyNode groupbyNode, PhysicalExec subOp) throws IOException {
- Column[] grpColumns = groupbyNode.getGroupingColumns();
- if (grpColumns.length == 0) {
- LOG.info("The planner chooses HashAggregationExec");
- return new HashAggregateExec(ctx, groupbyNode, subOp);
- } else {
- String [] outerLineage = PlannerUtil.getLineage(groupbyNode.getSubNode());
- long estimatedSize = estimateSizeRecursive(ctx, outerLineage);
- final long threshold = 1048576 * 256;
-
- // if the relation size is less than the reshold,
- // the hash aggregation will be used.
- if (estimatedSize <= threshold) {
- LOG.info("The planner chooses HashAggregationExec");
- return new HashAggregateExec(ctx, groupbyNode, subOp);
- } else {
- SortSpec[] specs = new SortSpec[grpColumns.length];
- for (int i = 0; i < grpColumns.length; i++) {
- specs[i] = new SortSpec(grpColumns[i], true, false);
- }
- SortNode sortNode = new SortNode(specs);
- sortNode.setInSchema(subOp.getSchema());
- sortNode.setOutSchema(subOp.getSchema());
- // SortExec sortExec = new SortExec(sortNode, child);
- ExternalSortExec sortExec = new ExternalSortExec(ctx, sm, sortNode,
- subOp);
- LOG.info("The planner chooses SortAggregationExec");
- return new SortAggregateExec(ctx, groupbyNode, sortExec);
- }
- }
- }
-
- public PhysicalExec createSortPlan(TaskAttemptContext ctx, SortNode sortNode,
- PhysicalExec subOp) throws IOException {
- return new ExternalSortExec(ctx, sm, sortNode, subOp);
- }
-
- public PhysicalExec createIndexWritePlan(StorageManager sm,
- TaskAttemptContext ctx, IndexWriteNode indexWriteNode, PhysicalExec subOp)
- throws IOException {
-
- return new IndexWriteExec(ctx, sm, indexWriteNode, ctx.getTable(indexWriteNode
- .getTableName()), subOp);
- }
-
- public PhysicalExec createIndexScanExec(TaskAttemptContext ctx,
- IndexScanNode annotation)
- throws IOException {
- //TODO-general Type Index
- Preconditions.checkNotNull(ctx.getTable(annotation.getTableId()),
- "Error: There is no table matched to %s", annotation.getTableId());
-
- Fragment[] fragments = ctx.getTables(annotation.getTableId());
-
- String indexName = IndexUtil.getIndexNameOfFrag(fragments[0],
- annotation.getSortKeys());
- Path indexPath = new Path(sm.getTablePath(annotation.getTableId()), "index");
-
- TupleComparator comp = new TupleComparator(annotation.getKeySchema(),
- annotation.getSortKeys());
- return new BSTIndexScanExec(ctx, sm, annotation, fragments[0], new Path(
- indexPath, indexName), annotation.getKeySchema(), comp,
- annotation.getDatum());
-
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/bc6359b8/tajo-core/tajo-core-backend/src/main/java/tajo/engine/planner/PlannerUtil.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/tajo/engine/planner/PlannerUtil.java b/tajo-core/tajo-core-backend/src/main/java/tajo/engine/planner/PlannerUtil.java
deleted file mode 100644
index edf2e4b..0000000
--- a/tajo-core/tajo-core-backend/src/main/java/tajo/engine/planner/PlannerUtil.java
+++ /dev/null
@@ -1,616 +0,0 @@
-/**
- * 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 tajo.engine.planner;
-
-import com.google.common.base.Preconditions;
-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 tajo.catalog.CatalogUtil;
-import tajo.catalog.Column;
-import tajo.catalog.Schema;
-import tajo.catalog.SortSpec;
-import tajo.common.TajoDataTypes.DataType;
-import tajo.common.TajoDataTypes.Type;
-import tajo.engine.eval.*;
-import tajo.engine.parser.QueryBlock;
-import tajo.engine.planner.logical.*;
-import tajo.engine.query.exception.InvalidQueryException;
-import tajo.storage.TupleComparator;
-
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Set;
-
-public class PlannerUtil {
- private static final Log LOG = LogFactory.getLog(PlannerUtil.class);
-
- public static String [] getLineage(LogicalNode node) {
- LogicalNode [] scans = PlannerUtil.findAllNodes(node, ExprType.SCAN);
- String [] tableNames = new String[scans.length];
- ScanNode scan;
- for (int i = 0; i < scans.length; i++) {
- scan = (ScanNode) scans[i];
- /*if (scan.hasAlias()) {
- tableNames[i] = scan.getAlias();
- } else {*/
- tableNames[i] = scan.getTableId();
- //}
- }
- return tableNames;
- }
-
- public static LogicalNode insertNode(LogicalNode parent, LogicalNode newNode) {
- Preconditions.checkArgument(parent instanceof UnaryNode);
- Preconditions.checkArgument(newNode instanceof UnaryNode);
-
- UnaryNode p = (UnaryNode) parent;
- LogicalNode c = p.getSubNode();
- UnaryNode m = (UnaryNode) newNode;
- m.setInSchema(c.getOutSchema());
- m.setOutSchema(c.getOutSchema());
- m.setSubNode(c);
- p.setSubNode(m);
-
- return p;
- }
-
- /**
- * Delete the child of a given parent operator.
- *
- * @param parent Must be a unary logical operator.
- * @return input parent node
- */
- public static LogicalNode deleteNode(LogicalNode parent) {
- if (parent instanceof UnaryNode) {
- UnaryNode unary = (UnaryNode) parent;
- if (unary.getSubNode() instanceof UnaryNode) {
- UnaryNode child = (UnaryNode) unary.getSubNode();
- LogicalNode grandChild = child.getSubNode();
- unary.setSubNode(grandChild);
- } else {
- throw new InvalidQueryException("Unexpected logical plan: " + parent);
- }
- } else {
- throw new InvalidQueryException("Unexpected logical plan: " + parent);
- }
- return parent;
- }
-
- public static void replaceNode(LogicalNode plan, LogicalNode newNode, ExprType type) {
- LogicalNode parent = findTopParentNode(plan, type);
- Preconditions.checkArgument(parent instanceof UnaryNode);
- Preconditions.checkArgument(!(newNode instanceof BinaryNode));
- UnaryNode parentNode = (UnaryNode) parent;
- LogicalNode child = parentNode.getSubNode();
- if (child instanceof UnaryNode) {
- ((UnaryNode) newNode).setSubNode(((UnaryNode)child).getSubNode());
- }
- parentNode.setSubNode(newNode);
- }
-
- public static LogicalNode insertOuterNode(LogicalNode parent, LogicalNode outer) {
- Preconditions.checkArgument(parent instanceof BinaryNode);
- Preconditions.checkArgument(outer instanceof UnaryNode);
-
- BinaryNode p = (BinaryNode) parent;
- LogicalNode c = p.getOuterNode();
- UnaryNode m = (UnaryNode) outer;
- m.setInSchema(c.getOutSchema());
- m.setOutSchema(c.getOutSchema());
- m.setSubNode(c);
- p.setOuter(m);
- return p;
- }
-
- public static LogicalNode insertInnerNode(LogicalNode parent, LogicalNode inner) {
- Preconditions.checkArgument(parent instanceof BinaryNode);
- Preconditions.checkArgument(inner instanceof UnaryNode);
-
- BinaryNode p = (BinaryNode) parent;
- LogicalNode c = p.getInnerNode();
- UnaryNode m = (UnaryNode) inner;
- m.setInSchema(c.getOutSchema());
- m.setOutSchema(c.getOutSchema());
- m.setSubNode(c);
- p.setInner(m);
- return p;
- }
-
- public static LogicalNode insertNode(LogicalNode parent,
- LogicalNode left, LogicalNode right) {
- Preconditions.checkArgument(parent instanceof BinaryNode);
- Preconditions.checkArgument(left instanceof UnaryNode);
- Preconditions.checkArgument(right instanceof UnaryNode);
-
- BinaryNode p = (BinaryNode)parent;
- LogicalNode lc = p.getOuterNode();
- LogicalNode rc = p.getInnerNode();
- UnaryNode lm = (UnaryNode)left;
- UnaryNode rm = (UnaryNode)right;
- lm.setInSchema(lc.getOutSchema());
- lm.setOutSchema(lc.getOutSchema());
- lm.setSubNode(lc);
- rm.setInSchema(rc.getOutSchema());
- rm.setOutSchema(rc.getOutSchema());
- rm.setSubNode(rc);
- p.setOuter(lm);
- p.setInner(rm);
- return p;
- }
-
- public static LogicalNode transformGroupbyTo2P(GroupbyNode gp) {
- Preconditions.checkNotNull(gp);
-
- try {
- // cloning groupby node
- GroupbyNode child = (GroupbyNode) gp.clone();
-
- List<QueryBlock.Target> newChildTargets = Lists.newArrayList();
- QueryBlock.Target[] secondTargets = gp.getTargets();
- QueryBlock.Target[] firstTargets = child.getTargets();
-
- QueryBlock.Target second;
- QueryBlock.Target first;
- int firstTargetId = 0;
- for (int i = 0; i < firstTargets.length; i++) {
- second = secondTargets[i];
- first = firstTargets[i];
-
- List<AggFuncCallEval> secondFuncs = EvalTreeUtil
- .findDistinctAggFunction(second.getEvalTree());
- List<AggFuncCallEval> firstFuncs = EvalTreeUtil.findDistinctAggFunction(first.getEvalTree());
-
- if (firstFuncs.size() == 0) {
- newChildTargets.add(first);
- firstTargetId++;
- } else {
- for (AggFuncCallEval func : firstFuncs) {
- func.setFirstPhase();
- QueryBlock.Target
- newTarget = new QueryBlock.Target(func, firstTargetId++);
- newTarget.setAlias("column_"+ firstTargetId);
-
-
- AggFuncCallEval secondFunc = null;
- for (AggFuncCallEval sf : secondFuncs) {
- if (func.equals(sf)) {
- secondFunc = sf;
- break;
- }
- }
- if (func.getValueType().length > 1) { // hack for partial result
- secondFunc.setArgs(new EvalNode[] {new FieldEval(
- new Column("column_"+firstTargetId, Type.ARRAY))});
- } else {
- secondFunc.setArgs(new EvalNode [] {new FieldEval(
- new Column("column_"+firstTargetId, newTarget.getEvalTree().getValueType()[0]))});
- }
- newChildTargets.add(newTarget);
- }
- }
- }
-
- QueryBlock.Target[] targetArray = newChildTargets.toArray(new QueryBlock.Target[newChildTargets.size()]);
- child.setTargetList(targetArray);
- child.setOutSchema(PlannerUtil.targetToSchema(targetArray));
- // set the groupby chaining
- gp.setSubNode(child);
- gp.setInSchema(child.getOutSchema());
- } catch (CloneNotSupportedException e) {
- LOG.error(e);
- }
-
- return gp;
- }
-
- public static LogicalNode transformSortTo2P(SortNode sort) {
- Preconditions.checkNotNull(sort);
-
- try {
- SortNode child = (SortNode) sort.clone();
- sort.setSubNode(child);
- sort.setInSchema(child.getOutSchema());
- sort.setOutSchema(child.getOutSchema());
- } catch (CloneNotSupportedException e) {
- LOG.error(e);
- }
- return sort;
- }
-
- public static LogicalNode transformGroupbyTo2PWithStore(GroupbyNode gb,
- String tableId) {
- GroupbyNode groupby = (GroupbyNode) transformGroupbyTo2P(gb);
- return insertStore(groupby, tableId);
- }
-
- public static LogicalNode transformSortTo2PWithStore(SortNode sort,
- String tableId) {
- SortNode sort2p = (SortNode) transformSortTo2P(sort);
- return insertStore(sort2p, tableId);
- }
-
- private static LogicalNode insertStore(LogicalNode parent,
- String tableId) {
- StoreTableNode store = new StoreTableNode(tableId);
- store.setLocal(true);
- insertNode(parent, store);
-
- return parent;
- }
-
- /**
- * Find the top logical node matched to type from the given node
- *
- * @param node start node
- * @param type to find
- * @return a found logical node
- */
- public static LogicalNode findTopNode(LogicalNode node, ExprType type) {
- Preconditions.checkNotNull(node);
- Preconditions.checkNotNull(type);
-
- LogicalNodeFinder finder = new LogicalNodeFinder(type);
- node.postOrder(finder);
-
- if (finder.getFoundNodes().size() == 0) {
- return null;
- }
- return finder.getFoundNodes().get(0);
- }
-
- /**
- * Find the all logical node matched to type from the given node
- *
- * @param node start node
- * @param type to find
- * @return a found logical node
- */
- public static LogicalNode [] findAllNodes(LogicalNode node, ExprType type) {
- Preconditions.checkNotNull(node);
- Preconditions.checkNotNull(type);
-
- LogicalNodeFinder finder = new LogicalNodeFinder(type);
- node.postOrder(finder);
-
- if (finder.getFoundNodes().size() == 0) {
- return new LogicalNode[] {};
- }
- List<LogicalNode> founds = finder.getFoundNodes();
- return founds.toArray(new LogicalNode[founds.size()]);
- }
-
- /**
- * Find a parent node of a given-typed operator.
- *
- * @param node start node
- * @param type to find
- * @return the parent node of a found logical node
- */
- public static LogicalNode findTopParentNode(LogicalNode node, ExprType type) {
- Preconditions.checkNotNull(node);
- Preconditions.checkNotNull(type);
-
- ParentNodeFinder finder = new ParentNodeFinder(type);
- node.postOrder(finder);
-
- if (finder.getFoundNodes().size() == 0) {
- return null;
- }
- return finder.getFoundNodes().get(0);
- }
-
- private static class LogicalNodeFinder implements LogicalNodeVisitor {
- private List<LogicalNode> list = new ArrayList<LogicalNode>();
- private final ExprType [] tofind;
- private boolean topmost = false;
- private boolean finished = false;
-
- public LogicalNodeFinder(ExprType...type) {
- this.tofind = type;
- }
-
- public LogicalNodeFinder(ExprType [] type, boolean topmost) {
- this(type);
- this.topmost = topmost;
- }
-
- @Override
- public void visit(LogicalNode node) {
- if (!finished) {
- for (ExprType type : tofind) {
- if (node.getType() == type) {
- list.add(node);
- }
- if (topmost && list.size() > 0) {
- finished = true;
- }
- }
- }
- }
-
- public List<LogicalNode> getFoundNodes() {
- return list;
- }
- }
-
- private static class ParentNodeFinder implements LogicalNodeVisitor {
- private List<LogicalNode> list = new ArrayList<LogicalNode>();
- private ExprType tofind;
-
- public ParentNodeFinder(ExprType type) {
- this.tofind = type;
- }
-
- @Override
- public void visit(LogicalNode node) {
- if (node instanceof UnaryNode) {
- UnaryNode unary = (UnaryNode) node;
- if (unary.getSubNode().getType() == tofind) {
- list.add(node);
- }
- } else if (node instanceof BinaryNode){
- BinaryNode bin = (BinaryNode) node;
- if (bin.getOuterNode().getType() == tofind ||
- bin.getInnerNode().getType() == tofind) {
- list.add(node);
- }
- }
- }
-
- public List<LogicalNode> getFoundNodes() {
- return list;
- }
- }
-
- public static Set<Column> collectColumnRefs(LogicalNode node) {
- ColumnRefCollector collector = new ColumnRefCollector();
- node.postOrder(collector);
- return collector.getColumns();
- }
-
- private static class ColumnRefCollector implements LogicalNodeVisitor {
- private Set<Column> collected = Sets.newHashSet();
-
- public Set<Column> getColumns() {
- return this.collected;
- }
-
- @Override
- public void visit(LogicalNode node) {
- Set<Column> temp;
- switch (node.getType()) {
- case PROJECTION:
- ProjectionNode projNode = (ProjectionNode) node;
-
- for (QueryBlock.Target t : projNode.getTargets()) {
- temp = EvalTreeUtil.findDistinctRefColumns(t.getEvalTree());
- if (!temp.isEmpty()) {
- collected.addAll(temp);
- }
- }
-
- break;
-
- case SELECTION:
- SelectionNode selNode = (SelectionNode) node;
- temp = EvalTreeUtil.findDistinctRefColumns(selNode.getQual());
- if (!temp.isEmpty()) {
- collected.addAll(temp);
- }
-
- break;
-
- case GROUP_BY:
- GroupbyNode groupByNode = (GroupbyNode)node;
- collected.addAll(Lists.newArrayList(groupByNode.getGroupingColumns()));
- for (QueryBlock.Target t : groupByNode.getTargets()) {
- temp = EvalTreeUtil.findDistinctRefColumns(t.getEvalTree());
- if (!temp.isEmpty()) {
- collected.addAll(temp);
- }
- }
- if(groupByNode.hasHavingCondition()) {
- temp = EvalTreeUtil.findDistinctRefColumns(groupByNode.
- getHavingCondition());
- if (!temp.isEmpty()) {
- collected.addAll(temp);
- }
- }
-
- break;
-
- case SORT:
- SortNode sortNode = (SortNode) node;
- for (SortSpec key : sortNode.getSortKeys()) {
- collected.add(key.getSortKey());
- }
-
- break;
-
- case JOIN:
- JoinNode joinNode = (JoinNode) node;
- if (joinNode.hasJoinQual()) {
- temp = EvalTreeUtil.findDistinctRefColumns(joinNode.getJoinQual());
- if (!temp.isEmpty()) {
- collected.addAll(temp);
- }
- }
-
- break;
-
- case SCAN:
- ScanNode scanNode = (ScanNode) node;
- if (scanNode.hasQual()) {
- temp = EvalTreeUtil.findDistinctRefColumns(scanNode.getQual());
- if (!temp.isEmpty()) {
- collected.addAll(temp);
- }
- }
-
- break;
-
- default:
- }
- }
- }
-
- public static QueryBlock.Target[] schemaToTargets(Schema schema) {
- QueryBlock.Target[] targets = new QueryBlock.Target[schema.getColumnNum()];
-
- FieldEval eval;
- for (int i = 0; i < schema.getColumnNum(); i++) {
- eval = new FieldEval(schema.getColumn(i));
- targets[i] = new QueryBlock.Target(eval, i);
- }
- return targets;
- }
-
- public static SortSpec[] schemaToSortSpecs(Schema schema) {
- SortSpec[] specs = new SortSpec[schema.getColumnNum()];
-
- for (int i = 0; i < schema.getColumnNum(); i++) {
- specs[i] = new SortSpec(schema.getColumn(i), true, false);
- }
-
- return specs;
- }
-
- public static Schema sortSpecsToSchema(SortSpec[] sortSpecs) {
- Schema schema = new Schema();
- for (SortSpec spec : sortSpecs) {
- schema.addColumn(spec.getSortKey());
- }
-
- return schema;
- }
-
- /**
- * is it join qual or not?
- * TODO - this method does not support the self join (NTA-740)
- * @param qual
- * @return true if two operands refers to columns and the operator is comparison,
- */
- public static boolean isJoinQual(EvalNode qual) {
- if (EvalTreeUtil.isComparisonOperator(qual)) {
- List<Column> left = EvalTreeUtil.findAllColumnRefs(qual.getLeftExpr());
- List<Column> right = EvalTreeUtil.findAllColumnRefs(qual.getRightExpr());
-
- if (left.size() == 1 && right.size() == 1 &&
- !left.get(0).getTableName().equals(right.get(0).getTableName()))
- return true;
- }
-
- return false;
- }
-
- public static SortSpec[][] getSortKeysFromJoinQual(EvalNode joinQual, Schema outer, Schema inner) {
- List<Column []> joinKeyPairs = getJoinKeyPairs(joinQual, outer, inner);
- SortSpec[] outerSortSpec = new SortSpec[joinKeyPairs.size()];
- SortSpec[] innerSortSpec = new SortSpec[joinKeyPairs.size()];
-
- for (int i = 0; i < joinKeyPairs.size(); i++) {
- outerSortSpec[i] = new SortSpec(joinKeyPairs.get(i)[0]);
- innerSortSpec[i] = new SortSpec(joinKeyPairs.get(i)[1]);
- }
-
- return new SortSpec[][] {outerSortSpec, innerSortSpec};
- }
-
- public static TupleComparator[] getComparatorsFromJoinQual(EvalNode joinQual, Schema outer, Schema inner) {
- SortSpec[][] sortSpecs = getSortKeysFromJoinQual(joinQual, outer, inner);
- TupleComparator [] comparators = new TupleComparator[2];
- comparators[0] = new TupleComparator(outer, sortSpecs[0]);
- comparators[1] = new TupleComparator(inner, sortSpecs[1]);
- return comparators;
- }
-
- public static List<Column []> getJoinKeyPairs(EvalNode joinQual, Schema outer, Schema inner) {
- JoinKeyPairFinder finder = new JoinKeyPairFinder(outer, inner);
- joinQual.preOrder(finder);
- return finder.getPairs();
- }
-
- public static class JoinKeyPairFinder implements EvalNodeVisitor {
- private final List<Column []> pairs = Lists.newArrayList();
- private Schema [] schemas = new Schema[2];
-
- public JoinKeyPairFinder(Schema outer, Schema inner) {
- schemas[0] = outer;
- schemas[1] = inner;
- }
-
- @Override
- public void visit(EvalNode node) {
- if (EvalTreeUtil.isJoinQual(node)) {
- Column [] pair = new Column[2];
-
- for (int i = 0; i <= 1; i++) { // access left, right sub expression
- Column column = EvalTreeUtil.findAllColumnRefs(node.getExpr(i)).get(0);
- for (int j = 0; j < schemas.length; j++) {
- // check whether the column is for either outer or inner
- // 0 is outer, and 1 is inner
- if (schemas[j].contains(column.getQualifiedName())) {
- pair[j] = column;
- }
- }
- }
-
- if (pair[0] == null || pair[1] == null) {
- throw new IllegalStateException("Wrong join key: " + node);
- }
- pairs.add(pair);
- }
- }
-
- public List<Column []> getPairs() {
- return this.pairs;
- }
- }
-
- public static Schema targetToSchema(QueryBlock.Target[] targets) {
- Schema schema = new Schema();
- for(QueryBlock.Target t : targets) {
- DataType type;
- if (t.getEvalTree().getValueType().length > 1) {
- type = CatalogUtil.newDataTypeWithoutLen(Type.ARRAY);
- } else {
- type = t.getEvalTree().getValueType()[0];
- }
- String name;
- if (t.hasAlias()) {
- name = t.getAlias();
- } else {
- name = t.getEvalTree().getName();
- }
- schema.addColumn(name, type);
- }
-
- return schema;
- }
-
- public static EvalNode [] columnsToEvals(Column [] columns) {
- EvalNode [] exprs = new EvalNode[columns.length];
- for (int i = 0; i < columns.length; i++) {
- exprs[i] = new FieldEval(columns[i]);
- }
- return exprs;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/bc6359b8/tajo-core/tajo-core-backend/src/main/java/tajo/engine/planner/PlanningContext.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/tajo/engine/planner/PlanningContext.java b/tajo-core/tajo-core-backend/src/main/java/tajo/engine/planner/PlanningContext.java
deleted file mode 100644
index e62c237..0000000
--- a/tajo-core/tajo-core-backend/src/main/java/tajo/engine/planner/PlanningContext.java
+++ /dev/null
@@ -1,40 +0,0 @@
-/**
- * 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 tajo.engine.planner;
-
-import org.antlr.runtime.tree.Tree;
-import tajo.engine.parser.ParseTree;
-
-public interface PlanningContext {
- String getRawQuery();
-
- Tree getAST();
-
- ParseTree getParseTree();
-
- //TableMap getTableMap();
-
- String getGeneratedColumnName();
-
- String getGeneratedColumnName(String prefix);
-
- String getExplicitOutputTable();
-
- boolean hasExplicitOutputTable();
-}