[40/51] [partial] TAJO-752: Escalate sub modules in tajo-core into the top-level modules. (hyunsik)

[hy...@apache.org]

http://git-wip-us.apache.org/repos/asf/tajo/blob/6594ac1c/tajo-core/src/main/java/org/apache/tajo/engine/planner/BaseAlgebraVisitor.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/BaseAlgebraVisitor.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/BaseAlgebraVisitor.java
new file mode 100644
index 0000000..b8f3311
--- /dev/null
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/BaseAlgebraVisitor.java
@@ -0,0 +1,739 @@
+/**
+ * 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.engine.planner;
+
+import org.apache.tajo.algebra.*;
+
+import java.util.Stack;
+
+public class BaseAlgebraVisitor<CONTEXT, RESULT> implements AlgebraVisitor<CONTEXT, RESULT> {
+
+  /**
+   * The prehook is called before each expression is visited.
+   */
+  public void preHook(CONTEXT ctx, Stack<Expr> stack, Expr expr) throws PlanningException {
+  }
+
+
+  /**
+   * The posthook is called before each expression is visited.
+   */
+  public RESULT postHook(CONTEXT ctx, Stack<Expr> stack, Expr expr, RESULT current) throws PlanningException {
+    return current;
+  }
+
+  /**
+   * visit visits each relational operator expression recursively.
+   *
+   * @param stack The stack contains the upper operators' type.
+   * @param expr The visiting relational operator
+   */
+  public RESULT visit(CONTEXT ctx, Stack<Expr> stack, Expr expr) throws PlanningException {
+    preHook(ctx, stack, expr);
+
+    RESULT current;
+
+    switch (expr.getType()) {
+
+    case Projection:
+      current = visitProjection(ctx, stack, (Projection) expr);
+      break;
+    case Limit:
+      current = visitLimit(ctx, stack, (Limit) expr);
+      break;
+    case Sort:
+      current = visitSort(ctx, stack, (Sort) expr);
+      break;
+    case Having:
+      current = visitHaving(ctx, stack, (Having) expr);
+      break;
+    case Aggregation:
+      current = visitGroupBy(ctx, stack, (Aggregation) expr);
+      break;
+    case Join:
+      current = visitJoin(ctx, stack, (Join) expr);
+      break;
+    case Filter:
+      current = visitFilter(ctx, stack, (Selection) expr);
+      break;
+    case Union:
+      current = visitUnion(ctx, stack, (SetOperation) expr);
+      break;
+    case Except:
+      current = visitExcept(ctx, stack, (SetOperation) expr);
+      break;
+    case Intersect:
+      current = visitIntersect(ctx, stack, (SetOperation) expr);
+      break;
+    case SimpleTableSubQuery:
+      current = visitSimpleTableSubQuery(ctx, stack, (SimpleTableSubQuery) expr);
+      break;
+    case TablePrimaryTableSubQuery:
+      current = visitTableSubQuery(ctx, stack, (TablePrimarySubQuery) expr);
+      break;
+    case RelationList:
+      current = visitRelationList(ctx, stack, (RelationList) expr);
+      break;
+    case Relation:
+      current = visitRelation(ctx, stack, (Relation) expr);
+      break;
+    case ScalarSubQuery:
+      current = visitScalarSubQuery(ctx, stack, (ScalarSubQuery) expr);
+      break;
+    case Explain:
+      current = visitExplain(ctx, stack, (Explain) expr);
+      break;
+
+    case CreateDatabase:
+      current = visitCreateDatabase(ctx, stack, (CreateDatabase) expr);
+      break;
+    case DropDatabase:
+      current = visitDropDatabase(ctx, stack, (DropDatabase) expr);
+      break;
+    case CreateTable:
+      current = visitCreateTable(ctx, stack, (CreateTable) expr);
+      break;
+    case DropTable:
+      current = visitDropTable(ctx, stack, (DropTable) expr);
+      break;
+    case AlterTablespace:
+      current = visitAlterTablespace(ctx, stack, (AlterTablespace) expr);
+      break;
+    case AlterTable:
+      current = visitAlterTable(ctx, stack, (AlterTable) expr);
+      break;
+
+    case Insert:
+      current = visitInsert(ctx, stack, (Insert) expr);
+      break;
+
+    case And:
+      current = visitAnd(ctx, stack, (BinaryOperator) expr);
+      break;
+    case Or:
+      current = visitOr(ctx, stack, (BinaryOperator) expr);
+      break;
+    case Not:
+      current = visitNot(ctx, stack, (NotExpr) expr);
+      break;
+
+    case Equals:
+      current = visitEquals(ctx, stack, (BinaryOperator) expr);
+      break;
+    case NotEquals:
+      current = visitNotEquals(ctx, stack, (BinaryOperator) expr);
+      break;
+    case LessThan:
+      current = visitLessThan(ctx, stack, (BinaryOperator) expr);
+      break;
+    case LessThanOrEquals:
+      current = visitLessThanOrEquals(ctx, stack, (BinaryOperator) expr);
+      break;
+    case GreaterThan:
+      current = visitGreaterThan(ctx, stack, (BinaryOperator) expr);
+      break;
+    case GreaterThanOrEquals:
+      current = visitGreaterThanOrEquals(ctx, stack, (BinaryOperator) expr);
+      break;
+
+    // Other Predicates
+    case Between:
+      current = visitBetween(ctx, stack, (BetweenPredicate) expr);
+      break;
+    case CaseWhen:
+      current = visitCaseWhen(ctx, stack, (CaseWhenPredicate) expr);
+      break;
+    case IsNullPredicate:
+      current = visitIsNullPredicate(ctx, stack, (IsNullPredicate) expr);
+      break;
+    case InPredicate:
+      current = visitInPredicate(ctx, stack, (InPredicate) expr);
+      break;
+    case ValueList:
+      current = visitValueListExpr(ctx, stack, (ValueListExpr) expr);
+      break;
+    case ExistsPredicate:
+      current = visitExistsPredicate(ctx, stack, (ExistsPredicate) expr);
+      break;
+
+    // String Operator or Pattern Matching Predicates
+    case LikePredicate:
+      current = visitLikePredicate(ctx, stack, (PatternMatchPredicate) expr);
+      break;
+    case SimilarToPredicate:
+      current = visitSimilarToPredicate(ctx, stack, (PatternMatchPredicate) expr);
+      break;
+    case Regexp:
+      current = visitRegexpPredicate(ctx, stack, (PatternMatchPredicate) expr);
+      break;
+    case Concatenate:
+      current = visitConcatenate(ctx, stack, (BinaryOperator) expr);
+      break;
+
+    // Arithmetic Operators
+    case Plus:
+      current = visitPlus(ctx, stack, (BinaryOperator) expr);
+      break;
+    case Minus:
+      current = visitMinus(ctx, stack, (BinaryOperator) expr);
+      break;
+    case Multiply:
+      current = visitMultiply(ctx, stack, (BinaryOperator) expr);
+      break;
+    case Divide:
+      current = visitDivide(ctx, stack, (BinaryOperator) expr);
+      break;
+    case Modular:
+      current = visitModular(ctx, stack, (BinaryOperator) expr);
+      break;
+
+    // Other Expressions
+    case Sign:
+      current = visitSign(ctx, stack, (SignedExpr) expr);
+      break;
+    case Column:
+      current = visitColumnReference(ctx, stack, (ColumnReferenceExpr) expr);
+      break;
+    case Target:
+      current = visitTargetExpr(ctx, stack, (NamedExpr) expr);
+      break;
+    case Function:
+      current = visitFunction(ctx, stack, (FunctionExpr) expr);
+      break;
+    case Asterisk:
+      current = visitQualifiedAsterisk(ctx, stack, (QualifiedAsteriskExpr) expr);
+      break;
+
+
+    case CountRowsFunction:
+      current = visitCountRowsFunction(ctx, stack, (CountRowsFunctionExpr) expr);
+      break;
+    case GeneralSetFunction:
+      current = visitGeneralSetFunction(ctx, stack, (GeneralSetFunctionExpr) expr);
+      break;
+
+    case DataType:
+      current = visitDataType(ctx, stack, (DataTypeExpr) expr);
+      break;
+    case Cast:
+      current = visitCastExpr(ctx, stack, (CastExpr) expr);
+      break;
+    case Literal:
+      current = visitLiteral(ctx, stack, (LiteralValue) expr);
+      break;
+    case NullLiteral:
+      current = visitNullLiteral(ctx, stack, (NullLiteral) expr);
+      break;
+    case DateLiteral:
+      current = visitDateLiteral(ctx, stack, (DateLiteral) expr);
+      break;
+    case TimeLiteral:
+      current = visitTimeLiteral(ctx, stack, (TimeLiteral) expr);
+      break;
+    case TimestampLiteral:
+      current = visitTimestampLiteral(ctx, stack, (TimestampLiteral) expr);
+      break;
+
+
+
+    default:
+      throw new PlanningException("Cannot support this type algebra \"" + expr.getType() + "\"");
+    }
+
+    // skip postHook against only one relation
+    if (expr.getType() == OpType.RelationList) {
+      RelationList relationList = (RelationList)expr;
+      if (relationList.size() == 1 && relationList.getRelations()[0].getType() == OpType.Relation) {
+        return current;
+      }
+    }
+
+    postHook(ctx, stack, expr, current);
+    return current;
+  }
+
+  private RESULT visitDefaultUnaryExpr(CONTEXT ctx, Stack<Expr> stack, UnaryOperator expr) throws PlanningException {
+    stack.push(expr);
+    RESULT child = visit(ctx, stack, expr.getChild());
+    stack.pop();
+    return child;
+  }
+
+  private RESULT visitDefaultBinaryExpr(CONTEXT ctx, Stack<Expr> stack, BinaryOperator expr)
+      throws PlanningException {
+    stack.push(expr);
+    RESULT child = visit(ctx, stack, expr.getLeft());
+    visit(ctx, stack, expr.getRight());
+    stack.pop();
+    return child;
+  }
+
+  @Override
+  public RESULT visitProjection(CONTEXT ctx, Stack<Expr> stack, Projection expr) throws PlanningException {
+    stack.push(expr);
+    try {
+      for (NamedExpr target : expr.getNamedExprs()) {
+        visit(ctx, stack, target);
+      }
+      if (expr.hasChild()) {
+        return visit(ctx, stack, expr.getChild());
+      }
+    } finally {
+      stack.pop();
+    }
+    return null;
+  }
+
+  @Override
+  public RESULT visitLimit(CONTEXT ctx, Stack<Expr> stack, Limit expr) throws PlanningException {
+    stack.push(expr);
+    visit(ctx, stack, expr.getFetchFirstNum());
+    RESULT result = visit(ctx, stack, expr.getChild());
+    stack.pop();
+    return result;
+  }
+
+  @Override
+  public RESULT visitSort(CONTEXT ctx, Stack<Expr> stack, Sort expr) throws PlanningException {
+    stack.push(expr);
+    for (Sort.SortSpec sortSpec : expr.getSortSpecs()) {
+      visit(ctx, stack, sortSpec.getKey());
+    }
+    RESULT result = visit(ctx, stack, expr.getChild());
+    return result;
+  }
+
+  @Override
+  public RESULT visitHaving(CONTEXT ctx, Stack<Expr> stack, Having expr) throws PlanningException {
+    stack.push(expr);
+    visit(ctx, stack, expr.getQual());
+    RESULT result = visit(ctx, stack, expr.getChild());
+    stack.pop();
+    return result;
+  }
+
+  @Override
+  public RESULT visitGroupBy(CONTEXT ctx, Stack<Expr> stack, Aggregation expr) throws PlanningException {
+    stack.push(expr);
+
+    for (org.apache.tajo.algebra.Aggregation.GroupElement groupElement : expr.getGroupSet()) {
+      for (Expr groupingSet : groupElement.getGroupingSets()) {
+        visit(ctx, stack, groupingSet);
+      }
+    }
+
+    RESULT result = visit(ctx, stack, expr.getChild());
+    stack.pop();
+    return result;
+  }
+
+  @Override
+  public RESULT visitJoin(CONTEXT ctx, Stack<Expr> stack, Join expr) throws PlanningException {
+    stack.push(expr);
+    visit(ctx, stack, expr.getQual());
+    visit(ctx, stack, expr.getLeft());
+    RESULT result = visit(ctx, stack, expr.getRight());
+    stack.pop();
+    return result;
+  }
+
+  @Override
+  public RESULT visitFilter(CONTEXT ctx, Stack<Expr> stack, Selection expr) throws PlanningException {
+    stack.push(expr);
+    visit(ctx, stack, expr.getQual());
+    RESULT result = visit(ctx, stack, expr.getChild());
+    stack.pop();
+    return result;
+  }
+
+  @Override
+  public RESULT visitUnion(CONTEXT ctx, Stack<Expr> stack, SetOperation expr) throws PlanningException {
+    return visitDefaultBinaryExpr(ctx, stack, expr);
+  }
+
+  @Override
+  public RESULT visitExcept(CONTEXT ctx, Stack<Expr> stack, SetOperation expr) throws PlanningException {
+    return visitDefaultBinaryExpr(ctx, stack, expr);
+  }
+
+  @Override
+  public RESULT visitIntersect(CONTEXT ctx, Stack<Expr> stack, SetOperation expr) throws PlanningException {
+    return visitDefaultBinaryExpr(ctx, stack, expr);
+  }
+
+  @Override
+  public RESULT visitSimpleTableSubQuery(CONTEXT ctx, Stack<Expr> stack, SimpleTableSubQuery expr)
+      throws PlanningException {
+    return visitDefaultUnaryExpr(ctx, stack, expr);
+  }
+
+  @Override
+  public RESULT visitTableSubQuery(CONTEXT ctx, Stack<Expr> stack, TablePrimarySubQuery expr)
+      throws PlanningException {
+    stack.push(expr);
+    RESULT child = visit(ctx, stack, expr.getSubQuery());
+    stack.pop();
+    return child;
+  }
+
+  @Override
+  public RESULT visitRelationList(CONTEXT ctx, Stack<Expr> stack, RelationList expr) throws PlanningException {
+    stack.push(expr);
+    RESULT child = null;
+    for (Expr e : expr.getRelations()) {
+      child = visit(ctx, stack, e);
+    }
+    stack.pop();
+    return child;
+  }
+
+  @Override
+  public RESULT visitRelation(CONTEXT ctx, Stack<Expr> stack, Relation expr) throws PlanningException {
+    return null;
+  }
+
+  @Override
+  public RESULT visitScalarSubQuery(CONTEXT ctx, Stack<Expr> stack, ScalarSubQuery expr) throws PlanningException {
+    return visitDefaultUnaryExpr(ctx, stack, expr);
+  }
+
+  @Override
+  public RESULT visitExplain(CONTEXT ctx, Stack<Expr> stack, Explain expr) throws PlanningException {
+    stack.push(expr);
+    RESULT child = visit(ctx, stack, expr.getChild());
+    stack.pop();
+    return child;
+  }
+
+  ///////////////////////////////////////////////////////////////////////////////////////////////////////////
+  // Data Definition Language Section
+  ///////////////////////////////////////////////////////////////////////////////////////////////////////////
+
+  @Override
+  public RESULT visitCreateDatabase(CONTEXT ctx, Stack<Expr> stack, CreateDatabase expr) throws PlanningException {
+    return null;
+  }
+
+  @Override
+  public RESULT visitDropDatabase(CONTEXT ctx, Stack<Expr> stack, DropDatabase expr) throws PlanningException {
+    return null;
+  }
+
+  @Override
+  public RESULT visitCreateTable(CONTEXT ctx, Stack<Expr> stack, CreateTable expr) throws PlanningException {
+    stack.push(expr);
+    RESULT child = null;
+    if (expr.hasSubQuery()) {
+      child = visit(ctx, stack, expr.getSubQuery());
+    }
+    stack.pop();
+    return child;
+  }
+
+  @Override
+  public RESULT visitDropTable(CONTEXT ctx, Stack<Expr> stack, DropTable expr) throws PlanningException {
+    return null;
+  }
+
+  @Override
+  public RESULT visitAlterTablespace(CONTEXT ctx, Stack<Expr> stack, AlterTablespace expr) throws PlanningException {
+    return null;
+  }
+
+  @Override
+  public RESULT visitAlterTable(CONTEXT ctx, Stack<Expr> stack, AlterTable expr) throws PlanningException {
+    return null;
+  }
+
+  ///////////////////////////////////////////////////////////////////////////////////////////////////////////
+  // Insert or Update Section
+  ///////////////////////////////////////////////////////////////////////////////////////////////////////////
+
+  public RESULT visitInsert(CONTEXT ctx, Stack<Expr> stack, Insert expr) throws PlanningException {
+    stack.push(expr);
+    RESULT child = visit(ctx, stack, expr.getSubQuery());
+    stack.pop();
+    return child;
+  }
+
+  ///////////////////////////////////////////////////////////////////////////////////////////////////////////
+  // Logical Operator Section
+  ///////////////////////////////////////////////////////////////////////////////////////////////////////////
+
+  @Override
+  public RESULT visitAnd(CONTEXT ctx, Stack<Expr> stack, BinaryOperator expr) throws PlanningException {
+    return visitDefaultBinaryExpr(ctx, stack, expr);
+  }
+
+  @Override
+  public RESULT visitOr(CONTEXT ctx, Stack<Expr> stack, BinaryOperator expr) throws PlanningException {
+    return visitDefaultBinaryExpr(ctx, stack, expr);
+  }
+
+  @Override
+  public RESULT visitNot(CONTEXT ctx, Stack<Expr> stack, NotExpr expr) throws PlanningException {
+    return visitDefaultUnaryExpr(ctx, stack, expr);
+  }
+
+  ///////////////////////////////////////////////////////////////////////////////////////////////////////////
+  // Comparison Predicates Section
+  ///////////////////////////////////////////////////////////////////////////////////////////////////////////
+  @Override
+  public RESULT visitEquals(CONTEXT ctx, Stack<Expr> stack, BinaryOperator expr) throws PlanningException {
+    return visitDefaultBinaryExpr(ctx, stack, expr);
+  }
+
+  @Override
+  public RESULT visitNotEquals(CONTEXT ctx, Stack<Expr> stack, BinaryOperator expr) throws PlanningException {
+    return visitDefaultBinaryExpr(ctx, stack, expr);
+  }
+
+  @Override
+  public RESULT visitLessThan(CONTEXT ctx, Stack<Expr> stack, BinaryOperator expr) throws PlanningException {
+    return visitDefaultBinaryExpr(ctx, stack, expr);
+  }
+
+  @Override
+  public RESULT visitLessThanOrEquals(CONTEXT ctx, Stack<Expr> stack, BinaryOperator expr) throws PlanningException {
+    return visitDefaultBinaryExpr(ctx, stack, expr);
+  }
+
+  @Override
+  public RESULT visitGreaterThan(CONTEXT ctx, Stack<Expr> stack, BinaryOperator expr) throws PlanningException {
+    return visitDefaultBinaryExpr(ctx, stack, expr);
+  }
+
+  @Override
+  public RESULT visitGreaterThanOrEquals(CONTEXT ctx, Stack<Expr> stack, BinaryOperator expr)
+      throws PlanningException {
+    return visitDefaultBinaryExpr(ctx, stack, expr);
+  }
+
+  ///////////////////////////////////////////////////////////////////////////////////////////////////////////
+  // Other Predicates Section
+  ///////////////////////////////////////////////////////////////////////////////////////////////////////////
+
+  @Override
+  public RESULT visitBetween(CONTEXT ctx, Stack<Expr> stack, BetweenPredicate expr) throws PlanningException {
+    stack.push(expr);
+    RESULT result = visit(ctx, stack, expr.predicand());
+    visit(ctx, stack, expr.begin());
+    visit(ctx, stack, expr.end());
+    stack.pop();
+    return result;
+  }
+
+  @Override
+  public RESULT visitCaseWhen(CONTEXT ctx, Stack<Expr> stack, CaseWhenPredicate expr) throws PlanningException {
+    stack.push(expr);
+    RESULT result = null;
+    for (CaseWhenPredicate.WhenExpr when : expr.getWhens()) {
+      result = visit(ctx, stack, when.getCondition());
+      visit(ctx, stack, when.getResult());
+    }
+    if (expr.hasElseResult()) {
+      visit(ctx, stack, expr.getElseResult());
+    }
+    stack.pop();
+    return result;
+  }
+
+  @Override
+  public RESULT visitIsNullPredicate(CONTEXT ctx, Stack<Expr> stack, IsNullPredicate expr) throws PlanningException {
+    return visitDefaultUnaryExpr(ctx, stack, expr);
+  }
+
+  @Override
+  public RESULT visitInPredicate(CONTEXT ctx, Stack<Expr> stack, InPredicate expr) throws PlanningException {
+    return visitDefaultBinaryExpr(ctx, stack, expr);
+  }
+
+  @Override
+  public RESULT visitValueListExpr(CONTEXT ctx, Stack<Expr> stack, ValueListExpr expr) throws PlanningException {
+    stack.push(expr);
+    RESULT result = null;
+    for (Expr value : expr.getValues()) {
+      result = visit(ctx, stack, value);
+    }
+    stack.pop();
+    return result;
+  }
+
+  @Override
+  public RESULT visitExistsPredicate(CONTEXT ctx, Stack<Expr> stack, ExistsPredicate expr) throws PlanningException {
+    return visitDefaultUnaryExpr(ctx, stack, expr);
+  }
+
+  ///////////////////////////////////////////////////////////////////////////////////////////////////////////
+  // String Operator or Pattern Matching Predicates Section
+  ///////////////////////////////////////////////////////////////////////////////////////////////////////////
+  @Override
+  public RESULT visitLikePredicate(CONTEXT ctx, Stack<Expr> stack, PatternMatchPredicate expr)
+      throws PlanningException {
+    return visitDefaultBinaryExpr(ctx, stack, expr);
+  }
+
+  @Override
+  public RESULT visitSimilarToPredicate(CONTEXT ctx, Stack<Expr> stack, PatternMatchPredicate expr)
+      throws PlanningException {
+    return visitDefaultBinaryExpr(ctx, stack, expr);
+  }
+
+  @Override
+  public RESULT visitRegexpPredicate(CONTEXT ctx, Stack<Expr> stack, PatternMatchPredicate expr)
+      throws PlanningException {
+    return visitDefaultBinaryExpr(ctx, stack, expr);
+  }
+
+  @Override
+  public RESULT visitConcatenate(CONTEXT ctx, Stack<Expr> stack, BinaryOperator expr) throws PlanningException {
+    return visitDefaultBinaryExpr(ctx, stack, expr);
+  }
+
+  ///////////////////////////////////////////////////////////////////////////////////////////////////////////
+  // Arithmetic Operators
+  ///////////////////////////////////////////////////////////////////////////////////////////////////////////
+
+  @Override
+  public RESULT visitPlus(CONTEXT ctx, Stack<Expr> stack, BinaryOperator expr) throws PlanningException {
+    return visitDefaultBinaryExpr(ctx, stack, expr);
+  }
+
+  @Override
+  public RESULT visitMinus(CONTEXT ctx, Stack<Expr> stack, BinaryOperator expr) throws PlanningException {
+    return visitDefaultBinaryExpr(ctx, stack, expr);
+  }
+
+  @Override
+  public RESULT visitMultiply(CONTEXT ctx, Stack<Expr> stack, BinaryOperator expr) throws PlanningException {
+    return visitDefaultBinaryExpr(ctx, stack, expr);
+  }
+
+  @Override
+  public RESULT visitDivide(CONTEXT ctx, Stack<Expr> stack, BinaryOperator expr) throws PlanningException {
+    return visitDefaultBinaryExpr(ctx, stack, expr);
+  }
+
+  @Override
+  public RESULT visitModular(CONTEXT ctx, Stack<Expr> stack, BinaryOperator expr) throws PlanningException {
+    return visitDefaultBinaryExpr(ctx, stack, expr);
+  }
+
+  ///////////////////////////////////////////////////////////////////////////////////////////////////////////
+  // Other Expressions
+  ///////////////////////////////////////////////////////////////////////////////////////////////////////////
+
+  @Override
+  public RESULT visitSign(CONTEXT ctx, Stack<Expr> stack, SignedExpr expr) throws PlanningException {
+    return visitDefaultUnaryExpr(ctx, stack, expr);
+  }
+
+  @Override
+  public RESULT visitColumnReference(CONTEXT ctx, Stack<Expr> stack, ColumnReferenceExpr expr)
+      throws PlanningException {
+    return null;
+  }
+
+  @Override
+  public RESULT visitTargetExpr(CONTEXT ctx, Stack<Expr> stack, NamedExpr expr) throws PlanningException {
+    return visitDefaultUnaryExpr(ctx, stack, expr);
+  }
+
+  @Override
+  public RESULT visitFunction(CONTEXT ctx, Stack<Expr> stack, FunctionExpr expr) throws PlanningException {
+    stack.push(expr);
+    RESULT result = null;
+    if (expr.hasParams()) {
+      for (Expr param : expr.getParams()) {
+        result = visit(ctx, stack, param);
+      }
+    }
+    stack.pop();
+    return result;
+  }
+
+  @Override
+  public RESULT visitQualifiedAsterisk(CONTEXT ctx, Stack<Expr> stack, QualifiedAsteriskExpr expr) throws PlanningException {
+    return null;
+  }
+
+  ///////////////////////////////////////////////////////////////////////////////////////////////////////////
+  // General Set Section
+  ///////////////////////////////////////////////////////////////////////////////////////////////////////////
+
+  @Override
+  public RESULT visitCountRowsFunction(CONTEXT ctx, Stack<Expr> stack, CountRowsFunctionExpr expr)
+      throws PlanningException {
+    return null;
+  }
+
+  @Override
+  public RESULT visitGeneralSetFunction(CONTEXT ctx, Stack<Expr> stack, GeneralSetFunctionExpr expr)
+      throws PlanningException {
+    stack.push(expr);
+    RESULT result = null;
+    for (Expr param : expr.getParams()) {
+      result = visit(ctx, stack, param);
+    }
+    stack.pop();
+    return result;
+  }
+
+  ///////////////////////////////////////////////////////////////////////////////////////////////////////////
+  // Literal Section
+  ///////////////////////////////////////////////////////////////////////////////////////////////////////////
+
+  @Override
+  public RESULT visitDataType(CONTEXT ctx, Stack<Expr> stack, DataTypeExpr expr) throws PlanningException {
+    return null;
+  }
+
+  @Override
+  public RESULT visitCastExpr(CONTEXT ctx, Stack<Expr> stack, CastExpr expr) throws PlanningException {
+    stack.push(expr);
+    RESULT result = visit(ctx, stack, expr.getOperand());
+    stack.pop();
+    return result;
+  }
+
+  @Override
+  public RESULT visitLiteral(CONTEXT ctx, Stack<Expr> stack, LiteralValue expr) throws PlanningException {
+    return null;
+  }
+
+  @Override
+  public RESULT visitNullLiteral(CONTEXT ctx, Stack<Expr> stack, NullLiteral expr) throws PlanningException {
+    return null;
+  }
+
+  @Override
+  public RESULT visitTimestampLiteral(CONTEXT ctx, Stack<Expr> stack, TimestampLiteral expr) throws PlanningException {
+    return null;
+  }
+
+  @Override
+  public RESULT visitTimeLiteral(CONTEXT ctx, Stack<Expr> stack, TimeLiteral expr) throws PlanningException {
+    return null;
+  }
+
+  @Override
+  public RESULT visitDateLiteral(CONTEXT ctx, Stack<Expr> stack, DateLiteral expr) throws PlanningException {
+    return null;
+  }
+}

http://git-wip-us.apache.org/repos/asf/tajo/blob/6594ac1c/tajo-core/src/main/java/org/apache/tajo/engine/planner/BasicLogicalPlanVisitor.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/BasicLogicalPlanVisitor.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/BasicLogicalPlanVisitor.java
new file mode 100644
index 0000000..0f758bf
--- /dev/null
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/BasicLogicalPlanVisitor.java
@@ -0,0 +1,319 @@
+/**
+ * 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.engine.planner;
+
+import org.apache.tajo.engine.planner.logical.*;
+
+import java.util.Stack;
+
+public class BasicLogicalPlanVisitor<CONTEXT, RESULT> implements LogicalPlanVisitor<CONTEXT, RESULT> {
+
+  /**
+   * The prehook is called before each node is visited.
+   */
+  @SuppressWarnings("unused")
+  public void preHook(LogicalPlan plan, LogicalNode node, Stack<LogicalNode> stack, CONTEXT data)
+      throws PlanningException {
+  }
+
+  /**
+   * The posthook is called after each node is visited.
+   */
+  @SuppressWarnings("unused")
+  public void postHook(LogicalPlan plan, LogicalNode node, Stack<LogicalNode> stack, CONTEXT data)
+      throws PlanningException {
+  }
+
+  public CONTEXT visit(CONTEXT context, LogicalPlan plan, LogicalPlan.QueryBlock block)
+      throws PlanningException {
+    visit(context, plan, block, block.getRoot(), new Stack<LogicalNode>());
+    return context;
+  }
+
+  /**
+   * visit visits each logicalNode recursively.
+   */
+  public RESULT visit(CONTEXT context, LogicalPlan plan, LogicalPlan.QueryBlock block, LogicalNode node,
+                      Stack<LogicalNode> stack)
+      throws PlanningException {
+    RESULT current;
+    switch (node.getType()) {
+      case ROOT:
+        current = visitRoot(context, plan, block, (LogicalRootNode) node, stack);
+        break;
+      case EXPRS:
+        return null;
+      case PROJECTION:
+        current = visitProjection(context, plan, block, (ProjectionNode) node, stack);
+        break;
+      case LIMIT:
+        current = visitLimit(context, plan, block, (LimitNode) node, stack);
+        break;
+      case SORT:
+        current = visitSort(context, plan, block, (SortNode) node, stack);
+        break;
+      case HAVING:
+        current = visitHaving(context, plan, block, (HavingNode) node, stack);
+        break;
+      case GROUP_BY:
+        current = visitGroupBy(context, plan, block, (GroupbyNode) node, stack);
+        break;
+      case SELECTION:
+        current = visitFilter(context, plan, block, (SelectionNode) node, stack);
+        break;
+      case JOIN:
+        current = visitJoin(context, plan, block, (JoinNode) node, stack);
+        break;
+      case UNION:
+        current = visitUnion(context, plan, block, (UnionNode) node, stack);
+        break;
+      case EXCEPT:
+        current = visitExcept(context, plan, block, (ExceptNode) node, stack);
+        break;
+      case INTERSECT:
+        current = visitIntersect(context, plan, block, (IntersectNode) node, stack);
+        break;
+      case TABLE_SUBQUERY:
+        current = visitTableSubQuery(context, plan, block, (TableSubQueryNode) node, stack);
+        break;
+      case SCAN:
+        current = visitScan(context, plan, block, (ScanNode) node, stack);
+        break;
+      case PARTITIONS_SCAN:
+        current = visitPartitionedTableScan(context, plan, block, (PartitionedTableScanNode) node, stack);
+        break;
+      case STORE:
+        current = visitStoreTable(context, plan, block, (StoreTableNode) node, stack);
+        break;
+      case INSERT:
+        current = visitInsert(context, plan, block, (InsertNode) node, stack);
+        break;
+      case CREATE_DATABASE:
+        current = visitCreateDatabase(context, plan, block, (CreateDatabaseNode) node, stack);
+        break;
+      case DROP_DATABASE:
+        current = visitDropDatabase(context, plan, block, (DropDatabaseNode) node, stack);
+        break;
+      case CREATE_TABLE:
+        current = visitCreateTable(context, plan, block, (CreateTableNode) node, stack);
+        break;
+      case DROP_TABLE:
+        current = visitDropTable(context, plan, block, (DropTableNode) node, stack);
+        break;
+      case ALTER_TABLESPACE:
+        current = visitAlterTablespace(context, plan, block, (AlterTablespaceNode) node, stack);
+        break;
+      case ALTER_TABLE:
+        current = visitAlterTable(context, plan, block, (AlterTableNode) node, stack);
+        break;
+      default:
+        throw new PlanningException("Unknown logical node type: " + node.getType());
+    }
+
+    return current;
+  }
+
+  @Override
+  public RESULT visitRoot(CONTEXT context, LogicalPlan plan, LogicalPlan.QueryBlock block, LogicalRootNode node,
+                          Stack<LogicalNode> stack) throws PlanningException {
+    stack.push(node);
+    RESULT result = visit(context, plan, block, node.getChild(), stack);
+    stack.pop();
+    return result;
+  }
+
+  @Override
+  public RESULT visitProjection(CONTEXT context, LogicalPlan plan, LogicalPlan.QueryBlock block, ProjectionNode node,
+                                Stack<LogicalNode> stack)
+      throws PlanningException {
+    stack.push(node);
+    RESULT result = visit(context, plan, block, node.getChild(), stack);
+    stack.pop();
+    return result;
+  }
+
+  @Override
+  public RESULT visitLimit(CONTEXT context, LogicalPlan plan, LogicalPlan.QueryBlock block, LimitNode node,
+                           Stack<LogicalNode> stack) throws PlanningException {
+    stack.push(node);
+    RESULT result = visit(context, plan, block, node.getChild(), stack);
+    stack.pop();
+    return result;
+  }
+
+  @Override
+  public RESULT visitSort(CONTEXT context, LogicalPlan plan, LogicalPlan.QueryBlock block, SortNode node,
+                          Stack<LogicalNode> stack) throws PlanningException {
+    stack.push(node);
+    RESULT result = visit(context, plan, block, node.getChild(), stack);
+    stack.pop();
+    return result;
+  }
+
+  @Override
+  public RESULT visitHaving(CONTEXT context, LogicalPlan plan, LogicalPlan.QueryBlock block, HavingNode node,
+                            Stack<LogicalNode> stack) throws PlanningException {
+    stack.push(node);
+    RESULT result = visit(context, plan, block, node.getChild(), stack);
+    stack.pop();
+    return result;
+  }
+
+  @Override
+  public RESULT visitGroupBy(CONTEXT context, LogicalPlan plan, LogicalPlan.QueryBlock block, GroupbyNode node,
+                             Stack<LogicalNode> stack) throws PlanningException {
+    stack.push(node);
+    RESULT result = visit(context, plan, block, node.getChild(), stack);
+    stack.pop();
+    return result;
+  }
+
+  @Override
+  public RESULT visitFilter(CONTEXT context, LogicalPlan plan, LogicalPlan.QueryBlock block, SelectionNode node,
+                            Stack<LogicalNode> stack) throws PlanningException {
+    stack.push(node);
+    RESULT result = visit(context, plan, block, node.getChild(), stack);
+    stack.pop();
+    return result;
+  }
+
+  @Override
+  public RESULT visitJoin(CONTEXT context, LogicalPlan plan, LogicalPlan.QueryBlock block, JoinNode node,
+                          Stack<LogicalNode> stack) throws PlanningException {
+    stack.push(node);
+    RESULT result = visit(context, plan, block, node.getLeftChild(), stack);
+    visit(context, plan, block, node.getRightChild(), stack);
+    stack.pop();
+    return result;
+  }
+
+  @Override
+  public RESULT visitUnion(CONTEXT context, LogicalPlan plan, LogicalPlan.QueryBlock block, UnionNode node,
+                           Stack<LogicalNode> stack) throws PlanningException {
+    stack.push(node);
+    LogicalPlan.QueryBlock leftBlock = plan.getBlock(node.getLeftChild());
+    RESULT result = visit(context, plan, leftBlock, leftBlock.getRoot(), stack);
+    LogicalPlan.QueryBlock rightBlock = plan.getBlock(node.getRightChild());
+    visit(context, plan, rightBlock, rightBlock.getRoot(), stack);
+    stack.pop();
+    return result;
+  }
+
+  @Override
+  public RESULT visitExcept(CONTEXT context, LogicalPlan plan, LogicalPlan.QueryBlock block, ExceptNode node,
+                            Stack<LogicalNode> stack) throws PlanningException {
+    stack.push(node);
+    RESULT result = visit(context, plan, block, node.getLeftChild(), stack);
+    visit(context, plan, block, node.getRightChild(), stack);
+    stack.pop();
+    return result;
+  }
+
+  @Override
+  public RESULT visitIntersect(CONTEXT context, LogicalPlan plan, LogicalPlan.QueryBlock block, IntersectNode node,
+                               Stack<LogicalNode> stack) throws PlanningException {
+    stack.push(node);
+    RESULT result = visit(context, plan, block, node.getLeftChild(), stack);
+    visit(context, plan, block, node.getRightChild(), stack);
+    stack.pop();
+    return result;
+  }
+
+  @Override
+  public RESULT visitTableSubQuery(CONTEXT context, LogicalPlan plan, LogicalPlan.QueryBlock block,
+                                   TableSubQueryNode node, Stack<LogicalNode> stack) throws PlanningException {
+    stack.push(node);
+    LogicalPlan.QueryBlock childBlock = plan.getBlock(node.getSubQuery());
+    RESULT result = visit(context, plan, childBlock, childBlock.getRoot(), new Stack<LogicalNode>());
+    stack.pop();
+    return result;
+  }
+
+  @Override
+  public RESULT visitScan(CONTEXT context, LogicalPlan plan, LogicalPlan.QueryBlock block, ScanNode node,
+                          Stack<LogicalNode> stack) throws PlanningException {
+    return null;
+  }
+
+  @Override
+  public RESULT visitPartitionedTableScan(CONTEXT context, LogicalPlan plan, LogicalPlan.QueryBlock block,
+                                          PartitionedTableScanNode node, Stack<LogicalNode> stack)
+      throws PlanningException {
+    return null;
+  }
+
+  @Override
+  public RESULT visitStoreTable(CONTEXT context, LogicalPlan plan, LogicalPlan.QueryBlock block, StoreTableNode node,
+                                Stack<LogicalNode> stack) throws PlanningException {
+    stack.push(node);
+    RESULT result = visit(context, plan, block, node.getChild(), stack);
+    stack.pop();
+    return result;
+  }
+
+  @Override
+  public RESULT visitInsert(CONTEXT context, LogicalPlan plan, LogicalPlan.QueryBlock block, InsertNode node,
+                            Stack<LogicalNode> stack) throws PlanningException {
+    stack.push(node);
+    RESULT result = visit(context, plan, block, node.getChild(), stack);
+    stack.pop();
+    return result;
+  }
+
+  @Override
+  public RESULT visitCreateDatabase(CONTEXT context, LogicalPlan plan, LogicalPlan.QueryBlock block,
+                                    CreateDatabaseNode node, Stack<LogicalNode> stack) throws PlanningException {
+    return null;
+  }
+
+  @Override
+  public RESULT visitDropDatabase(CONTEXT context, LogicalPlan plan, LogicalPlan.QueryBlock block, DropDatabaseNode node, Stack<LogicalNode> stack) throws PlanningException {
+    return null;
+  }
+
+  @Override
+  public RESULT visitCreateTable(CONTEXT context, LogicalPlan plan, LogicalPlan.QueryBlock block, CreateTableNode node,
+                                 Stack<LogicalNode> stack) throws PlanningException {
+    RESULT result = null;
+    stack.push(node);
+    if (node.hasSubQuery()) {
+      result = visit(context, plan, block, node.getChild(), stack);
+    }
+    stack.pop();
+    return result;
+  }
+
+  @Override
+  public RESULT visitDropTable(CONTEXT context, LogicalPlan plan, LogicalPlan.QueryBlock block, DropTableNode node,
+                               Stack<LogicalNode> stack) {
+    return null;
+  }
+
+  @Override
+  public RESULT visitAlterTablespace(CONTEXT context, LogicalPlan plan, LogicalPlan.QueryBlock block,
+                                     AlterTablespaceNode node, Stack<LogicalNode> stack) throws PlanningException {
+    return null;
+  }
+
+  @Override
+  public RESULT visitAlterTable(CONTEXT context, LogicalPlan plan, LogicalPlan.QueryBlock block, AlterTableNode node,
+                                 Stack<LogicalNode> stack) {
+        return null;
+    }
+}

http://git-wip-us.apache.org/repos/asf/tajo/blob/6594ac1c/tajo-core/src/main/java/org/apache/tajo/engine/planner/BroadcastJoinPlanVisitor.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/BroadcastJoinPlanVisitor.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/BroadcastJoinPlanVisitor.java
new file mode 100644
index 0000000..14ac85f
--- /dev/null
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/BroadcastJoinPlanVisitor.java
@@ -0,0 +1,86 @@
+/**
+ * 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.engine.planner;
+
+import org.apache.tajo.engine.planner.global.GlobalPlanner;
+import org.apache.tajo.engine.planner.logical.JoinNode;
+import org.apache.tajo.engine.planner.logical.LogicalNode;
+import org.apache.tajo.engine.planner.logical.NodeType;
+
+import java.util.Stack;
+
+public class BroadcastJoinPlanVisitor extends BasicLogicalPlanVisitor<GlobalPlanner.GlobalPlanContext, LogicalNode> {
+
+  @Override
+  public LogicalNode visitJoin(GlobalPlanner.GlobalPlanContext context, LogicalPlan plan, LogicalPlan.QueryBlock block,
+                               JoinNode node, Stack<LogicalNode> stack) throws PlanningException {
+    LogicalNode leftChild = node.getLeftChild();
+    LogicalNode rightChild = node.getRightChild();
+
+    if (isScanNode(leftChild) && isScanNode(rightChild)) {
+      node.setCandidateBroadcast(true);
+      node.getBroadcastTargets().add(leftChild);
+      node.getBroadcastTargets().add(rightChild);
+      return node;
+    }
+
+    if(!isScanNode(leftChild)) {
+      visit(context, plan, block, leftChild, stack);
+    }
+
+    if(!isScanNode(rightChild)) {
+      visit(context, plan, block, rightChild, stack);
+    }
+
+    if(isBroadcastCandidateNode(leftChild) && isBroadcastCandidateNode(rightChild)) {
+      node.setCandidateBroadcast(true);
+      if(leftChild.getType() == NodeType.JOIN) {
+        node.getBroadcastTargets().addAll(((JoinNode)leftChild).getBroadcastTargets());
+      } else {
+        node.getBroadcastTargets().add(leftChild);
+      }
+
+      if(rightChild.getType() == NodeType.JOIN) {
+        node.getBroadcastTargets().addAll(((JoinNode)rightChild).getBroadcastTargets());
+      } else {
+        node.getBroadcastTargets().add(rightChild);
+      }
+    }
+
+    return node;
+  }
+
+  private static boolean isBroadcastCandidateNode(LogicalNode node) {
+    if(node.getType() == NodeType.SCAN ||
+        node.getType() == NodeType.PARTITIONS_SCAN) {
+      return true;
+    }
+
+    if(node.getType() == NodeType.JOIN && ((JoinNode)node).isCandidateBroadcast()) {
+      return true;
+    }
+
+    return false;
+  }
+
+  private static boolean isScanNode(LogicalNode node) {
+    return node.getType() == NodeType.SCAN ||
+        node.getType() == NodeType.PARTITIONS_SCAN;
+  }
+}

http://git-wip-us.apache.org/repos/asf/tajo/blob/6594ac1c/tajo-core/src/main/java/org/apache/tajo/engine/planner/ExplainLogicalPlanVisitor.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/ExplainLogicalPlanVisitor.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/ExplainLogicalPlanVisitor.java
new file mode 100644
index 0000000..9dd8700
--- /dev/null
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/ExplainLogicalPlanVisitor.java
@@ -0,0 +1,236 @@
+/**
+ * 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.engine.planner;
+
+import org.apache.tajo.annotation.Nullable;
+import org.apache.tajo.engine.planner.logical.*;
+
+import java.util.Stack;
+
+/**
+ * It returns a list of node plan strings.
+ */
+public class ExplainLogicalPlanVisitor extends BasicLogicalPlanVisitor<ExplainLogicalPlanVisitor.Context, LogicalNode> {
+
+  public static class Context {
+    public int maxDepth  = -1;
+    public int depth = 0;
+    public Stack<DepthString> explains = new Stack<DepthString>();
+
+    public void add(int depth, PlanString planString) {
+      maxDepth = Math.max(maxDepth, depth);
+      explains.push(new DepthString(depth, planString));
+    }
+
+    public int getMaxDepth() {
+      return this.maxDepth;
+    }
+
+    public Stack<DepthString> getExplains() {
+      return explains;
+    }
+  }
+
+  public static class DepthString {
+    private int depth;
+    private PlanString planStr;
+
+    DepthString(int depth, PlanString planStr) {
+      this.depth = depth;
+      this.planStr = planStr;
+    }
+
+    public int getDepth() {
+      return depth;
+    }
+
+    public PlanString getPlanString() {
+      return planStr;
+    }
+  }
+
+  public Context getBlockPlanStrings(@Nullable LogicalPlan plan, LogicalNode node) throws PlanningException {
+    Stack<LogicalNode> stack = new Stack<LogicalNode>();
+    Context explainContext = new Context();
+    visit(explainContext, plan, null, node, stack);
+    return explainContext;
+  }
+
+  @Override
+  public LogicalNode visitRoot(Context context, LogicalPlan plan, LogicalPlan.QueryBlock block, LogicalRootNode node, Stack<LogicalNode> stack)
+      throws PlanningException {
+    return visit(context, plan, block, node.getChild(), stack);
+  }
+
+  @Override
+  public LogicalNode visitProjection(Context context, LogicalPlan plan, LogicalPlan.QueryBlock block,
+                                     ProjectionNode node, Stack<LogicalNode> stack)
+      throws PlanningException {
+    return visitUnaryNode(context, plan, block, node, stack);
+  }
+
+  @Override
+  public LogicalNode visitLimit(Context context, LogicalPlan plan, LogicalPlan.QueryBlock block,
+                                LimitNode node, Stack<LogicalNode> stack) throws PlanningException {
+    return visitUnaryNode(context, plan, block, node, stack);
+  }
+
+  @Override
+  public LogicalNode visitSort(Context context, LogicalPlan plan, LogicalPlan.QueryBlock block, SortNode node,
+                               Stack<LogicalNode> stack) throws PlanningException {
+    return visitUnaryNode(context, plan, block, node, stack);
+  }
+
+  public LogicalNode visitHaving(Context context, LogicalPlan plan, LogicalPlan.QueryBlock block, HavingNode node,
+                                  Stack<LogicalNode> stack) throws PlanningException {
+    return visitUnaryNode(context, plan, block, node, stack);
+  }
+
+  @Override
+  public LogicalNode visitGroupBy(Context context, LogicalPlan plan, LogicalPlan.QueryBlock block, GroupbyNode node,
+                                  Stack<LogicalNode> stack) throws PlanningException {
+    return visitUnaryNode(context, plan, block, node, stack);
+  }
+
+  private LogicalNode visitUnaryNode(Context context, LogicalPlan plan, LogicalPlan.QueryBlock block,
+                                     UnaryNode node, Stack<LogicalNode> stack) throws PlanningException {
+    context.depth++;
+    stack.push(node);
+    visit(context, plan, block, node.getChild(), stack);
+    context.depth--;
+    context.add(context.depth, node.getPlanString());
+    return node;
+  }
+
+  private LogicalNode visitBinaryNode(Context context, LogicalPlan plan, LogicalPlan.QueryBlock block, BinaryNode node,
+                                      Stack<LogicalNode> stack)
+      throws PlanningException {
+    context.depth++;
+    stack.push(node);
+    visit(context, plan, block, node.getLeftChild(), stack);
+    visit(context, plan, block, node.getRightChild(), stack);
+    stack.pop();
+    context.depth--;
+    context.add(context.depth, node.getPlanString());
+    return node;
+  }
+
+  @Override
+  public LogicalNode visitFilter(Context context, LogicalPlan plan, LogicalPlan.QueryBlock block, SelectionNode node,
+                                 Stack<LogicalNode> stack) throws PlanningException {
+    return visitUnaryNode(context, plan, block, node, stack);
+  }
+
+  @Override
+  public LogicalNode visitJoin(Context context, LogicalPlan plan, LogicalPlan.QueryBlock block, JoinNode node,
+                               Stack<LogicalNode> stack) throws PlanningException {
+    return visitBinaryNode(context, plan, block, node, stack);
+  }
+
+  @Override
+  public LogicalNode visitUnion(Context context, LogicalPlan plan, LogicalPlan.QueryBlock block, UnionNode node,
+                                Stack<LogicalNode> stack) throws PlanningException {
+    return visitBinaryNode(context, plan, block, node, stack);
+  }
+
+  @Override
+  public LogicalNode visitExcept(Context context, LogicalPlan plan, LogicalPlan.QueryBlock block, ExceptNode node,
+                                 Stack<LogicalNode> stack) throws PlanningException {
+    return visitBinaryNode(context, plan, block, node, stack);
+  }
+
+  @Override
+  public LogicalNode visitIntersect(Context context, LogicalPlan plan, LogicalPlan.QueryBlock block, IntersectNode node,
+                                    Stack<LogicalNode> stack) throws PlanningException {
+    return visitBinaryNode(context, plan, block, node, stack);
+  }
+
+  @Override
+  public LogicalNode visitTableSubQuery(Context context, LogicalPlan plan, LogicalPlan.QueryBlock block,
+                                        TableSubQueryNode node, Stack<LogicalNode> stack) throws PlanningException {
+    context.depth++;
+    stack.push(node);
+    visit(context, plan, block, node.getSubQuery(), new Stack<LogicalNode>());
+    stack.pop();
+    context.depth--;
+    context.add(context.depth, node.getPlanString());
+
+    return node;
+  }
+
+  @Override
+  public LogicalNode visitScan(Context context, LogicalPlan plan, LogicalPlan.QueryBlock block, ScanNode node,
+                               Stack<LogicalNode> stack) throws PlanningException {
+    context.add(context.depth, node.getPlanString());
+    return node;
+  }
+
+  @Override
+  public LogicalNode visitPartitionedTableScan(Context context, LogicalPlan plan, LogicalPlan.QueryBlock block,
+                                          PartitionedTableScanNode node, Stack<LogicalNode> stack)
+      throws PlanningException {
+    context.add(context.depth, node.getPlanString());
+    return node;
+  }
+
+  @Override
+  public LogicalNode visitStoreTable(Context context, LogicalPlan plan, LogicalPlan.QueryBlock block,
+                                     StoreTableNode node, Stack<LogicalNode> stack) throws PlanningException {
+    return visitUnaryNode(context, plan, block, node, stack);
+  }
+
+  public LogicalNode visitCreateDatabase(Context context, LogicalPlan plan, LogicalPlan.QueryBlock block,
+                                         CreateDatabaseNode node, Stack<LogicalNode> stack) throws PlanningException {
+    context.add(context.depth, node.getPlanString());
+    return node;
+  }
+
+  public LogicalNode visitDropDatabase(Context context, LogicalPlan plan, LogicalPlan.QueryBlock block,
+                                         DropDatabaseNode node, Stack<LogicalNode> stack) throws PlanningException {
+    context.add(context.depth, node.getPlanString());
+    return node;
+  }
+
+  @Override
+  public LogicalNode visitInsert(Context context, LogicalPlan plan, LogicalPlan.QueryBlock block, InsertNode node,
+                                 Stack<LogicalNode> stack) throws PlanningException {
+    context.depth++;
+    stack.push(node);
+    super.visitInsert(context, plan, block, node, stack);
+    stack.pop();
+    context.depth--;
+    context.add(context.depth, node.getPlanString());
+    return node;
+  }
+
+  public static String printDepthString(int maxDepth, DepthString planStr) {
+    StringBuilder output = new StringBuilder();
+    String pad = new String(new char[planStr.getDepth() * 3]).replace('\0', ' ');
+    output.append(pad + planStr.getPlanString().getTitle()).append("\n");
+
+    for (String str : planStr.getPlanString().getExplanations()) {
+      output.append(pad).append("  => ").append(str).append("\n");
+    }
+
+    for (String str : planStr.getPlanString().getDetails()) {
+      output.append(pad).append("  => ").append(str).append("\n");
+    }
+    return output.toString();
+  }
+}

http://git-wip-us.apache.org/repos/asf/tajo/blob/6594ac1c/tajo-core/src/main/java/org/apache/tajo/engine/planner/ExprAnnotator.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/ExprAnnotator.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/ExprAnnotator.java
new file mode 100644
index 0000000..1b57b98
--- /dev/null
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/ExprAnnotator.java
@@ -0,0 +1,631 @@
+/**
+ * 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.engine.planner;
+
+import org.apache.tajo.algebra.*;
+import org.apache.tajo.catalog.CatalogService;
+import org.apache.tajo.catalog.CatalogUtil;
+import org.apache.tajo.catalog.Column;
+import org.apache.tajo.catalog.FunctionDesc;
+import org.apache.tajo.catalog.exception.NoSuchFunctionException;
+import org.apache.tajo.catalog.proto.CatalogProtos;
+import org.apache.tajo.common.TajoDataTypes;
+import org.apache.tajo.datum.*;
+import org.apache.tajo.engine.eval.*;
+import org.apache.tajo.engine.function.AggFunction;
+import org.apache.tajo.engine.function.GeneralFunction;
+import org.apache.tajo.engine.planner.logical.NodeType;
+import org.apache.tajo.exception.InternalException;
+import org.joda.time.DateTime;
+
+import java.util.Stack;
+
+/**
+ * <code>ExprAnnotator</code> makes an annotated expression called <code>EvalNode</code> from an
+ * {@link org.apache.tajo.algebra.Expr}. It visits descendants recursively from a given expression, and finally
+ * it returns an EvalNode.
+ */
+public class ExprAnnotator extends BaseAlgebraVisitor<ExprAnnotator.Context, EvalNode> {
+  private CatalogService catalog;
+
+  public ExprAnnotator(CatalogService catalog) {
+    this.catalog = catalog;
+  }
+
+  static class Context {
+    LogicalPlan plan;
+    LogicalPlan.QueryBlock currentBlock;
+
+    public Context(LogicalPlan plan, LogicalPlan.QueryBlock block) {
+      this.plan = plan;
+      this.currentBlock = block;
+    }
+  }
+
+  public EvalNode createEvalNode(LogicalPlan plan, LogicalPlan.QueryBlock block, Expr expr)
+      throws PlanningException {
+    Context context = new Context(plan, block);
+    return visit(context, new Stack<Expr>(), expr);
+  }
+
+  ///////////////////////////////////////////////////////////////////////////////////////////////////////////
+  // Logical Operator Section
+  ///////////////////////////////////////////////////////////////////////////////////////////////////////////
+
+  @Override
+  public EvalNode visitAnd(Context ctx, Stack<Expr> stack, BinaryOperator expr) throws PlanningException {
+    stack.push(expr);
+    EvalNode left = visit(ctx, stack, expr.getLeft());
+    EvalNode right = visit(ctx, stack, expr.getRight());
+    stack.pop();
+
+    return new BinaryEval(EvalType.AND, left, right);
+  }
+
+  @Override
+  public EvalNode visitOr(Context ctx, Stack<Expr> stack, BinaryOperator expr) throws PlanningException {
+    stack.push(expr);
+    EvalNode left = visit(ctx, stack, expr.getLeft());
+    EvalNode right = visit(ctx, stack, expr.getRight());
+    stack.pop();
+
+    return new BinaryEval(EvalType.OR, left, right);
+  }
+
+  @Override
+  public EvalNode visitNot(Context ctx, Stack<Expr> stack, NotExpr expr) throws PlanningException {
+    stack.push(expr);
+    EvalNode child = visit(ctx, stack, expr.getChild());
+    stack.pop();
+    return new NotEval(child);
+  }
+
+  ///////////////////////////////////////////////////////////////////////////////////////////////////////////
+  // Comparison Predicates Section
+  ///////////////////////////////////////////////////////////////////////////////////////////////////////////
+  @Override
+  public EvalNode visitEquals(Context ctx, Stack<Expr> stack, BinaryOperator expr) throws PlanningException {
+    return visitCommonComparison(ctx, stack, expr);
+  }
+
+  @Override
+  public EvalNode visitNotEquals(Context ctx, Stack<Expr> stack, BinaryOperator expr) throws PlanningException {
+    return visitCommonComparison(ctx, stack, expr);
+  }
+
+  @Override
+  public EvalNode visitLessThan(Context ctx, Stack<Expr> stack, BinaryOperator expr) throws PlanningException {
+    return visitCommonComparison(ctx, stack, expr);
+  }
+
+  @Override
+  public EvalNode visitLessThanOrEquals(Context ctx, Stack<Expr> stack, BinaryOperator expr) throws PlanningException {
+    return visitCommonComparison(ctx, stack, expr);
+  }
+
+  @Override
+  public EvalNode visitGreaterThan(Context ctx, Stack<Expr> stack, BinaryOperator expr) throws PlanningException {
+    return visitCommonComparison(ctx, stack, expr);
+  }
+
+  @Override
+  public EvalNode visitGreaterThanOrEquals(Context ctx, Stack<Expr> stack, BinaryOperator expr)
+      throws PlanningException {
+    return visitCommonComparison(ctx, stack, expr);
+  }
+
+  public EvalNode visitCommonComparison(Context ctx, Stack<Expr> stack, BinaryOperator expr) throws PlanningException {
+    stack.push(expr);
+    EvalNode left = visit(ctx, stack, expr.getLeft());
+    EvalNode right = visit(ctx, stack, expr.getRight());
+    stack.pop();
+
+    EvalType evalType;
+    switch (expr.getType()) {
+      case Equals:
+        evalType = EvalType.EQUAL;
+        break;
+      case NotEquals:
+        evalType = EvalType.NOT_EQUAL;
+        break;
+      case LessThan:
+        evalType = EvalType.LTH;
+        break;
+      case LessThanOrEquals:
+        evalType = EvalType.LEQ;
+        break;
+      case GreaterThan:
+        evalType = EvalType.GTH;
+        break;
+      case GreaterThanOrEquals:
+        evalType = EvalType.GEQ;
+        break;
+      default:
+      throw new IllegalStateException("Wrong Expr Type: " + expr.getType());
+    }
+
+    return new BinaryEval(evalType, left, right);
+  }
+
+  ///////////////////////////////////////////////////////////////////////////////////////////////////////////
+  // Other Predicates Section
+  ///////////////////////////////////////////////////////////////////////////////////////////////////////////
+
+  @Override
+  public EvalNode visitBetween(Context ctx, Stack<Expr> stack, BetweenPredicate between) throws PlanningException {
+    stack.push(between);
+    EvalNode predicand = visit(ctx, stack, between.predicand());
+    EvalNode begin = visit(ctx, stack, between.begin());
+    EvalNode end = visit(ctx, stack, between.end());
+    stack.pop();
+
+    BetweenPredicateEval betweenEval = new BetweenPredicateEval(
+        between.isNot(),
+        between.isSymmetric(),
+        predicand, begin, end);
+    return betweenEval;
+  }
+
+  @Override
+  public EvalNode visitCaseWhen(Context ctx, Stack<Expr> stack, CaseWhenPredicate caseWhen) throws PlanningException {
+    CaseWhenEval caseWhenEval = new CaseWhenEval();
+
+    EvalNode condition;
+    EvalNode result;
+    for (CaseWhenPredicate.WhenExpr when : caseWhen.getWhens()) {
+      condition = visit(ctx, stack, when.getCondition());
+      result = visit(ctx, stack, when.getResult());
+      caseWhenEval.addWhen(condition, result);
+    }
+
+    if (caseWhen.hasElseResult()) {
+      caseWhenEval.setElseResult(visit(ctx, stack, caseWhen.getElseResult()));
+    }
+
+    return caseWhenEval;
+  }
+
+  @Override
+  public EvalNode visitIsNullPredicate(Context ctx, Stack<Expr> stack, IsNullPredicate expr) throws PlanningException {
+    stack.push(expr);
+    EvalNode child = visit(ctx, stack, expr.getPredicand());
+    stack.pop();
+    return new IsNullEval(expr.isNot(), child);
+  }
+
+  @Override
+  public EvalNode visitInPredicate(Context ctx, Stack<Expr> stack, InPredicate expr) throws PlanningException {
+    stack.push(expr);
+    EvalNode lhs = visit(ctx, stack, expr.getLeft());
+    RowConstantEval rowConstantEval = (RowConstantEval) visit(ctx, stack, expr.getInValue());
+    stack.pop();
+    return new InEval(lhs, rowConstantEval, expr.isNot());
+  }
+
+  @Override
+  public EvalNode visitValueListExpr(Context ctx, Stack<Expr> stack, ValueListExpr expr) throws PlanningException {
+    Datum[] values = new Datum[expr.getValues().length];
+    EvalNode [] evalNodes = new EvalNode[expr.getValues().length];
+    for (int i = 0; i < expr.getValues().length; i++) {
+      evalNodes[i] = visit(ctx, stack, expr.getValues()[i]);
+      if (!EvalTreeUtil.checkIfCanBeConstant(evalNodes[i])) {
+        throw new PlanningException("Non constant values cannot be included in IN PREDICATE.");
+      }
+      values[i] = EvalTreeUtil.evaluateImmediately(evalNodes[i]);
+    }
+    return new RowConstantEval(values);
+  }
+
+  @Override
+  public EvalNode visitExistsPredicate(Context ctx, Stack<Expr> stack, ExistsPredicate expr) throws PlanningException {
+    throw new PlanningException("Cannot support EXISTS clause yet");
+  }
+
+  ///////////////////////////////////////////////////////////////////////////////////////////////////////////
+  // String Operator or Pattern Matching Predicates Section
+  ///////////////////////////////////////////////////////////////////////////////////////////////////////////
+  @Override
+  public EvalNode visitLikePredicate(Context ctx, Stack<Expr> stack, PatternMatchPredicate expr)
+      throws PlanningException {
+    return visitPatternMatchPredicate(ctx, stack, expr);
+  }
+
+  @Override
+  public EvalNode visitSimilarToPredicate(Context ctx, Stack<Expr> stack, PatternMatchPredicate expr)
+      throws PlanningException {
+    return visitPatternMatchPredicate(ctx, stack, expr);
+  }
+
+  @Override
+  public EvalNode visitRegexpPredicate(Context ctx, Stack<Expr> stack, PatternMatchPredicate expr)
+      throws PlanningException {
+    return visitPatternMatchPredicate(ctx, stack, expr);
+  }
+
+  @Override
+  public EvalNode visitConcatenate(Context ctx, Stack<Expr> stack, BinaryOperator expr) throws PlanningException {
+    stack.push(expr);
+    EvalNode left = visit(ctx, stack, expr.getLeft());
+    EvalNode right = visit(ctx, stack, expr.getRight());
+    stack.pop();
+
+    return new BinaryEval(EvalType.CONCATENATE, left, right);
+  }
+
+  private EvalNode visitPatternMatchPredicate(Context ctx, Stack<Expr> stack, PatternMatchPredicate expr)
+      throws PlanningException {
+    EvalNode field = visit(ctx, stack, expr.getPredicand());
+    ConstEval pattern = (ConstEval) visit(ctx, stack, expr.getPattern());
+
+    // A pattern is a const value in pattern matching predicates.
+    // In a binary expression, the result is always null if a const value in left or right side is null.
+    if (pattern.getValue() instanceof NullDatum) {
+      return new ConstEval(NullDatum.get());
+    } else {
+      if (expr.getType() == OpType.LikePredicate) {
+        return new LikePredicateEval(expr.isNot(), field, pattern, expr.isCaseInsensitive());
+      } else if (expr.getType() == OpType.SimilarToPredicate) {
+        return new SimilarToPredicateEval(expr.isNot(), field, pattern);
+      } else {
+        return new RegexPredicateEval(expr.isNot(), field, pattern, expr.isCaseInsensitive());
+      }
+    }
+  }
+
+  ///////////////////////////////////////////////////////////////////////////////////////////////////////////
+  // Arithmetic Operators
+  ///////////////////////////////////////////////////////////////////////////////////////////////////////////
+
+  @Override
+  public EvalNode visitPlus(Context ctx, Stack<Expr> stack, BinaryOperator expr) throws PlanningException {
+    stack.push(expr);
+    EvalNode left = visit(ctx, stack, expr.getLeft());
+    EvalNode right = visit(ctx, stack, expr.getRight());
+    stack.pop();
+
+    return new BinaryEval(EvalType.PLUS, left, right);
+  }
+
+  @Override
+  public EvalNode visitMinus(Context ctx, Stack<Expr> stack, BinaryOperator expr) throws PlanningException {
+    stack.push(expr);
+    EvalNode left = visit(ctx, stack, expr.getLeft());
+    EvalNode right = visit(ctx, stack, expr.getRight());
+    stack.pop();
+
+    return new BinaryEval(EvalType.MINUS, left, right);
+  }
+
+  @Override
+  public EvalNode visitMultiply(Context ctx, Stack<Expr> stack, BinaryOperator expr) throws PlanningException {
+    stack.push(expr);
+    EvalNode left = visit(ctx, stack, expr.getLeft());
+    EvalNode right = visit(ctx, stack, expr.getRight());
+    stack.pop();
+
+    return new BinaryEval(EvalType.MULTIPLY, left, right);
+  }
+
+  @Override
+  public EvalNode visitDivide(Context ctx, Stack<Expr> stack, BinaryOperator expr) throws PlanningException {
+    stack.push(expr);
+    EvalNode left = visit(ctx, stack, expr.getLeft());
+    EvalNode right = visit(ctx, stack, expr.getRight());
+    stack.pop();
+
+    return new BinaryEval(EvalType.DIVIDE, left, right);
+  }
+
+  @Override
+  public EvalNode visitModular(Context ctx, Stack<Expr> stack, BinaryOperator expr) throws PlanningException {
+    stack.push(expr);
+    EvalNode left = visit(ctx, stack, expr.getLeft());
+    EvalNode right = visit(ctx, stack, expr.getRight());
+    stack.pop();
+
+    return new BinaryEval(EvalType.MODULAR, left, right);
+  }
+
+  ///////////////////////////////////////////////////////////////////////////////////////////////////////////
+  // Other Expressions
+  ///////////////////////////////////////////////////////////////////////////////////////////////////////////
+
+  @Override
+  public EvalNode visitSign(Context ctx, Stack<Expr> stack, SignedExpr expr) throws PlanningException {
+    stack.push(expr);
+    EvalNode numericExpr = visit(ctx, stack, expr.getChild());
+    stack.pop();
+
+    if (expr.isNegative()) {
+      return new SignedEval(expr.isNegative(), numericExpr);
+    } else {
+      return numericExpr;
+    }
+  }
+
+  @Override
+  public EvalNode visitColumnReference(Context ctx, Stack<Expr> stack, ColumnReferenceExpr expr)
+      throws PlanningException {
+    Column column = ctx.plan.resolveColumn(ctx.currentBlock, expr);
+    return new FieldEval(column);
+  }
+
+  @Override
+  public EvalNode visitTargetExpr(Context ctx, Stack<Expr> stack, NamedExpr expr) throws PlanningException {
+    throw new PlanningException("ExprAnnotator cannot take NamedExpr");
+  }
+
+  @Override
+  public EvalNode visitFunction(Context ctx, Stack<Expr> stack, FunctionExpr expr) throws PlanningException {
+    stack.push(expr); // <--- Push
+
+    // Given parameters
+    Expr[] params = expr.getParams();
+    if (params == null) {
+      params = new Expr[0];
+    }
+
+    EvalNode[] givenArgs = new EvalNode[params.length];
+    TajoDataTypes.DataType[] paramTypes = new TajoDataTypes.DataType[params.length];
+
+    for (int i = 0; i < params.length; i++) {
+      givenArgs[i] = visit(ctx, stack, params[i]);
+      paramTypes[i] = givenArgs[i].getValueType();
+    }
+
+    stack.pop(); // <--- Pop
+
+    if (!catalog.containFunction(expr.getSignature(), paramTypes)) {
+      throw new NoSuchFunctionException(expr.getSignature(), paramTypes);
+    }
+
+    FunctionDesc funcDesc = catalog.getFunction(expr.getSignature(), paramTypes);
+
+    try {
+    CatalogProtos.FunctionType functionType = funcDesc.getFuncType();
+    if (functionType == CatalogProtos.FunctionType.GENERAL
+        || functionType == CatalogProtos.FunctionType.UDF) {
+      return new GeneralFunctionEval(funcDesc, (GeneralFunction) funcDesc.newInstance(), givenArgs);
+    } else if (functionType == CatalogProtos.FunctionType.AGGREGATION
+        || functionType == CatalogProtos.FunctionType.UDA) {
+      if (!ctx.currentBlock.hasNode(NodeType.GROUP_BY)) {
+        ctx.currentBlock.setAggregationRequire();
+      }
+      return new AggregationFunctionCallEval(funcDesc, (AggFunction) funcDesc.newInstance(), givenArgs);
+    } else if (functionType == CatalogProtos.FunctionType.DISTINCT_AGGREGATION
+        || functionType == CatalogProtos.FunctionType.DISTINCT_UDA) {
+      throw new PlanningException("Unsupported function: " + funcDesc.toString());
+    } else {
+      throw new PlanningException("Unsupported Function Type: " + functionType.name());
+    }
+    } catch (InternalException e) {
+      throw new PlanningException(e);
+    }
+  }
+
+  ///////////////////////////////////////////////////////////////////////////////////////////////////////////
+  // General Set Section
+  ///////////////////////////////////////////////////////////////////////////////////////////////////////////
+
+  @Override
+  public EvalNode visitCountRowsFunction(Context ctx, Stack<Expr> stack, CountRowsFunctionExpr expr)
+      throws PlanningException {
+    FunctionDesc countRows = catalog.getFunction("count", CatalogProtos.FunctionType.AGGREGATION,
+        new TajoDataTypes.DataType[] {});
+    if (countRows == null) {
+      throw new NoSuchFunctionException(countRows.getSignature(), new TajoDataTypes.DataType[]{});
+    }
+
+    try {
+      ctx.currentBlock.setAggregationRequire();
+
+      return new AggregationFunctionCallEval(countRows, (AggFunction) countRows.newInstance(),
+          new EvalNode[] {});
+    } catch (InternalException e) {
+      throw new NoSuchFunctionException(countRows.getSignature(), new TajoDataTypes.DataType[]{});
+    }
+  }
+
+  @Override
+  public EvalNode visitGeneralSetFunction(Context ctx, Stack<Expr> stack, GeneralSetFunctionExpr setFunction)
+      throws PlanningException {
+
+    Expr[] params = setFunction.getParams();
+    EvalNode[] givenArgs = new EvalNode[params.length];
+    TajoDataTypes.DataType[] paramTypes = new TajoDataTypes.DataType[params.length];
+
+    CatalogProtos.FunctionType functionType = setFunction.isDistinct() ?
+        CatalogProtos.FunctionType.DISTINCT_AGGREGATION : CatalogProtos.FunctionType.AGGREGATION;
+    givenArgs[0] = visit(ctx, stack, params[0]);
+    if (setFunction.getSignature().equalsIgnoreCase("count")) {
+      paramTypes[0] = CatalogUtil.newSimpleDataType(TajoDataTypes.Type.ANY);
+    } else {
+      paramTypes[0] = givenArgs[0].getValueType();
+    }
+
+    if (!catalog.containFunction(setFunction.getSignature(), functionType, paramTypes)) {
+      throw new NoSuchFunctionException(setFunction.getSignature(), paramTypes);
+    }
+
+    FunctionDesc funcDesc = catalog.getFunction(setFunction.getSignature(), functionType, paramTypes);
+    if (!ctx.currentBlock.hasNode(NodeType.GROUP_BY)) {
+      ctx.currentBlock.setAggregationRequire();
+    }
+
+    try {
+      return new AggregationFunctionCallEval(funcDesc, (AggFunction) funcDesc.newInstance(), givenArgs);
+    } catch (InternalException e) {
+      throw new PlanningException(e);
+    }
+  }
+
+  ///////////////////////////////////////////////////////////////////////////////////////////////////////////
+  // Literal Section
+  ///////////////////////////////////////////////////////////////////////////////////////////////////////////
+
+  @Override
+  public EvalNode visitDataType(Context ctx, Stack<Expr> stack, DataTypeExpr expr) throws PlanningException {
+    return super.visitDataType(ctx, stack, expr);
+  }
+
+  @Override
+  public EvalNode visitCastExpr(Context ctx, Stack<Expr> stack, CastExpr expr) throws PlanningException {
+    EvalNode child = super.visitCastExpr(ctx, stack, expr);
+
+    if (child.getType() == EvalType.CONST) { // if it is a casting operation for a constant value
+      ConstEval constEval = (ConstEval) child; // it will be pre-computed and casted to a constant value
+      return new ConstEval(DatumFactory.cast(constEval.getValue(), LogicalPlanner.convertDataType(expr.getTarget())));
+    } else {
+      return new CastEval(child, LogicalPlanner.convertDataType(expr.getTarget()));
+    }
+  }
+
+  @Override
+  public EvalNode visitLiteral(Context ctx, Stack<Expr> stack, LiteralValue expr) throws PlanningException {
+    switch (expr.getValueType()) {
+    case Boolean:
+      return new ConstEval(DatumFactory.createBool(((BooleanLiteral) expr).isTrue()));
+    case String:
+      return new ConstEval(DatumFactory.createText(expr.getValue()));
+    case Unsigned_Integer:
+      return new ConstEval(DatumFactory.createInt4(expr.getValue()));
+    case Unsigned_Large_Integer:
+      return new ConstEval(DatumFactory.createInt8(expr.getValue()));
+    case Unsigned_Float:
+      return new ConstEval(DatumFactory.createFloat8(expr.getValue()));
+    default:
+      throw new RuntimeException("Unsupported type: " + expr.getValueType());
+    }
+  }
+
+  @Override
+  public EvalNode visitNullLiteral(Context ctx, Stack<Expr> stack, NullLiteral expr) throws PlanningException {
+    return new ConstEval(NullDatum.get());
+  }
+
+  @Override
+  public EvalNode visitDateLiteral(Context context, Stack<Expr> stack, DateLiteral expr) throws PlanningException {
+    DateValue dateValue = expr.getDate();
+    int [] dates = dateToIntArray(dateValue.getYears(), dateValue.getMonths(), dateValue.getDays());
+    return new ConstEval(new DateDatum(dates[0], dates[1], dates[2]));
+  }
+
+  @Override
+  public EvalNode visitTimestampLiteral(Context ctx, Stack<Expr> stack, TimestampLiteral expr)
+      throws PlanningException {
+    DateValue dateValue = expr.getDate();
+    TimeValue timeValue = expr.getTime();
+
+    int [] dates = dateToIntArray(dateValue.getYears(),
+        dateValue.getMonths(),
+        dateValue.getDays());
+    int [] times = timeToIntArray(timeValue.getHours(),
+        timeValue.getMinutes(),
+        timeValue.getSeconds(),
+        timeValue.getSecondsFraction());
+    DateTime dateTime;
+    if (timeValue.hasSecondsFraction()) {
+      dateTime = new DateTime(dates[0], dates[1], dates[2], times[0], times[1], times[2], times[3]);
+    } else {
+      dateTime = new DateTime(dates[0], dates[1], dates[2], times[0], times[1], times[2]);
+    }
+
+    return new ConstEval(new TimestampDatum(dateTime));
+  }
+
+  @Override
+  public EvalNode visitTimeLiteral(Context ctx, Stack<Expr> stack, TimeLiteral expr) throws PlanningException {
+    TimeValue timeValue = expr.getTime();
+    int [] times = timeToIntArray(timeValue.getHours(),
+        timeValue.getMinutes(),
+        timeValue.getSeconds(),
+        timeValue.getSecondsFraction());
+
+    TimeDatum datum;
+    if (timeValue.hasSecondsFraction()) {
+      datum = new TimeDatum(times[0], times[1], times[2], times[3]);
+    } else {
+      datum = new TimeDatum(times[0], times[1], times[2]);
+    }
+    return new ConstEval(datum);
+  }
+
+  public static int [] dateToIntArray(String years, String months, String days)
+      throws PlanningException {
+    int year = Integer.valueOf(years);
+    int month = Integer.valueOf(months);
+    int day = Integer.valueOf(days);
+
+    if (!(1 <= year && year <= 9999)) {
+      throw new PlanningException(String.format("Years (%d) must be between 1 and 9999 integer value", year));
+    }
+
+    if (!(1 <= month && month <= 12)) {
+      throw new PlanningException(String.format("Months (%d) must be between 1 and 12 integer value", month));
+    }
+
+    if (!(1<= day && day <= 31)) {
+      throw new PlanningException(String.format("Days (%d) must be between 1 and 31 integer value", day));
+    }
+
+    int [] results = new int[3];
+    results[0] = year;
+    results[1] = month;
+    results[2] = day;
+
+    return results;
+  }
+
+  public static int [] timeToIntArray(String hours, String minutes, String seconds, String fractionOfSecond)
+      throws PlanningException {
+    int hour = Integer.valueOf(hours);
+    int minute = Integer.valueOf(minutes);
+    int second = Integer.valueOf(seconds);
+    int fraction = 0;
+    if (fractionOfSecond != null) {
+      fraction = Integer.valueOf(fractionOfSecond);
+    }
+
+    if (!(0 <= hour && hour <= 23)) {
+      throw new PlanningException(String.format("Hours (%d) must be between 0 and 24 integer value", hour));
+    }
+
+    if (!(0 <= minute && minute <= 59)) {
+      throw new PlanningException(String.format("Minutes (%d) must be between 0 and 59 integer value", minute));
+    }
+
+    if (!(0 <= second && second <= 59)) {
+      throw new PlanningException(String.format("Seconds (%d) must be between 0 and 59 integer value", second));
+    }
+
+    if (fraction != 0) {
+      if (!(0 <= fraction && fraction <= 999)) {
+        throw new PlanningException(String.format("Seconds (%d) must be between 0 and 999 integer value", fraction));
+      }
+    }
+
+    int [] results = new int[4];
+    results[0] = hour;
+    results[1] = minute;
+    results[2] = second;
+    results[3] = fraction;
+
+    return results;
+  }
+}

http://git-wip-us.apache.org/repos/asf/tajo/blob/6594ac1c/tajo-core/src/main/java/org/apache/tajo/engine/planner/ExprFinder.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/ExprFinder.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/ExprFinder.java
new file mode 100644
index 0000000..89eed91
--- /dev/null
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/ExprFinder.java
@@ -0,0 +1,74 @@
+/**
+ * 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.engine.planner;
+
+import org.apache.tajo.algebra.BinaryOperator;
+import org.apache.tajo.algebra.Expr;
+import org.apache.tajo.algebra.OpType;
+import org.apache.tajo.algebra.UnaryOperator;
+
+import java.util.HashSet;
+import java.util.Set;
+import java.util.Stack;
+
+class ExprFinder extends SimpleAlgebraVisitor<ExprFinder.Context, Object> {
+
+  static class Context {
+    Set<Expr> set = new HashSet<Expr>();
+    OpType targetType;
+
+    Context(OpType type) {
+      this.targetType = type;
+    }
+  }
+
+  public static <T extends Expr> Set<T> finds(Expr expr, OpType type) {
+    Context context = new Context(type);
+    ExprFinder finder = new ExprFinder();
+    Stack<Expr> stack = new Stack<Expr>();
+    stack.push(expr);
+    try {
+      finder.visit(context, new Stack<Expr>(), expr);
+    } catch (PlanningException e) {
+      throw new RuntimeException(e);
+    }
+    stack.pop();
+    return (Set<T>) context.set;
+  }
+
+  public Object visit(Context ctx, Stack<Expr> stack, Expr expr) throws PlanningException {
+    if (expr instanceof UnaryOperator) {
+      preHook(ctx, stack, expr);
+      visitUnaryOperator(ctx, stack, (UnaryOperator) expr);
+      postHook(ctx, stack, expr, null);
+    } else if (expr instanceof BinaryOperator) {
+      preHook(ctx, stack, expr);
+      visitBinaryOperator(ctx, stack, (BinaryOperator) expr);
+      postHook(ctx, stack, expr, null);
+    } else {
+      super.visit(ctx, stack, expr);
+    }
+
+    if (ctx.targetType == expr.getType()) {
+      ctx.set.add(expr);
+    }
+
+    return null;
+  }
+}