diff --git a/tajo-common/src/main/java/org/apache/tajo/datum/DatumFactory.java b/tajo-common/src/main/java/org/apache/tajo/datum/DatumFactory.java
index 00a8b80065..8943642b9e 100644
--- a/tajo-common/src/main/java/org/apache/tajo/datum/DatumFactory.java
+++ b/tajo-common/src/main/java/org/apache/tajo/datum/DatumFactory.java
@@ -384,8 +384,8 @@ public static AnyDatum createAny(Datum val) {
     return new AnyDatum(val);
   }
 
-  public static Datum cast(Datum operandDatum, DataType target, @Nullable TimeZone tz) {
-    switch (target.getType()) {
+  public static Datum cast(Datum operandDatum, org.apache.tajo.type.Type target, @Nullable TimeZone tz) {
+    switch (target.kind()) {
     case BOOLEAN:
       return DatumFactory.createBool(operandDatum.asBool());
     case CHAR:
@@ -426,7 +426,7 @@ public static Datum cast(Datum operandDatum, DataType target, @Nullable TimeZone
     case ANY:
       return DatumFactory.createAny(operandDatum);
     default:
-      throw new TajoRuntimeException(new InvalidValueForCastException(operandDatum.type(), target.getType()));
+      throw new TajoRuntimeException(new InvalidValueForCastException(operandDatum.type(), target.kind()));
     }
   }
 }
diff --git a/tajo-core-tests/src/test/resources/results/TestCommonConditionReduce/test51.plan b/tajo-core-tests/src/test/resources/results/TestCommonConditionReduce/test51.plan
index 9dd7b846ae..5f1d6c0f0d 100644
--- a/tajo-core-tests/src/test/resources/results/TestCommonConditionReduce/test51.plan
+++ b/tajo-core-tests/src/test/resources/results/TestCommonConditionReduce/test51.plan
@@ -1,7 +1,7 @@
 explain
 -------------------------------
 SCAN(0) on default.lineitem
-  => filter: (default.lineitem.l_orderkey (INT4) = 1 AND (default.lineitem.l_linenumber (INT4) = 1 AND (default.lineitem.l_suppkey (INT4) = 7706 AND (default.lineitem.l_commitdate (TEXT) IS NOT NULL AND default.lineitem.l_comment (TEXT) IS NOT NULL))))
+  => filter: (default.lineitem.l_linenumber (INT4) = 1 AND (default.lineitem.l_orderkey (INT4) = 1 AND (default.lineitem.l_suppkey (INT4) = 7706 AND (default.lineitem.l_comment (TEXT) IS NOT NULL AND default.lineitem.l_commitdate (TEXT) IS NOT NULL))))
   => target list: default.lineitem.l_comment (TEXT), default.lineitem.l_commitdate (TEXT), default.lineitem.l_discount (FLOAT8), default.lineitem.l_extendedprice (FLOAT8), default.lineitem.l_linenumber (INT4), default.lineitem.l_linestatus (TEXT), default.lineitem.l_orderkey (INT4), default.lineitem.l_partkey (INT4), default.lineitem.l_quantity (FLOAT8), default.lineitem.l_receiptdate (TEXT), default.lineitem.l_returnflag (TEXT), default.lineitem.l_shipdate (TEXT), default.lineitem.l_shipinstruct (TEXT), default.lineitem.l_shipmode (TEXT), default.lineitem.l_suppkey (INT4), default.lineitem.l_tax (FLOAT8)
   => out schema: {(16) default.lineitem.l_comment (TEXT), default.lineitem.l_commitdate (TEXT), default.lineitem.l_discount (FLOAT8), default.lineitem.l_extendedprice (FLOAT8), default.lineitem.l_linenumber (INT4), default.lineitem.l_linestatus (TEXT), default.lineitem.l_orderkey (INT4), default.lineitem.l_partkey (INT4), default.lineitem.l_quantity (FLOAT8), default.lineitem.l_receiptdate (TEXT), default.lineitem.l_returnflag (TEXT), default.lineitem.l_shipdate (TEXT), default.lineitem.l_shipinstruct (TEXT), default.lineitem.l_shipmode (TEXT), default.lineitem.l_suppkey (INT4), default.lineitem.l_tax (FLOAT8)}
   => in schema: {(16) default.lineitem.l_comment (TEXT), default.lineitem.l_commitdate (TEXT), default.lineitem.l_discount (FLOAT8), default.lineitem.l_extendedprice (FLOAT8), default.lineitem.l_linenumber (INT4), default.lineitem.l_linestatus (TEXT), default.lineitem.l_orderkey (INT4), default.lineitem.l_partkey (INT4), default.lineitem.l_quantity (FLOAT8), default.lineitem.l_receiptdate (TEXT), default.lineitem.l_returnflag (TEXT), default.lineitem.l_shipdate (TEXT), default.lineitem.l_shipinstruct (TEXT), default.lineitem.l_shipmode (TEXT), default.lineitem.l_suppkey (INT4), default.lineitem.l_tax (FLOAT8)}
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/ExprAnnotator.java b/tajo-plan/src/main/java/org/apache/tajo/plan/ExprAnnotator.java
index 92ba368cc6..c9d16d4bf5 100644
--- a/tajo-plan/src/main/java/org/apache/tajo/plan/ExprAnnotator.java
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/ExprAnnotator.java
@@ -22,10 +22,7 @@
 import org.apache.commons.collections.set.UnmodifiableSet;
 import org.apache.tajo.OverridableConf;
 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.*;
 import org.apache.tajo.catalog.CatalogUtil.Direction;
 import org.apache.tajo.common.TajoDataTypes;
 import org.apache.tajo.datum.*;
@@ -36,6 +33,7 @@
 import org.apache.tajo.plan.logical.TableSubQueryNode;
 import org.apache.tajo.plan.nameresolver.NameResolver;
 import org.apache.tajo.plan.nameresolver.NameResolvingMode;
+import org.apache.tajo.type.TypeFactory;
 import org.apache.tajo.util.Pair;
 import org.apache.tajo.util.TUtil;
 import org.apache.tajo.util.datetime.DateTimeUtil;
@@ -51,10 +49,11 @@
 import static org.apache.tajo.catalog.TypeConverter.convert;
 import static org.apache.tajo.catalog.proto.CatalogProtos.FunctionType;
 import static org.apache.tajo.common.TajoDataTypes.DataType;
-import static org.apache.tajo.common.TajoDataTypes.Type;
+import static org.apache.tajo.common.TajoDataTypes.Type.NULL_TYPE;
 import static org.apache.tajo.function.FunctionUtil.buildSimpleFunctionSignature;
 import static org.apache.tajo.plan.logical.WindowSpec.*;
 import static org.apache.tajo.plan.verifier.SyntaxErrorUtil.makeSyntaxError;
+import static org.apache.tajo.type.Type.Text;
 
 /**
  * <code>ExprAnnotator</code> makes an annotated expression called <code>EvalNode</code> from an
@@ -112,25 +111,25 @@ public static void assertEval(boolean condition, String message) throws TajoExce
    * @return a pair including left/right hand side terms
    */
   private static Pair<EvalNode, EvalNode> convertTypesIfNecessary(Context ctx, EvalNode lhs, EvalNode rhs) {
-    Type lhsType = lhs.getValueType().kind();
-    Type rhsType = rhs.getValueType().kind();
+    TajoDataTypes.Type lhsType = lhs.getValueType().kind();
+    TajoDataTypes.Type rhsType = rhs.getValueType().kind();
 
     // If one of both is NULL, it just returns the original types without casting.
-    if (lhsType == Type.NULL_TYPE || rhsType == Type.NULL_TYPE) {
+    if (lhsType == NULL_TYPE || rhsType == NULL_TYPE) {
       return new Pair<>(lhs, rhs);
     }
 
-    Type toBeCasted = TUtil.getFromNestedMap(CatalogUtil.OPERATION_CASTING_MAP, lhsType, rhsType);
+    TajoDataTypes.Type toBeCasted = TUtil.getFromNestedMap(CatalogUtil.OPERATION_CASTING_MAP, lhsType, rhsType);
     if (toBeCasted != null) { // if not null, one of either should be converted to another type.
       // Overwrite lhs, rhs, or both with cast expression.
 
       Direction direction = CatalogUtil.getCastingDirection(lhsType, rhsType);
 
       if (lhsType != toBeCasted && (direction == Direction.BOTH || direction == Direction.LHS)) {
-        lhs = convertType(ctx, lhs, CatalogUtil.newSimpleDataType(toBeCasted));
+        lhs = convertType(ctx, lhs, TypeFactory.create(toBeCasted));
       }
       if (rhsType != toBeCasted && (direction == Direction.BOTH || direction == Direction.RHS)) {
-        rhs = convertType(ctx, rhs, CatalogUtil.newSimpleDataType(toBeCasted));
+        rhs = convertType(ctx, rhs, TypeFactory.create(toBeCasted));
       }
     }
 
@@ -145,14 +144,14 @@ private static Pair<EvalNode, EvalNode> convertTypesIfNecessary(Context ctx, Eva
    * @param toType target type
    * @return type converted expression.
    */
-  private static EvalNode convertType(Context ctx, EvalNode evalNode, DataType toType) {
+  private static EvalNode convertType(Context ctx, EvalNode evalNode, org.apache.tajo.type.Type toType) {
 
     // if original and toType is the same, we don't need type conversion.
-    if (evalNode.getValueType().equals(convert(toType))) {
+    if (evalNode.getValueType().equals(toType)) {
       return evalNode;
     }
     // the conversion to null is not allowed.
-    if (evalNode.getValueType().isNull() || toType.getType() == Type.NULL_TYPE) {
+    if (evalNode.getValueType().isNull() || toType.isNull()) {
       return evalNode;
     }
 
@@ -324,7 +323,7 @@ public EvalNode visitBetween(Context ctx, Stack<Expr> stack, BetweenPredicate be
         between.isSymmetric(),
         predicand, begin, end);
 
-    betweenEval = (BetweenPredicateEval) convertType(ctx, betweenEval, widestType);
+    betweenEval = (BetweenPredicateEval) convertType(ctx, betweenEval, TypeConverter.convert(widestType));
     return betweenEval;
   }
 
@@ -360,7 +359,7 @@ public EvalNode visitCaseWhen(Context ctx, Stack<Expr> stack, CaseWhenPredicate
     assertEval(widestType != null, "Invalid Type Conversion for CaseWhen");
 
     // implicit type conversion
-    caseWhenEval = (CaseWhenEval) convertType(ctx, caseWhenEval, widestType);
+    caseWhenEval = (CaseWhenEval) convertType(ctx, caseWhenEval, TypeConverter.convert(widestType));
 
     return caseWhenEval;
   }
@@ -442,11 +441,11 @@ public EvalNode visitConcatenate(Context ctx, Stack<Expr> stack, BinaryOperator
     EvalNode rhs = visit(ctx, stack, expr.getRight());
     stack.pop();
 
-    if (lhs.getValueType().kind() != Type.TEXT) {
-      lhs = convertType(ctx, lhs, CatalogUtil.newSimpleDataType(Type.TEXT));
+    if (lhs.getValueType().kind() != TajoDataTypes.Type.TEXT) {
+      lhs = convertType(ctx, lhs, Text);
     }
-    if (rhs.getValueType().kind() != Type.TEXT) {
-      rhs = convertType(ctx, rhs, CatalogUtil.newSimpleDataType(Type.TEXT));
+    if (rhs.getValueType().kind() != TajoDataTypes.Type.TEXT) {
+      rhs = convertType(ctx, rhs, Text);
     }
 
     return new BinaryEval(EvalType.CONCATENATE, lhs, rhs);
@@ -610,14 +609,14 @@ public EvalNode visitFunction(Context ctx, Stack<Expr> stack, FunctionExpr expr)
         } else {
           lastDataType = CatalogUtil.newSimpleDataType(CatalogUtil.getPrimitiveTypeOf(lastDataType.getType()));
         }
-        givenArgs[i] = convertType(ctx, givenArgs[i], lastDataType);
+        givenArgs[i] = convertType(ctx, givenArgs[i], TypeConverter.convert(lastDataType));
       }
     } else {
       assertEval(funcDesc.getParamTypes().length == givenArgs.length,
           "The number of parameters is mismatched to the function definition: " + funcDesc.toString());
       // According to our function matching method, each given argument can be casted to the definition parameter.
       for (int i = 0; i < givenArgs.length; i++) {
-        givenArgs[i] = convertType(ctx, givenArgs[i], funcDesc.getParamTypes()[i]);
+        givenArgs[i] = convertType(ctx, givenArgs[i], TypeConverter.convert(funcDesc.getParamTypes()[i]));
       }
     }
 
@@ -665,7 +664,7 @@ public EvalNode visitGeneralSetFunction(Context ctx, Stack<Expr> stack, GeneralS
         FunctionType.DISTINCT_AGGREGATION : FunctionType.AGGREGATION;
     givenArgs[0] = visit(ctx, stack, params[0]);
     if (setFunction.getSignature().equalsIgnoreCase("count")) {
-      paramTypes[0] = CatalogUtil.newSimpleDataType(Type.ANY);
+      paramTypes[0] = CatalogUtil.newSimpleDataType(TajoDataTypes.Type.ANY);
     } else {
       paramTypes[0] = convert(givenArgs[0].getValueType()).getDataType();
     }
@@ -722,7 +721,7 @@ public EvalNode visitWindowFunction(Context ctx, Stack<Expr> stack, WindowFuncti
       if (windowFunc.getSignature().equalsIgnoreCase("count")) {
         paramTypes[0] = CatalogUtil.newSimpleDataType(TajoDataTypes.Type.ANY);
       } else if (windowFunc.getSignature().equalsIgnoreCase("row_number")) {
-        paramTypes[0] = CatalogUtil.newSimpleDataType(Type.INT8);
+        paramTypes[0] = CatalogUtil.newSimpleDataType(TajoDataTypes.Type.INT8);
       } else {
         paramTypes[0] = convert(givenArgs[0].getValueType()).getDataType();
       }
@@ -788,12 +787,10 @@ public EvalNode visitCastExpr(Context ctx, Stack<Expr> stack, CastExpr expr) thr
 
       // some cast operation may require earlier evaluation with timezone.
       return new ConstEval(
-          DatumFactory.cast(constEval.getValue(),
-              convert(LogicalPlanner.convertDataType(expr.getTarget())).getDataType(), ctx.timeZone));
+          DatumFactory.cast(constEval.getValue(), LogicalPlanner.convertDataType(expr.getTarget()), ctx.timeZone));
 
     } else {
-      return new CastEval(ctx.queryContext, child,
-          convert(LogicalPlanner.convertDataType(expr.getTarget())).getDataType());
+      return new CastEval(ctx.queryContext, child, LogicalPlanner.convertDataType(expr.getTarget()));
     }
   }
 
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/expr/CastEval.java b/tajo-plan/src/main/java/org/apache/tajo/plan/expr/CastEval.java
index 975b69e0e1..533fa66051 100644
--- a/tajo-plan/src/main/java/org/apache/tajo/plan/expr/CastEval.java
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/expr/CastEval.java
@@ -22,7 +22,6 @@
 import org.apache.tajo.OverridableConf;
 import org.apache.tajo.annotation.Nullable;
 import org.apache.tajo.catalog.Schema;
-import org.apache.tajo.catalog.TypeConverter;
 import org.apache.tajo.datum.Datum;
 import org.apache.tajo.datum.DatumFactory;
 import org.apache.tajo.storage.Tuple;
@@ -30,13 +29,11 @@
 
 import java.util.TimeZone;
 
-import static org.apache.tajo.common.TajoDataTypes.DataType;
-
 public class CastEval extends UnaryEval implements Cloneable {
-  @Expose private DataType target;
+  @Expose private Type target;
   private TimeZone timezone;
 
-  public CastEval(OverridableConf context, EvalNode operand, DataType target) {
+  public CastEval(OverridableConf context, EvalNode operand, Type target) {
     super(EvalType.CAST, operand);
     this.target = target;
   }
@@ -47,12 +44,12 @@ public EvalNode getOperand() {
 
   @Override
   public Type getValueType() {
-    return TypeConverter.convert(target);
+    return target;
   }
 
   @Override
   public String getName() {
-    return target.getType().name();
+    return target.toString();
   }
 
   @Override
@@ -76,7 +73,7 @@ public Datum eval(Tuple tuple) {
   }
 
   public String toString() {
-    return "CAST (" + child + " AS " + target.getType() + ")";
+    return "CAST (" + child + " AS " + target + ")";
   }
 
   @Override
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/serder/EvalNodeDeserializer.java b/tajo-plan/src/main/java/org/apache/tajo/plan/serder/EvalNodeDeserializer.java
index 3b369a1f28..6b38feb450 100644
--- a/tajo-plan/src/main/java/org/apache/tajo/plan/serder/EvalNodeDeserializer.java
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/serder/EvalNodeDeserializer.java
@@ -20,26 +20,29 @@
 
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
-
 import org.apache.tajo.OverridableConf;
 import org.apache.tajo.algebra.WindowSpec.WindowFrameEndBoundType;
 import org.apache.tajo.algebra.WindowSpec.WindowFrameStartBoundType;
 import org.apache.tajo.catalog.Column;
 import org.apache.tajo.catalog.FunctionDesc;
 import org.apache.tajo.catalog.SortSpec;
-import org.apache.tajo.exception.UndefinedFunctionException;
 import org.apache.tajo.catalog.proto.CatalogProtos;
 import org.apache.tajo.catalog.proto.CatalogProtos.FunctionSignatureProto;
 import org.apache.tajo.common.TajoDataTypes.DataType;
 import org.apache.tajo.datum.*;
 import org.apache.tajo.exception.TajoInternalError;
+import org.apache.tajo.exception.UndefinedFunctionException;
 import org.apache.tajo.plan.expr.*;
 import org.apache.tajo.plan.function.python.PythonScriptEngine;
 import org.apache.tajo.plan.logical.TableSubQueryNode;
 import org.apache.tajo.plan.logical.WindowSpec;
 import org.apache.tajo.plan.serder.PlanProto.WinFunctionEvalSpec;
+import org.apache.tajo.type.TypeProtobufEncoder;
 
-import java.util.*;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
 
 import static org.apache.tajo.function.FunctionUtil.buildSimpleFunctionSignature;
 
@@ -87,7 +90,7 @@ public int compare(PlanProto.EvalNode o1, PlanProto.EvalNode o2) {
           current = new IsNullEval(unaryProto.getNegative(), child);
           break;
         case CAST:
-          current = new CastEval(context, child, unaryProto.getCastingType());
+          current = new CastEval(context, child, TypeProtobufEncoder.decode(unaryProto.getCastingType()));
           break;
         case SIGNED:
           current = new SignedEval(unaryProto.getNegative(), child);
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/serder/EvalNodeSerializer.java b/tajo-plan/src/main/java/org/apache/tajo/plan/serder/EvalNodeSerializer.java
index 92d59c6937..ecaa882214 100644
--- a/tajo-plan/src/main/java/org/apache/tajo/plan/serder/EvalNodeSerializer.java
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/serder/EvalNodeSerializer.java
@@ -23,7 +23,6 @@
 import com.google.protobuf.ByteString;
 import org.apache.tajo.algebra.WindowSpec.WindowFrameEndBoundType;
 import org.apache.tajo.algebra.WindowSpec.WindowFrameStartBoundType;
-import org.apache.tajo.catalog.TypeConverter;
 import org.apache.tajo.catalog.proto.CatalogProtos;
 import org.apache.tajo.datum.AnyDatum;
 import org.apache.tajo.datum.Datum;
@@ -97,7 +96,7 @@ private PlanProto.EvalNode.Builder createEvalBuilder(EvalTreeProtoBuilderContext
 
     PlanProto.EvalNode.Builder nodeBuilder = PlanProto.EvalNode.newBuilder();
     nodeBuilder.setId(sid);
-    nodeBuilder.setDataType(TypeConverter.convert(node.getValueType()).getDataType());
+    nodeBuilder.setDataType(node.getValueType().getProto());
     nodeBuilder.setType(PlanProto.EvalType.valueOf(node.getType().name()));
     return nodeBuilder;
   }
@@ -119,7 +118,7 @@ public EvalNode visitUnaryEval(EvalTreeProtoBuilderContext context, UnaryEval un
       unaryBuilder.setNegative(signedEval.isNegative());
     } else if (unary.getType() == EvalType.CAST) {
       CastEval castEval = (CastEval) unary;
-      unaryBuilder.setCastingType(TypeConverter.convert(castEval.getValueType()).getDataType());
+      unaryBuilder.setCastingType(castEval.getValueType().getProto());
     }
 
     // registering itself and building EvalNode
diff --git a/tajo-plan/src/main/proto/Plan.proto b/tajo-plan/src/main/proto/Plan.proto
index c50429f613..99a0914e91 100644
--- a/tajo-plan/src/main/proto/Plan.proto
+++ b/tajo-plan/src/main/proto/Plan.proto
@@ -410,7 +410,7 @@ message EvalNodeTree {
 message EvalNode {
   required int32 id = 1;
   required EvalType type = 2;
-  required DataType data_type = 3;
+  required TypeProto data_type = 3;
 
   optional UnaryEval unary = 4; // NOT
   optional BinaryEval binary = 5;
@@ -429,7 +429,7 @@ message EvalNode {
 
 message UnaryEval {
   required int32 child_id = 1;
-  optional DataType castingType = 2;
+  optional TypeProto castingType = 2;
   optional bool negative = 3;
 }