tajo-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jihoon...@apache.org
Subject [2/8] tajo git commit: TAJO-269: Protocol buffer De/Serialization for LogicalNode.
Date Wed, 31 Dec 2014 10:51:23 GMT
http://git-wip-us.apache.org/repos/asf/tajo/blob/32be38d4/tajo-plan/src/main/java/org/apache/tajo/plan/rewrite/RewriteRule.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/rewrite/RewriteRule.java b/tajo-plan/src/main/java/org/apache/tajo/plan/rewrite/RewriteRule.java
deleted file mode 100644
index 0ba7460..0000000
--- a/tajo-plan/src/main/java/org/apache/tajo/plan/rewrite/RewriteRule.java
+++ /dev/null
@@ -1,56 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.tajo.plan.rewrite;
-
-import org.apache.tajo.plan.LogicalPlan;
-import org.apache.tajo.plan.PlanningException;
-
-/**
- * An interface for a rewrite rule.
- */
-public interface RewriteRule {
-
-  /**
-   * It returns the rewrite rule name. It will be used for debugging and
-   * building a optimization history.
-   *
-   * @return The rewrite rule name
-   */
-  String getName();
-
-  /**
-   * This method checks if this rewrite rule can be applied to a given query plan.
-   * For example, the selection push down can not be applied to the query plan without any filter.
-   * In such case, it will return false.
-   *
-   * @param plan The plan to be checked
-   * @return True if this rule can be applied to a given plan. Otherwise, false.
-   */
-  boolean isEligible(LogicalPlan plan);
-
-  /**
-   * Updates a logical plan and returns an updated logical plan rewritten by this rule.
-   * It must be guaranteed that the input logical plan is not modified even after rewrite.
-   * In other words, the rewrite has to modify an plan copied from the input plan.
-   *
-   * @param plan Input logical plan. It will not be modified.
-   * @return The rewritten logical plan.
-   */
-  LogicalPlan rewrite(LogicalPlan plan) throws PlanningException;
-}

http://git-wip-us.apache.org/repos/asf/tajo/blob/32be38d4/tajo-plan/src/main/java/org/apache/tajo/plan/rewrite/rules/FilterPushDownRule.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/rewrite/rules/FilterPushDownRule.java b/tajo-plan/src/main/java/org/apache/tajo/plan/rewrite/rules/FilterPushDownRule.java
index ed410f9..15750a1 100644
--- a/tajo-plan/src/main/java/org/apache/tajo/plan/rewrite/rules/FilterPushDownRule.java
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/rewrite/rules/FilterPushDownRule.java
@@ -21,6 +21,7 @@ package org.apache.tajo.plan.rewrite.rules;
 import com.google.common.collect.*;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.apache.tajo.OverridableConf;
 import org.apache.tajo.algebra.JoinType;
 import org.apache.tajo.catalog.CatalogUtil;
 import org.apache.tajo.catalog.Column;
@@ -30,7 +31,7 @@ import org.apache.tajo.plan.*;
 import org.apache.tajo.plan.expr.*;
 import org.apache.tajo.plan.logical.*;
 import org.apache.tajo.plan.rewrite.rules.FilterPushDownRule.FilterPushDownContext;
-import org.apache.tajo.plan.rewrite.RewriteRule;
+import org.apache.tajo.plan.rewrite.LogicalPlanRewriteRule;
 import org.apache.tajo.plan.util.PlannerUtil;
 import org.apache.tajo.plan.visitor.BasicLogicalPlanVisitor;
 import org.apache.tajo.util.TUtil;
@@ -42,7 +43,7 @@ import java.util.*;
  * It is likely to significantly reduces the intermediate data.
  */
 public class FilterPushDownRule extends BasicLogicalPlanVisitor<FilterPushDownContext, LogicalNode>
-    implements RewriteRule {
+    implements LogicalPlanRewriteRule {
   private final static Log LOG = LogFactory.getLog(FilterPushDownRule.class);
   private static final String NAME = "FilterPushDown";
 
@@ -79,7 +80,7 @@ public class FilterPushDownRule extends BasicLogicalPlanVisitor<FilterPushDownCo
   }
 
   @Override
-  public boolean isEligible(LogicalPlan plan) {
+  public boolean isEligible(OverridableConf queryContext, LogicalPlan plan) {
     for (LogicalPlan.QueryBlock block : plan.getQueryBlocks()) {
       if (block.hasNode(NodeType.SELECTION) || block.hasNode(NodeType.JOIN)) {
         return true;
@@ -89,7 +90,7 @@ public class FilterPushDownRule extends BasicLogicalPlanVisitor<FilterPushDownCo
   }
 
   @Override
-  public LogicalPlan rewrite(LogicalPlan plan) throws PlanningException {
+  public LogicalPlan rewrite(OverridableConf queryContext, LogicalPlan plan) throws PlanningException {
     /*
     FilterPushDown rule: processing when visits each node
       - If a target which is corresponding on a filter EvalNode's column is not FieldEval, do not PushDown.

http://git-wip-us.apache.org/repos/asf/tajo/blob/32be38d4/tajo-plan/src/main/java/org/apache/tajo/plan/rewrite/rules/LogicalPlanEqualityTester.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/rewrite/rules/LogicalPlanEqualityTester.java b/tajo-plan/src/main/java/org/apache/tajo/plan/rewrite/rules/LogicalPlanEqualityTester.java
new file mode 100644
index 0000000..8a24add
--- /dev/null
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/rewrite/rules/LogicalPlanEqualityTester.java
@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tajo.plan.rewrite.rules;
+
+import org.apache.tajo.OverridableConf;
+import org.apache.tajo.plan.LogicalPlan;
+import org.apache.tajo.plan.PlanningException;
+import org.apache.tajo.plan.logical.LogicalNode;
+import org.apache.tajo.plan.rewrite.LogicalPlanRewriteRule;
+import org.apache.tajo.plan.serder.LogicalNodeDeserializer;
+import org.apache.tajo.plan.serder.LogicalNodeSerializer;
+import org.apache.tajo.plan.serder.PlanProto;
+
+/**
+ * It verifies the equality between the input and output of LogicalNodeTree(De)Serializer in logical planning.
+ * It is used only for testing.
+ */
+@SuppressWarnings("unused")
+public class LogicalPlanEqualityTester implements LogicalPlanRewriteRule {
+
+  @Override
+  public String getName() {
+    return "LogicalPlanEqualityTester";
+  }
+
+  @Override
+  public boolean isEligible(OverridableConf queryContext, LogicalPlan plan) {
+    return true;
+  }
+
+  @Override
+  public LogicalPlan rewrite(OverridableConf queryContext, LogicalPlan plan) throws PlanningException {
+    LogicalNode root = plan.getRootBlock().getRoot();
+    PlanProto.LogicalNodeTree serialized = LogicalNodeSerializer.serialize(plan.getRootBlock().getRoot());
+    LogicalNode deserialized = LogicalNodeDeserializer.deserialize(queryContext, serialized);
+    assert root.deepEquals(deserialized);
+    return plan;
+  }
+}

http://git-wip-us.apache.org/repos/asf/tajo/blob/32be38d4/tajo-plan/src/main/java/org/apache/tajo/plan/rewrite/rules/PartitionedTableRewriter.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/rewrite/rules/PartitionedTableRewriter.java b/tajo-plan/src/main/java/org/apache/tajo/plan/rewrite/rules/PartitionedTableRewriter.java
index ea58437..7604c53 100644
--- a/tajo-plan/src/main/java/org/apache/tajo/plan/rewrite/rules/PartitionedTableRewriter.java
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/rewrite/rules/PartitionedTableRewriter.java
@@ -23,15 +23,15 @@ import com.google.common.collect.Sets;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.fs.*;
+import org.apache.tajo.OverridableConf;
 import org.apache.tajo.catalog.Column;
 import org.apache.tajo.catalog.Schema;
 import org.apache.tajo.catalog.TableDesc;
 import org.apache.tajo.catalog.partition.PartitionMethodDesc;
-import org.apache.tajo.conf.TajoConf;
 import org.apache.tajo.datum.DatumFactory;
 import org.apache.tajo.datum.NullDatum;
 import org.apache.tajo.plan.LogicalPlan;
-import org.apache.tajo.plan.rewrite.RewriteRule;
+import org.apache.tajo.plan.rewrite.LogicalPlanRewriteRule;
 import org.apache.tajo.plan.util.PlannerUtil;
 import org.apache.tajo.plan.PlanningException;
 import org.apache.tajo.plan.expr.*;
@@ -46,25 +46,19 @@ import java.util.List;
 import java.util.Set;
 import java.util.Stack;
 
-public class PartitionedTableRewriter implements RewriteRule {
+public class PartitionedTableRewriter implements LogicalPlanRewriteRule {
   private static final Log LOG = LogFactory.getLog(PartitionedTableRewriter.class);
 
   private static final String NAME = "Partitioned Table Rewriter";
   private final Rewriter rewriter = new Rewriter();
 
-  private final TajoConf systemConf;
-
-  public PartitionedTableRewriter(TajoConf conf) {
-    systemConf = conf;
-  }
-
   @Override
   public String getName() {
     return NAME;
   }
 
   @Override
-  public boolean isEligible(LogicalPlan plan) {
+  public boolean isEligible(OverridableConf queryContext, LogicalPlan plan) {
     for (LogicalPlan.QueryBlock block : plan.getQueryBlocks()) {
       for (RelationNode relation : block.getRelations()) {
         if (relation.getType() == NodeType.SCAN) {
@@ -79,9 +73,9 @@ public class PartitionedTableRewriter implements RewriteRule {
   }
 
   @Override
-  public LogicalPlan rewrite(LogicalPlan plan) throws PlanningException {
+  public LogicalPlan rewrite(OverridableConf queryContext, LogicalPlan plan) throws PlanningException {
     LogicalPlan.QueryBlock rootBlock = plan.getRootBlock();
-    rewriter.visit(rootBlock, plan, rootBlock, rootBlock.getRoot(), new Stack<LogicalNode>());
+    rewriter.visit(queryContext, plan, rootBlock, rootBlock.getRoot(), new Stack<LogicalNode>());
     return plan;
   }
 
@@ -120,10 +114,11 @@ public class PartitionedTableRewriter implements RewriteRule {
    * @return
    * @throws IOException
    */
-  private Path [] findFilteredPaths(Schema partitionColumns, EvalNode [] conjunctiveForms, Path tablePath)
+  private Path [] findFilteredPaths(OverridableConf queryContext, Schema partitionColumns, EvalNode [] conjunctiveForms,
+                                    Path tablePath)
       throws IOException {
 
-    FileSystem fs = tablePath.getFileSystem(systemConf);
+    FileSystem fs = tablePath.getFileSystem(queryContext.getConf());
 
     PathFilter [] filters;
     if (conjunctiveForms == null) {
@@ -223,7 +218,7 @@ public class PartitionedTableRewriter implements RewriteRule {
     return paths;
   }
 
-  private Path [] findFilteredPartitionPaths(ScanNode scanNode) throws IOException {
+  private Path [] findFilteredPartitionPaths(OverridableConf queryContext, ScanNode scanNode) throws IOException {
     TableDesc table = scanNode.getTableDesc();
     PartitionMethodDesc partitionDesc = scanNode.getTableDesc().getPartitionMethod();
 
@@ -262,10 +257,10 @@ public class PartitionedTableRewriter implements RewriteRule {
     }
 
     if (indexablePredicateSet.size() > 0) { // There are at least one indexable predicates
-      return findFilteredPaths(paritionValuesSchema,
+      return findFilteredPaths(queryContext, paritionValuesSchema,
           indexablePredicateSet.toArray(new EvalNode[indexablePredicateSet.size()]), new Path(table.getPath()));
     } else { // otherwise, we will get all partition paths.
-      return findFilteredPaths(paritionValuesSchema, null, new Path(table.getPath()));
+      return findFilteredPaths(queryContext, paritionValuesSchema, null, new Path(table.getPath()));
     }
   }
 
@@ -314,10 +309,11 @@ public class PartitionedTableRewriter implements RewriteRule {
     }
   }
 
-  private void updateTableStat(PartitionedTableScanNode scanNode) throws PlanningException {
+  private void updateTableStat(OverridableConf queryContext, PartitionedTableScanNode scanNode)
+      throws PlanningException {
     if (scanNode.getInputPaths().length > 0) {
       try {
-        FileSystem fs = scanNode.getInputPaths()[0].getFileSystem(systemConf);
+        FileSystem fs = scanNode.getInputPaths()[0].getFileSystem(queryContext.getConf());
         long totalVolume = 0;
 
         for (Path input : scanNode.getInputPaths()) {
@@ -396,10 +392,10 @@ public class PartitionedTableRewriter implements RewriteRule {
     return sb.toString();
   }
 
-  private final class Rewriter extends BasicLogicalPlanVisitor<Object, Object> {
+  private final class Rewriter extends BasicLogicalPlanVisitor<OverridableConf, Object> {
     @Override
-    public Object visitScan(Object object, LogicalPlan plan, LogicalPlan.QueryBlock block, ScanNode scanNode,
-                            Stack<LogicalNode> stack) throws PlanningException {
+    public Object visitScan(OverridableConf queryContext, LogicalPlan plan, LogicalPlan.QueryBlock block,
+                            ScanNode scanNode, Stack<LogicalNode> stack) throws PlanningException {
 
       TableDesc table = scanNode.getTableDesc();
       if (!table.hasPartition()) {
@@ -407,11 +403,11 @@ public class PartitionedTableRewriter implements RewriteRule {
       }
 
       try {
-        Path [] filteredPaths = findFilteredPartitionPaths(scanNode);
+        Path [] filteredPaths = findFilteredPartitionPaths(queryContext, scanNode);
         plan.addHistory("PartitionTableRewriter chooses " + filteredPaths.length + " of partitions");
         PartitionedTableScanNode rewrittenScanNode = plan.createNode(PartitionedTableScanNode.class);
         rewrittenScanNode.init(scanNode, filteredPaths);
-        updateTableStat(rewrittenScanNode);
+        updateTableStat(queryContext, rewrittenScanNode);
 
         // if it is topmost node, set it as the rootnode of this block.
         if (stack.empty() || block.getRoot().equals(scanNode)) {

http://git-wip-us.apache.org/repos/asf/tajo/blob/32be38d4/tajo-plan/src/main/java/org/apache/tajo/plan/rewrite/rules/ProjectionPushDownRule.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/rewrite/rules/ProjectionPushDownRule.java b/tajo-plan/src/main/java/org/apache/tajo/plan/rewrite/rules/ProjectionPushDownRule.java
index f7fd90d..abd2814 100644
--- a/tajo-plan/src/main/java/org/apache/tajo/plan/rewrite/rules/ProjectionPushDownRule.java
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/rewrite/rules/ProjectionPushDownRule.java
@@ -21,6 +21,7 @@ package org.apache.tajo.plan.rewrite.rules;
 import com.google.common.collect.*;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.apache.tajo.OverridableConf;
 import org.apache.tajo.annotation.Nullable;
 import org.apache.tajo.catalog.Column;
 import org.apache.tajo.catalog.Schema;
@@ -29,7 +30,7 @@ import org.apache.tajo.plan.*;
 import org.apache.tajo.plan.LogicalPlan.QueryBlock;
 import org.apache.tajo.plan.expr.*;
 import org.apache.tajo.plan.logical.*;
-import org.apache.tajo.plan.rewrite.RewriteRule;
+import org.apache.tajo.plan.rewrite.LogicalPlanRewriteRule;
 import org.apache.tajo.plan.util.PlannerUtil;
 import org.apache.tajo.catalog.SchemaUtil;
 import org.apache.tajo.plan.visitor.BasicLogicalPlanVisitor;
@@ -44,7 +45,7 @@ import java.util.*;
  * It also enables scanners to read only necessary columns.
  */
 public class ProjectionPushDownRule extends
-    BasicLogicalPlanVisitor<ProjectionPushDownRule.Context, LogicalNode> implements RewriteRule {
+    BasicLogicalPlanVisitor<ProjectionPushDownRule.Context, LogicalNode> implements LogicalPlanRewriteRule {
   /** Class Logger */
   private final Log LOG = LogFactory.getLog(ProjectionPushDownRule.class);
   private static final String name = "ProjectionPushDown";
@@ -55,7 +56,7 @@ public class ProjectionPushDownRule extends
   }
 
   @Override
-  public boolean isEligible(LogicalPlan plan) {
+  public boolean isEligible(OverridableConf queryContext, LogicalPlan plan) {
     LogicalNode toBeOptimized = plan.getRootBlock().getRoot();
 
     if (PlannerUtil.checkIfDDLPlan(toBeOptimized)) {
@@ -70,7 +71,7 @@ public class ProjectionPushDownRule extends
   }
 
   @Override
-  public LogicalPlan rewrite(LogicalPlan plan) throws PlanningException {
+  public LogicalPlan rewrite(OverridableConf queryContext, LogicalPlan plan) throws PlanningException {
     LogicalPlan.QueryBlock rootBlock = plan.getRootBlock();
 
     LogicalPlan.QueryBlock topmostBlock = rootBlock;
@@ -1044,7 +1045,7 @@ public class ProjectionPushDownRule extends
     if (node.hasTargets()) {
       targets = node.getTargets();
     } else {
-      targets = PlannerUtil.schemaToTargets(node.getTableSchema());
+      targets = PlannerUtil.schemaToTargets(node.getLogicalSchema());
     }
 
     LinkedHashSet<Target> projectedTargets = Sets.newLinkedHashSet();

http://git-wip-us.apache.org/repos/asf/tajo/blob/32be38d4/tajo-plan/src/main/java/org/apache/tajo/plan/serder/EvalNodeDeserializer.java
----------------------------------------------------------------------
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
new file mode 100644
index 0000000..322c8db
--- /dev/null
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/serder/EvalNodeDeserializer.java
@@ -0,0 +1,301 @@
+/*
+ * Lisensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tajo.plan.serder;
+
+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.catalog.exception.NoSuchFunctionException;
+import org.apache.tajo.catalog.proto.CatalogProtos;
+import org.apache.tajo.datum.*;
+import org.apache.tajo.exception.InternalException;
+import org.apache.tajo.plan.expr.*;
+import org.apache.tajo.plan.function.AggFunction;
+import org.apache.tajo.plan.function.GeneralFunction;
+import org.apache.tajo.plan.logical.WindowSpec;
+import org.apache.tajo.plan.serder.PlanProto.WinFunctionEvalSpec;
+
+import java.util.*;
+
+/**
+ * It deserializes a serialized eval tree consisting of a number of EvalNodes.
+ *
+ * {@link EvalNodeSerializer} serializes an eval tree in a postfix traverse order.
+ * So, this class firstly sorts all serialized eval nodes in ascending order of their sequence IDs. Then,
+ * it sequentially restores each serialized node to EvalNode instance.
+ *
+ * @see EvalNodeSerializer
+ */
+public class EvalNodeDeserializer {
+
+  public static EvalNode deserialize(OverridableConf context, PlanProto.EvalNodeTree tree) {
+    Map<Integer, EvalNode> evalNodeMap = Maps.newHashMap();
+
+    // sort serialized eval nodes in an ascending order of their IDs.
+    List<PlanProto.EvalNode> nodeList = Lists.newArrayList(tree.getNodesList());
+    Collections.sort(nodeList, new Comparator<PlanProto.EvalNode>() {
+      @Override
+      public int compare(PlanProto.EvalNode o1, PlanProto.EvalNode o2) {
+        return o1.getId() - o2.getId();
+      }
+    });
+
+    EvalNode current = null;
+
+    // The sorted order is the same of a postfix traverse order.
+    // So, it sequentially transforms each serialized node into a EvalNode instance in a postfix order of
+    // the original eval tree.
+
+    Iterator<PlanProto.EvalNode> it = nodeList.iterator();
+    while (it.hasNext()) {
+      PlanProto.EvalNode protoNode = it.next();
+
+      EvalType type = EvalType.valueOf(protoNode.getType().name());
+
+      if (EvalType.isUnaryOperator(type)) {
+        PlanProto.UnaryEval unaryProto = protoNode.getUnary();
+        EvalNode child = evalNodeMap.get(unaryProto.getChildId());
+
+        switch (type) {
+        case NOT:
+          current = new NotEval(child);
+          break;
+        case IS_NULL:
+          current = new IsNullEval(unaryProto.getNegative(), child);
+          break;
+        case CAST:
+          current = new CastEval(context, child, unaryProto.getCastingType());
+          break;
+        case SIGNED:
+          current = new SignedEval(unaryProto.getNegative(), child);
+          break;
+        default:
+          throw new RuntimeException("Unknown EvalType: " + type.name());
+        }
+
+      } else if (EvalType.isBinaryOperator(type)) {
+        PlanProto.BinaryEval binProto = protoNode.getBinary();
+        EvalNode lhs = evalNodeMap.get(binProto.getLhsId());
+        EvalNode rhs = evalNodeMap.get(binProto.getRhsId());
+
+        switch (type) {
+        case IN:
+          current = new InEval(lhs, (RowConstantEval) rhs, binProto.getNegative());
+          break;
+        case LIKE: {
+          PlanProto.PatternMatchEvalSpec patternMatchProto = protoNode.getPatternMatch();
+          current = new LikePredicateEval(binProto.getNegative(), lhs, (ConstEval) rhs,
+              patternMatchProto.getCaseSensitive());
+          break;
+        }
+        case REGEX: {
+          PlanProto.PatternMatchEvalSpec patternMatchProto = protoNode.getPatternMatch();
+          current = new RegexPredicateEval(binProto.getNegative(), lhs, (ConstEval) rhs,
+              patternMatchProto.getCaseSensitive());
+          break;
+        }
+        case SIMILAR_TO: {
+          PlanProto.PatternMatchEvalSpec patternMatchProto = protoNode.getPatternMatch();
+          current = new SimilarToPredicateEval(binProto.getNegative(), lhs, (ConstEval) rhs,
+              patternMatchProto.getCaseSensitive());
+          break;
+        }
+
+        default:
+          current = new BinaryEval(type, lhs, rhs);
+        }
+
+      } else if (type == EvalType.CONST) {
+        PlanProto.ConstEval constProto = protoNode.getConst();
+        current = new ConstEval(deserialize(constProto.getValue()));
+
+      } else if (type == EvalType.ROW_CONSTANT) {
+        PlanProto.RowConstEval rowConstProto = protoNode.getRowConst();
+        Datum[] values = new Datum[rowConstProto.getValuesCount()];
+        for (int i = 0; i < rowConstProto.getValuesCount(); i++) {
+          values[i] = deserialize(rowConstProto.getValues(i));
+        }
+        current = new RowConstantEval(values);
+
+      } else if (type == EvalType.FIELD) {
+        CatalogProtos.ColumnProto columnProto = protoNode.getField();
+        current = new FieldEval(new Column(columnProto));
+
+      } else if (type == EvalType.BETWEEN) {
+        PlanProto.BetweenEval betweenProto = protoNode.getBetween();
+        current = new BetweenPredicateEval(betweenProto.getNegative(), betweenProto.getSymmetric(),
+            evalNodeMap.get(betweenProto.getPredicand()),
+            evalNodeMap.get(betweenProto.getBegin()),
+            evalNodeMap.get(betweenProto.getEnd()));
+
+      } else if (type == EvalType.CASE) {
+        PlanProto.CaseWhenEval caseWhenProto = protoNode.getCasewhen();
+        CaseWhenEval caseWhenEval = new CaseWhenEval();
+        for (int i = 0; i < caseWhenProto.getIfCondsCount(); i++) {
+          caseWhenEval.addIfCond((CaseWhenEval.IfThenEval) evalNodeMap.get(caseWhenProto.getIfConds(i)));
+        }
+        if (caseWhenProto.hasElse()) {
+          caseWhenEval.setElseResult(evalNodeMap.get(caseWhenProto.getElse()));
+        }
+        current = caseWhenEval;
+
+      } else if (type == EvalType.IF_THEN) {
+        PlanProto.IfCondEval ifCondProto = protoNode.getIfCond();
+        current = new CaseWhenEval.IfThenEval(evalNodeMap.get(ifCondProto.getCondition()),
+            evalNodeMap.get(ifCondProto.getThen()));
+
+      } else if (EvalType.isFunction(type)) {
+        PlanProto.FunctionEval funcProto = protoNode.getFunction();
+
+        EvalNode [] params = new EvalNode[funcProto.getParamIdsCount()];
+        for (int i = 0; i < funcProto.getParamIdsCount(); i++) {
+          params[i] = evalNodeMap.get(funcProto.getParamIds(i));
+        }
+
+        FunctionDesc funcDesc = null;
+        try {
+          funcDesc = new FunctionDesc(funcProto.getFuncion());
+          if (type == EvalType.FUNCTION) {
+            GeneralFunction instance = (GeneralFunction) funcDesc.newInstance();
+            current = new GeneralFunctionEval(context, new FunctionDesc(funcProto.getFuncion()), instance, params);
+
+          } else if (type == EvalType.AGG_FUNCTION || type == EvalType.WINDOW_FUNCTION) {
+            AggFunction instance = (AggFunction) funcDesc.newInstance();
+            if (type == EvalType.AGG_FUNCTION) {
+              AggregationFunctionCallEval aggFunc =
+                  new AggregationFunctionCallEval(new FunctionDesc(funcProto.getFuncion()), instance, params);
+
+              PlanProto.AggFunctionEvalSpec aggFunctionProto = protoNode.getAggFunction();
+              aggFunc.setIntermediatePhase(aggFunctionProto.getIntermediatePhase());
+              aggFunc.setFinalPhase(aggFunctionProto.getFinalPhase());
+              if (aggFunctionProto.hasAlias()) {
+                aggFunc.setAlias(aggFunctionProto.getAlias());
+              }
+              current = aggFunc;
+
+            } else {
+              WinFunctionEvalSpec windowFuncProto = protoNode.getWinFunction();
+
+              WindowFunctionEval winFunc =
+                  new WindowFunctionEval(new FunctionDesc(funcProto.getFuncion()), instance, params,
+                      convertWindowFrame(windowFuncProto.getWindowFrame()));
+
+              if (windowFuncProto.getSortSpecCount() > 0) {
+                SortSpec[] sortSpecs = LogicalNodeDeserializer.convertSortSpecs(windowFuncProto.getSortSpecList());
+                winFunc.setSortSpecs(sortSpecs);
+              }
+
+              current = winFunc;
+            }
+          }
+        } catch (ClassNotFoundException cnfe) {
+          throw new NoSuchFunctionException(funcDesc.getFunctionName(), funcDesc.getParamTypes());
+        } catch (InternalException ie) {
+          throw new NoSuchFunctionException(funcDesc.getFunctionName(), funcDesc.getParamTypes());
+        }
+      } else {
+        throw new RuntimeException("Unknown EvalType: " + type.name());
+      }
+
+      evalNodeMap.put(protoNode.getId(), current);
+    }
+
+    return current;
+  }
+
+  private static WindowSpec.WindowFrame convertWindowFrame(WinFunctionEvalSpec.WindowFrame windowFrame) {
+    WindowFrameStartBoundType startBoundType = convertWindowStartBound(windowFrame.getStartBound().getBoundType());
+    WindowSpec.WindowStartBound startBound = new WindowSpec.WindowStartBound(startBoundType);
+
+    WindowFrameEndBoundType endBoundType = convertWindowEndBound(windowFrame.getEndBound().getBoundType());
+    WindowSpec.WindowEndBound endBound = new WindowSpec.WindowEndBound(endBoundType);
+
+    WindowSpec.WindowFrame frame = new WindowSpec.WindowFrame(startBound, endBound);
+    return frame;
+  }
+
+  private static WindowFrameStartBoundType convertWindowStartBound(
+      WinFunctionEvalSpec.WindowFrameStartBoundType type) {
+    if (type == WinFunctionEvalSpec.WindowFrameStartBoundType.S_UNBOUNDED_PRECEDING) {
+      return WindowFrameStartBoundType.UNBOUNDED_PRECEDING;
+    } else if (type == WinFunctionEvalSpec.WindowFrameStartBoundType.S_CURRENT_ROW) {
+      return WindowFrameStartBoundType.CURRENT_ROW;
+    } else if (type == WinFunctionEvalSpec.WindowFrameStartBoundType.S_PRECEDING) {
+      return WindowFrameStartBoundType.PRECEDING;
+    } else {
+      throw new IllegalStateException("Unknown Window Start Bound type: " + type.name());
+    }
+  }
+
+  private static WindowFrameEndBoundType convertWindowEndBound(
+      WinFunctionEvalSpec.WindowFrameEndBoundType type) {
+    if (type == WinFunctionEvalSpec.WindowFrameEndBoundType.E_UNBOUNDED_FOLLOWING) {
+      return WindowFrameEndBoundType.UNBOUNDED_FOLLOWING;
+    } else if (type == WinFunctionEvalSpec.WindowFrameEndBoundType.E_CURRENT_ROW) {
+      return WindowFrameEndBoundType.CURRENT_ROW;
+    } else if (type == WinFunctionEvalSpec.WindowFrameEndBoundType.E_FOLLOWING) {
+      return WindowFrameEndBoundType.FOLLOWING;
+    } else {
+      throw new IllegalStateException("Unknown Window Start Bound type: " + type.name());
+    }
+  }
+
+  public static Datum deserialize(PlanProto.Datum datum) {
+    switch (datum.getType()) {
+    case BOOLEAN:
+      return DatumFactory.createBool(datum.getBoolean());
+    case CHAR:
+      return DatumFactory.createChar(datum.getText());
+    case INT1:
+    case INT2:
+      return DatumFactory.createInt2((short) datum.getInt4());
+    case INT4:
+      return DatumFactory.createInt4(datum.getInt4());
+    case INT8:
+      return DatumFactory.createInt8(datum.getInt8());
+    case FLOAT4:
+      return DatumFactory.createFloat4(datum.getFloat4());
+    case FLOAT8:
+      return DatumFactory.createFloat8(datum.getFloat8());
+    case VARCHAR:
+    case TEXT:
+      return DatumFactory.createText(datum.getText());
+    case TIMESTAMP:
+      return new TimestampDatum(datum.getInt8());
+    case DATE:
+      return DatumFactory.createDate(datum.getInt4());
+    case TIME:
+      return DatumFactory.createTime(datum.getInt8());
+    case BINARY:
+    case BLOB:
+      return DatumFactory.createBlob(datum.getBlob().toByteArray());
+    case INTERVAL:
+      return new IntervalDatum(datum.getInterval().getMonth(), datum.getInterval().getMsec());
+    case NULL_TYPE:
+      return NullDatum.get();
+    default:
+      throw new RuntimeException("Unknown data type: " + datum.getType().name());
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/tajo/blob/32be38d4/tajo-plan/src/main/java/org/apache/tajo/plan/serder/EvalNodeSerializer.java
----------------------------------------------------------------------
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
new file mode 100644
index 0000000..c7702c5
--- /dev/null
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/serder/EvalNodeSerializer.java
@@ -0,0 +1,397 @@
+/*
+ * Lisensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tajo.plan.serder;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Maps;
+import com.google.protobuf.ByteString;
+import org.apache.tajo.algebra.WindowSpec.WindowFrameEndBoundType;
+import org.apache.tajo.algebra.WindowSpec.WindowFrameStartBoundType;
+import org.apache.tajo.catalog.proto.CatalogProtos;
+import org.apache.tajo.datum.Datum;
+import org.apache.tajo.datum.IntervalDatum;
+import org.apache.tajo.plan.expr.*;
+import org.apache.tajo.plan.logical.WindowSpec;
+import org.apache.tajo.plan.serder.PlanProto.WinFunctionEvalSpec;
+import org.apache.tajo.plan.serder.PlanProto.WinFunctionEvalSpec.WindowFrame;
+import org.apache.tajo.util.ProtoUtil;
+
+import java.util.Map;
+import java.util.Stack;
+
+/**
+ * It traverses an eval tree consisting of a number of {@link org.apache.tajo.plan.expr.EvalNode}
+ * in a postfix traverse order. The postfix traverse order guarantees that all child nodes of some node N
+ * were already visited when the node N is visited. This manner makes tree serialization possible in a simple logic.
+ */
+public class EvalNodeSerializer
+    extends SimpleEvalNodeVisitor<EvalNodeSerializer.EvalTreeProtoBuilderContext> {
+
+  private static final EvalNodeSerializer instance;
+
+  static {
+    instance = new EvalNodeSerializer();
+  }
+
+  public static class EvalTreeProtoBuilderContext {
+    private int seqId = 0;
+    private Map<EvalNode, Integer> idMap = Maps.newHashMap();
+    private PlanProto.EvalNodeTree.Builder treeBuilder = PlanProto.EvalNodeTree.newBuilder();
+  }
+
+  public static PlanProto.EvalNodeTree serialize(EvalNode evalNode) {
+    EvalNodeSerializer.EvalTreeProtoBuilderContext context =
+        new EvalNodeSerializer.EvalTreeProtoBuilderContext();
+    instance.visit(context, evalNode, new Stack<EvalNode>());
+    return context.treeBuilder.build();
+  }
+
+  /**
+   * Return child's serialization IDs. Usually, 0 is used for a child id of unary node or left child of
+   * binary node. 1 is used for right child of binary node. Between will use 0 as predicand, 1 as begin, and 2 as
+   * end eval node. For more detail, you should refer to each EvalNode implementation.
+   *
+   * @param context Context
+   * @param evalNode EvalNode
+   * @return The array of IDs which points to stored EvalNode.
+   * @see org.apache.tajo.plan.expr.EvalNode
+   */
+  private int [] registerGetChildIds(EvalTreeProtoBuilderContext context, EvalNode evalNode) {
+    int [] childIds = new int[evalNode.childNum()];
+    for (int i = 0; i < evalNode.childNum(); i++) {
+      if (context.idMap.containsKey(evalNode.getChild(i))) {
+        childIds[i] = context.idMap.get(evalNode.getChild(i));
+      } else {
+        childIds[i] = context.seqId++;
+      }
+    }
+    return childIds;
+  }
+
+  private PlanProto.EvalNode.Builder createEvalBuilder(EvalTreeProtoBuilderContext context, EvalNode node) {
+    int sid; // serialization sequence id
+    if (context.idMap.containsKey(node)) {
+      sid = context.idMap.get(node);
+    } else {
+      sid = context.seqId++;
+      context.idMap.put(node, sid);
+    }
+
+    PlanProto.EvalNode.Builder nodeBuilder = PlanProto.EvalNode.newBuilder();
+    nodeBuilder.setId(sid);
+    nodeBuilder.setDataType(node.getValueType());
+    nodeBuilder.setType(PlanProto.EvalType.valueOf(node.getType().name()));
+    return nodeBuilder;
+  }
+
+  @Override
+  public EvalNode visitUnaryEval(EvalTreeProtoBuilderContext context, Stack<EvalNode> stack, UnaryEval unary) {
+    // visiting and registering childs
+    super.visitUnaryEval(context, stack, unary);
+    int [] childIds = registerGetChildIds(context, unary);
+
+    // building itself
+    PlanProto.UnaryEval.Builder unaryBuilder = PlanProto.UnaryEval.newBuilder();
+    unaryBuilder.setChildId(childIds[0]);
+    if (unary.getType() == EvalType.IS_NULL) {
+      IsNullEval isNullEval = (IsNullEval) unary;
+      unaryBuilder.setNegative(isNullEval.isNot());
+    } else if (unary.getType() == EvalType.SIGNED) {
+      SignedEval signedEval = (SignedEval) unary;
+      unaryBuilder.setNegative(signedEval.isNegative());
+    } else if (unary.getType() == EvalType.CAST) {
+      CastEval castEval = (CastEval) unary;
+      unaryBuilder.setCastingType(castEval.getValueType());
+      if (castEval.hasTimeZone()) {
+        unaryBuilder.setTimezone(castEval.getTimezone().getID());
+      }
+    }
+
+    // registering itself and building EvalNode
+    PlanProto.EvalNode.Builder builder = createEvalBuilder(context, unary);
+    builder.setUnary(unaryBuilder);
+    context.treeBuilder.addNodes(builder);
+    return unary;
+  }
+
+  @Override
+  public EvalNode visitBinaryEval(EvalTreeProtoBuilderContext context, Stack<EvalNode> stack, BinaryEval binary) {
+    // visiting and registering childs
+    super.visitBinaryEval(context, stack, binary);
+    int [] childIds = registerGetChildIds(context, binary);
+
+    // registering itself and building EvalNode
+    PlanProto.EvalNode.Builder builder = createEvalBuilder(context, binary);
+
+    // building itself
+    PlanProto.BinaryEval.Builder binaryBuilder = PlanProto.BinaryEval.newBuilder();
+    binaryBuilder.setLhsId(childIds[0]);
+    binaryBuilder.setRhsId(childIds[1]);
+
+    if (binary instanceof InEval) {
+      binaryBuilder.setNegative(((InEval)binary).isNot());
+    } else if (binary instanceof PatternMatchPredicateEval) {
+      PatternMatchPredicateEval patternMatch = (PatternMatchPredicateEval) binary;
+      binaryBuilder.setNegative(patternMatch.isNot());
+      builder.setPatternMatch(
+          PlanProto.PatternMatchEvalSpec.newBuilder().setCaseSensitive(patternMatch.isCaseInsensitive()));
+    }
+
+    builder.setBinary(binaryBuilder);
+    context.treeBuilder.addNodes(builder);
+    return binary;
+  }
+
+  @Override
+  public EvalNode visitConst(EvalTreeProtoBuilderContext context, ConstEval constant, Stack<EvalNode> stack) {
+    PlanProto.EvalNode.Builder builder = createEvalBuilder(context, constant);
+    builder.setConst(PlanProto.ConstEval.newBuilder().setValue(serialize(constant.getValue())));
+    context.treeBuilder.addNodes(builder);
+    return constant;
+  }
+
+  @Override
+  public EvalNode visitRowConstant(EvalTreeProtoBuilderContext context, RowConstantEval rowConst,
+                                   Stack<EvalNode> stack) {
+
+    PlanProto.RowConstEval.Builder rowConstBuilder = PlanProto.RowConstEval.newBuilder();
+    for (Datum d : rowConst.getValues()) {
+      rowConstBuilder.addValues(serialize(d));
+    }
+
+    PlanProto.EvalNode.Builder builder = createEvalBuilder(context, rowConst);
+    builder.setRowConst(rowConstBuilder);
+    context.treeBuilder.addNodes(builder);
+    return rowConst;
+  }
+
+  public EvalNode visitField(EvalTreeProtoBuilderContext context, Stack<EvalNode> stack, FieldEval field) {
+    PlanProto.EvalNode.Builder builder = createEvalBuilder(context, field);
+    builder.setField(field.getColumnRef().getProto());
+    context.treeBuilder.addNodes(builder);
+    return field;
+  }
+
+  public EvalNode visitBetween(EvalTreeProtoBuilderContext context, BetweenPredicateEval between,
+                               Stack<EvalNode> stack) {
+    // visiting and registering childs
+    super.visitBetween(context, between, stack);
+    int [] childIds = registerGetChildIds(context, between);
+    Preconditions.checkState(childIds.length == 3, "Between must have three childs, but there are " + childIds.length
+        + " child nodes");
+
+    // building itself
+    PlanProto.BetweenEval.Builder betweenBuilder = PlanProto.BetweenEval.newBuilder();
+    betweenBuilder.setNegative(between.isNot());
+    betweenBuilder.setSymmetric(between.isSymmetric());
+    betweenBuilder.setPredicand(childIds[0]);
+    betweenBuilder.setBegin(childIds[1]);
+    betweenBuilder.setEnd(childIds[2]);
+
+    // registering itself and building EvalNode
+    PlanProto.EvalNode.Builder builder = createEvalBuilder(context, between);
+    builder.setBetween(betweenBuilder);
+    context.treeBuilder.addNodes(builder);
+    return between;
+  }
+
+  public EvalNode visitCaseWhen(EvalTreeProtoBuilderContext context, CaseWhenEval caseWhen, Stack<EvalNode> stack) {
+    // visiting and registering childs
+    super.visitCaseWhen(context, caseWhen, stack);
+    int [] childIds = registerGetChildIds(context, caseWhen);
+    Preconditions.checkState(childIds.length > 0, "Case When must have at least one child, but there is no child");
+
+    // building itself
+    PlanProto.CaseWhenEval.Builder caseWhenBuilder = PlanProto.CaseWhenEval.newBuilder();
+    int ifCondsNum = childIds.length - (caseWhen.hasElse() ? 1 : 0);
+    for (int i = 0; i < ifCondsNum; i++) {
+      caseWhenBuilder.addIfConds(childIds[i]);
+    }
+    if (caseWhen.hasElse()) {
+      caseWhenBuilder.setElse(childIds[childIds.length - 1]);
+    }
+
+    // registering itself and building EvalNode
+    PlanProto.EvalNode.Builder builder = createEvalBuilder(context, caseWhen);
+    builder.setCasewhen(caseWhenBuilder);
+    context.treeBuilder.addNodes(builder);
+
+    return caseWhen;
+  }
+
+  public EvalNode visitIfThen(EvalTreeProtoBuilderContext context, CaseWhenEval.IfThenEval ifCond,
+                              Stack<EvalNode> stack) {
+    // visiting and registering childs
+    super.visitIfThen(context, ifCond, stack);
+    int [] childIds = registerGetChildIds(context, ifCond);
+
+    // building itself
+    PlanProto.IfCondEval.Builder ifCondBuilder = PlanProto.IfCondEval.newBuilder();
+    ifCondBuilder.setCondition(childIds[0]);
+    ifCondBuilder.setThen(childIds[1]);
+
+    // registering itself and building EvalNode
+    PlanProto.EvalNode.Builder builder = createEvalBuilder(context, ifCond);
+    builder.setIfCond(ifCondBuilder);
+    context.treeBuilder.addNodes(builder);
+
+    return ifCond;
+  }
+
+  public EvalNode visitFuncCall(EvalTreeProtoBuilderContext context, FunctionEval function, Stack<EvalNode> stack) {
+    // visiting and registering childs
+    super.visitFuncCall(context, function, stack);
+    int [] childIds = registerGetChildIds(context, function);
+
+    // building itself
+    PlanProto.FunctionEval.Builder funcBuilder = PlanProto.FunctionEval.newBuilder();
+    funcBuilder.setFuncion(function.getFuncDesc().getProto());
+    for (int i = 0; i < childIds.length; i++) {
+      funcBuilder.addParamIds(childIds[i]);
+    }
+
+    // registering itself and building EvalNode
+    PlanProto.EvalNode.Builder builder = createEvalBuilder(context, function);
+    builder.setFunction(funcBuilder);
+
+    if (function instanceof AggregationFunctionCallEval) {
+      AggregationFunctionCallEval aggFunc = (AggregationFunctionCallEval) function;
+
+      PlanProto.AggFunctionEvalSpec.Builder aggFunctionEvalBuilder = PlanProto.AggFunctionEvalSpec.newBuilder();
+      aggFunctionEvalBuilder.setIntermediatePhase(aggFunc.isIntermediatePhase());
+      aggFunctionEvalBuilder.setFinalPhase(aggFunc.isFinalPhase());
+      if (aggFunc.hasAlias()) {
+        aggFunctionEvalBuilder.setAlias(aggFunc.getAlias());
+      }
+
+      builder.setAggFunction(aggFunctionEvalBuilder);
+    }
+
+
+    if (function instanceof WindowFunctionEval) {
+      WindowFunctionEval winFunc = (WindowFunctionEval) function;
+      WinFunctionEvalSpec.Builder windowFuncBuilder = WinFunctionEvalSpec.newBuilder();
+
+      if (winFunc.hasSortSpecs()) {
+        windowFuncBuilder.addAllSortSpec(ProtoUtil.<CatalogProtos.SortSpecProto>toProtoObjects
+            (winFunc.getSortSpecs()));
+      }
+
+      windowFuncBuilder.setWindowFrame(buildWindowFrame(winFunc.getWindowFrame()));
+      builder.setWinFunction(windowFuncBuilder);
+    }
+
+
+    context.treeBuilder.addNodes(builder);
+    return function;
+  }
+
+  private WindowFrame buildWindowFrame(WindowSpec.WindowFrame frame) {
+    WindowFrame.Builder windowFrameBuilder = WindowFrame.newBuilder();
+
+    WindowSpec.WindowStartBound startBound = frame.getStartBound();
+    WindowSpec.WindowEndBound endBound = frame.getEndBound();
+
+    WinFunctionEvalSpec.WindowStartBound.Builder startBoundBuilder = WinFunctionEvalSpec.WindowStartBound.newBuilder();
+    startBoundBuilder.setBoundType(convertStartBoundType(startBound.getBoundType()));
+
+    WinFunctionEvalSpec.WindowEndBound.Builder endBoundBuilder = WinFunctionEvalSpec.WindowEndBound.newBuilder();
+    endBoundBuilder.setBoundType(convertEndBoundType(endBound.getBoundType()));
+
+    windowFrameBuilder.setStartBound(startBoundBuilder);
+    windowFrameBuilder.setEndBound(endBoundBuilder);
+
+    return windowFrameBuilder.build();
+  }
+
+  private WinFunctionEvalSpec.WindowFrameStartBoundType convertStartBoundType(WindowFrameStartBoundType type) {
+    if (type == WindowFrameStartBoundType.UNBOUNDED_PRECEDING) {
+      return WinFunctionEvalSpec.WindowFrameStartBoundType.S_UNBOUNDED_PRECEDING;
+    } else if (type == WindowFrameStartBoundType.CURRENT_ROW) {
+      return WinFunctionEvalSpec.WindowFrameStartBoundType.S_CURRENT_ROW;
+    } else if (type == WindowFrameStartBoundType.PRECEDING) {
+      return WinFunctionEvalSpec.WindowFrameStartBoundType.S_PRECEDING;
+    } else {
+      throw new IllegalStateException("Unknown Window Start Bound type: " + type.name());
+    }
+  }
+
+  private WinFunctionEvalSpec.WindowFrameEndBoundType convertEndBoundType(WindowFrameEndBoundType type) {
+    if (type == WindowFrameEndBoundType.UNBOUNDED_FOLLOWING) {
+      return WinFunctionEvalSpec.WindowFrameEndBoundType.E_UNBOUNDED_FOLLOWING;
+    } else if (type == WindowFrameEndBoundType.CURRENT_ROW) {
+      return WinFunctionEvalSpec.WindowFrameEndBoundType.E_CURRENT_ROW;
+    } else if (type == WindowFrameEndBoundType.FOLLOWING) {
+      return WinFunctionEvalSpec.WindowFrameEndBoundType.E_FOLLOWING;
+    } else {
+      throw new IllegalStateException("Unknown Window End Bound type: " + type.name());
+    }
+  }
+
+  public static PlanProto.Datum serialize(Datum datum) {
+    PlanProto.Datum.Builder builder = PlanProto.Datum.newBuilder();
+
+    builder.setType(datum.type());
+
+    switch (datum.type()) {
+    case NULL_TYPE:
+      break;
+    case BOOLEAN:
+      builder.setBoolean(datum.asBool());
+      break;
+    case INT1:
+    case INT2:
+    case INT4:
+    case DATE:
+      builder.setInt4(datum.asInt4());
+      break;
+    case INT8:
+    case TIMESTAMP:
+    case TIME:
+      builder.setInt8(datum.asInt8());
+      break;
+    case FLOAT4:
+      builder.setFloat4(datum.asFloat4());
+      break;
+    case FLOAT8:
+      builder.setFloat8(datum.asFloat8());
+      break;
+    case CHAR:
+    case VARCHAR:
+    case TEXT:
+      builder.setText(datum.asChars());
+      break;
+    case BINARY:
+    case BLOB:
+      builder.setBlob(ByteString.copyFrom(datum.asByteArray()));
+      break;
+    case INTERVAL:
+      IntervalDatum interval = (IntervalDatum) datum;
+      PlanProto.Interval.Builder intervalBuilder = PlanProto.Interval.newBuilder();
+      intervalBuilder.setMonth(interval.getMonths());
+      intervalBuilder.setMsec(interval.getMilliSeconds());
+      builder.setInterval(intervalBuilder);
+      break;
+    default:
+      throw new RuntimeException("Unknown data type: " + datum.type().name());
+    }
+
+    return builder.build();
+  }
+}

http://git-wip-us.apache.org/repos/asf/tajo/blob/32be38d4/tajo-plan/src/main/java/org/apache/tajo/plan/serder/EvalTreeProtoDeserializer.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/serder/EvalTreeProtoDeserializer.java b/tajo-plan/src/main/java/org/apache/tajo/plan/serder/EvalTreeProtoDeserializer.java
deleted file mode 100644
index 89b4fc0..0000000
--- a/tajo-plan/src/main/java/org/apache/tajo/plan/serder/EvalTreeProtoDeserializer.java
+++ /dev/null
@@ -1,218 +0,0 @@
-/*
- * Lisensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.tajo.plan.serder;
-
-import com.google.common.collect.Lists;
-import com.google.common.collect.Maps;
-import org.apache.tajo.OverridableConf;
-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.datum.*;
-import org.apache.tajo.exception.InternalException;
-import org.apache.tajo.plan.expr.*;
-import org.apache.tajo.plan.function.AggFunction;
-import org.apache.tajo.plan.function.GeneralFunction;
-
-import java.util.*;
-
-/**
- * It deserializes a serialized eval tree consisting of a number of EvalNodes.
- *
- * {@link EvalTreeProtoSerializer} serializes an eval tree in a postfix traverse order.
- * So, this class firstly sorts all serialized eval nodes in ascending order of their sequence IDs. Then,
- * it sequentially restores each serialized node to EvalNode instance.
- *
- * @see EvalTreeProtoSerializer
- */
-public class EvalTreeProtoDeserializer {
-
-  public static EvalNode deserialize(OverridableConf context, PlanProto.EvalTree tree) {
-    Map<Integer, EvalNode> evalNodeMap = Maps.newHashMap();
-
-    // sort serialized eval nodes in an ascending order of their IDs.
-    List<PlanProto.EvalNode> nodeList = Lists.newArrayList(tree.getNodesList());
-    Collections.sort(nodeList, new Comparator<PlanProto.EvalNode>() {
-      @Override
-      public int compare(PlanProto.EvalNode o1, PlanProto.EvalNode o2) {
-        return o1.getId() - o2.getId();
-      }
-    });
-
-    EvalNode current = null;
-
-    // The sorted order is the same of a postfix traverse order.
-    // So, it sequentially transforms each serialized node into a EvalNode instance in a postfix order of
-    // the original eval tree.
-
-    Iterator<PlanProto.EvalNode> it = nodeList.iterator();
-    while (it.hasNext()) {
-      PlanProto.EvalNode protoNode = it.next();
-
-      EvalType type = EvalType.valueOf(protoNode.getType().name());
-
-      if (EvalType.isUnaryOperator(type)) {
-        PlanProto.UnaryEval unaryProto = protoNode.getUnary();
-        EvalNode child = evalNodeMap.get(unaryProto.getChildId());
-
-        switch (type) {
-        case NOT:
-          current = new NotEval(child);
-          break;
-        case IS_NULL:
-          current = new IsNullEval(unaryProto.getNegative(), child);
-          break;
-        case CAST:
-          current = new CastEval(context, child, unaryProto.getCastingType());
-          break;
-        case SIGNED:
-          current = new SignedEval(unaryProto.getNegative(), child);
-          break;
-        default:
-          throw new RuntimeException("Unknown EvalType: " + type.name());
-        }
-
-      } else if (EvalType.isBinaryOperator(type)) {
-        PlanProto.BinaryEval binProto = protoNode.getBinary();
-        EvalNode lhs = evalNodeMap.get(binProto.getLhsId());
-        EvalNode rhs = evalNodeMap.get(binProto.getRhsId());
-
-        switch (type) {
-        case IN:
-          current = new InEval(lhs, (RowConstantEval) rhs, binProto.getNegative());
-          break;
-        default:
-          current = new BinaryEval(type, lhs, rhs);
-        }
-
-      } else if (type == EvalType.CONST) {
-        PlanProto.ConstEval constProto = protoNode.getConst();
-        current = new ConstEval(deserialize(constProto.getValue()));
-
-      } else if (type == EvalType.ROW_CONSTANT) {
-        PlanProto.RowConstEval rowConstProto = protoNode.getRowConst();
-        Datum[] values = new Datum[rowConstProto.getValuesCount()];
-        for (int i = 0; i < rowConstProto.getValuesCount(); i++) {
-          values[i] = deserialize(rowConstProto.getValues(i));
-        }
-        current = new RowConstantEval(values);
-
-      } else if (type == EvalType.FIELD) {
-        CatalogProtos.ColumnProto columnProto = protoNode.getField();
-        current = new FieldEval(new Column(columnProto));
-
-      } else if (type == EvalType.BETWEEN) {
-        PlanProto.BetweenEval betweenProto = protoNode.getBetween();
-        current = new BetweenPredicateEval(betweenProto.getNegative(), betweenProto.getSymmetric(),
-            evalNodeMap.get(betweenProto.getPredicand()),
-            evalNodeMap.get(betweenProto.getBegin()),
-            evalNodeMap.get(betweenProto.getEnd()));
-
-      } else if (type == EvalType.CASE) {
-        PlanProto.CaseWhenEval caseWhenProto = protoNode.getCasewhen();
-        CaseWhenEval caseWhenEval = new CaseWhenEval();
-        for (int i = 0; i < caseWhenProto.getIfCondsCount(); i++) {
-          caseWhenEval.addIfCond((CaseWhenEval.IfThenEval) evalNodeMap.get(caseWhenProto.getIfConds(i)));
-        }
-        if (caseWhenProto.hasElse()) {
-          caseWhenEval.setElseResult(evalNodeMap.get(caseWhenProto.getElse()));
-        }
-        current = caseWhenEval;
-
-      } else if (type == EvalType.IF_THEN) {
-        PlanProto.IfCondEval ifCondProto = protoNode.getIfCond();
-        current = new CaseWhenEval.IfThenEval(evalNodeMap.get(ifCondProto.getCondition()),
-            evalNodeMap.get(ifCondProto.getThen()));
-
-      } else if (EvalType.isFunction(type)) {
-        PlanProto.FunctionEval funcProto = protoNode.getFunction();
-
-        EvalNode [] params = new EvalNode[funcProto.getParamIdsCount()];
-        for (int i = 0; i < funcProto.getParamIdsCount(); i++) {
-          params[i] = evalNodeMap.get(funcProto.getParamIds(i));
-        }
-
-        FunctionDesc funcDesc = null;
-        try {
-          funcDesc = new FunctionDesc(funcProto.getFuncion());
-          if (type == EvalType.FUNCTION) {
-            GeneralFunction instance = (GeneralFunction) funcDesc.newInstance();
-            current = new GeneralFunctionEval(context, new FunctionDesc(funcProto.getFuncion()), instance, params);
-          } else if (type == EvalType.AGG_FUNCTION || type == EvalType.WINDOW_FUNCTION) {
-            AggFunction instance = (AggFunction) funcDesc.newInstance();
-            if (type == EvalType.AGG_FUNCTION) {
-              current = new AggregationFunctionCallEval(new FunctionDesc(funcProto.getFuncion()), instance, params);
-            } else {
-              current = new WindowFunctionEval(new FunctionDesc(funcProto.getFuncion()), instance, params, null);
-            }
-          }
-        } catch (ClassNotFoundException cnfe) {
-          throw new NoSuchFunctionException(funcDesc.getFunctionName(), funcDesc.getParamTypes());
-        } catch (InternalException ie) {
-          throw new NoSuchFunctionException(funcDesc.getFunctionName(), funcDesc.getParamTypes());
-        }
-      } else {
-        throw new RuntimeException("Unknown EvalType: " + type.name());
-      }
-
-      evalNodeMap.put(protoNode.getId(), current);
-    }
-
-    return current;
-  }
-
-  public static Datum deserialize(PlanProto.Datum datum) {
-    switch (datum.getType()) {
-    case BOOLEAN:
-      return DatumFactory.createBool(datum.getBoolean());
-    case CHAR:
-      return DatumFactory.createChar(datum.getText());
-    case INT1:
-    case INT2:
-      return DatumFactory.createInt2((short) datum.getInt4());
-    case INT4:
-      return DatumFactory.createInt4(datum.getInt4());
-    case INT8:
-      return DatumFactory.createInt8(datum.getInt8());
-    case FLOAT4:
-      return DatumFactory.createFloat4(datum.getFloat4());
-    case FLOAT8:
-      return DatumFactory.createFloat8(datum.getFloat8());
-    case VARCHAR:
-    case TEXT:
-      return DatumFactory.createText(datum.getText());
-    case TIMESTAMP:
-      return new TimestampDatum(datum.getInt8());
-    case DATE:
-      return DatumFactory.createDate(datum.getInt4());
-    case TIME:
-      return DatumFactory.createTime(datum.getInt8());
-    case BINARY:
-    case BLOB:
-      return DatumFactory.createBlob(datum.getBlob().toByteArray());
-    case INTERVAL:
-      return new IntervalDatum(datum.getInterval().getMonth(), datum.getInterval().getMsec());
-    case NULL_TYPE:
-      return NullDatum.get();
-    default:
-      throw new RuntimeException("Unknown data type: " + datum.getType().name());
-    }
-  }
-}

http://git-wip-us.apache.org/repos/asf/tajo/blob/32be38d4/tajo-plan/src/main/java/org/apache/tajo/plan/serder/EvalTreeProtoSerializer.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/serder/EvalTreeProtoSerializer.java b/tajo-plan/src/main/java/org/apache/tajo/plan/serder/EvalTreeProtoSerializer.java
deleted file mode 100644
index 92a245f..0000000
--- a/tajo-plan/src/main/java/org/apache/tajo/plan/serder/EvalTreeProtoSerializer.java
+++ /dev/null
@@ -1,310 +0,0 @@
-/*
- * Lisensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.tajo.plan.serder;
-
-import com.google.common.base.Preconditions;
-import com.google.common.collect.Maps;
-import com.google.protobuf.ByteString;
-import org.apache.tajo.datum.Datum;
-import org.apache.tajo.datum.IntervalDatum;
-import org.apache.tajo.plan.expr.*;
-
-import java.util.Map;
-import java.util.Stack;
-
-/**
- * It traverses an eval tree consisting of a number of {@link org.apache.tajo.plan.expr.EvalNode}
- * in a postfix traverse order. The postfix traverse order guarantees that all child nodes of some node N
- * were already visited when the node N is visited. This manner makes tree serialization possible in a simple logic.
- */
-public class EvalTreeProtoSerializer
-    extends SimpleEvalNodeVisitor<EvalTreeProtoSerializer.EvalTreeProtoBuilderContext> {
-
-  private static final EvalTreeProtoSerializer instance;
-
-  static {
-    instance = new EvalTreeProtoSerializer();
-  }
-
-  public static class EvalTreeProtoBuilderContext {
-    private int seqId = 0;
-    private Map<EvalNode, Integer> idMap = Maps.newHashMap();
-    private PlanProto.EvalTree.Builder treeBuilder = PlanProto.EvalTree.newBuilder();
-  }
-
-  public static PlanProto.EvalTree serialize(EvalNode evalNode) {
-    EvalTreeProtoSerializer.EvalTreeProtoBuilderContext context =
-        new EvalTreeProtoSerializer.EvalTreeProtoBuilderContext();
-    instance.visit(context, evalNode, new Stack<EvalNode>());
-    return context.treeBuilder.build();
-  }
-
-  /**
-   * Return child's serialization IDs. Usually, 0 is used for a child id of unary node or left child of
-   * binary node. 1 is used for right child of binary node. Between will use 0 as predicand, 1 as begin, and 2 as
-   * end eval node. For more detail, you should refer to each EvalNode implementation.
-   *
-   * @param context Context
-   * @param evalNode EvalNode
-   * @return The array of IDs which points to stored EvalNode.
-   * @see org.apache.tajo.plan.expr.EvalNode
-   */
-  private int [] registerGetChildIds(EvalTreeProtoBuilderContext context, EvalNode evalNode) {
-    int [] childIds = new int[evalNode.childNum()];
-    for (int i = 0; i < evalNode.childNum(); i++) {
-      if (context.idMap.containsKey(evalNode.getChild(i))) {
-        childIds[i] = context.idMap.get(evalNode.getChild(i));
-      } else {
-        childIds[i] = context.seqId++;
-      }
-    }
-    return childIds;
-  }
-
-  private PlanProto.EvalNode.Builder createEvalBuilder(EvalTreeProtoBuilderContext context, EvalNode node) {
-    int sid; // serialization sequence id
-    if (context.idMap.containsKey(node)) {
-      sid = context.idMap.get(node);
-    } else {
-      sid = context.seqId++;
-      context.idMap.put(node, sid);
-    }
-
-    PlanProto.EvalNode.Builder nodeBuilder = PlanProto.EvalNode.newBuilder();
-    nodeBuilder.setId(sid);
-    nodeBuilder.setDataType(node.getValueType());
-    nodeBuilder.setType(PlanProto.EvalType.valueOf(node.getType().name()));
-    return nodeBuilder;
-  }
-
-  @Override
-  public EvalNode visitUnaryEval(EvalTreeProtoBuilderContext context, Stack<EvalNode> stack, UnaryEval unary) {
-    // visiting and registering childs
-    super.visitUnaryEval(context, stack, unary);
-    int [] childIds = registerGetChildIds(context, unary);
-
-    // building itself
-    PlanProto.UnaryEval.Builder unaryBuilder = PlanProto.UnaryEval.newBuilder();
-    unaryBuilder.setChildId(childIds[0]);
-    if (unary.getType() == EvalType.IS_NULL) {
-      IsNullEval isNullEval = (IsNullEval) unary;
-      unaryBuilder.setNegative(isNullEval.isNot());
-    } else if (unary.getType() == EvalType.SIGNED) {
-      SignedEval signedEval = (SignedEval) unary;
-      unaryBuilder.setNegative(signedEval.isNegative());
-    } else if (unary.getType() == EvalType.CAST) {
-      CastEval castEval = (CastEval) unary;
-      unaryBuilder.setCastingType(castEval.getValueType());
-      if (castEval.hasTimeZone()) {
-        unaryBuilder.setTimezone(castEval.getTimezone().getID());
-      }
-    }
-
-    // registering itself and building EvalNode
-    PlanProto.EvalNode.Builder builder = createEvalBuilder(context, unary);
-    builder.setUnary(unaryBuilder);
-    context.treeBuilder.addNodes(builder);
-    return unary;
-  }
-
-  @Override
-  public EvalNode visitBinaryEval(EvalTreeProtoBuilderContext context, Stack<EvalNode> stack, BinaryEval binary) {
-    // visiting and registering childs
-    super.visitBinaryEval(context, stack, binary);
-    int [] childIds = registerGetChildIds(context, binary);
-
-    // building itself
-    PlanProto.BinaryEval.Builder binaryBuilder = PlanProto.BinaryEval.newBuilder();
-    binaryBuilder.setLhsId(childIds[0]);
-    binaryBuilder.setRhsId(childIds[1]);
-
-    // registering itself and building EvalNode
-    PlanProto.EvalNode.Builder builder = createEvalBuilder(context, binary);
-    builder.setBinary(binaryBuilder);
-    context.treeBuilder.addNodes(builder);
-    return binary;
-  }
-
-  @Override
-  public EvalNode visitConst(EvalTreeProtoBuilderContext context, ConstEval constant, Stack<EvalNode> stack) {
-    PlanProto.EvalNode.Builder builder = createEvalBuilder(context, constant);
-    builder.setConst(PlanProto.ConstEval.newBuilder().setValue(serialize(constant.getValue())));
-    context.treeBuilder.addNodes(builder);
-    return constant;
-  }
-
-  @Override
-  public EvalNode visitRowConstant(EvalTreeProtoBuilderContext context, RowConstantEval rowConst,
-                                   Stack<EvalNode> stack) {
-
-    PlanProto.RowConstEval.Builder rowConstBuilder = PlanProto.RowConstEval.newBuilder();
-    for (Datum d : rowConst.getValues()) {
-      rowConstBuilder.addValues(serialize(d));
-    }
-
-    PlanProto.EvalNode.Builder builder = createEvalBuilder(context, rowConst);
-    builder.setRowConst(rowConstBuilder);
-    context.treeBuilder.addNodes(builder);
-    return rowConst;
-  }
-
-  public EvalNode visitField(EvalTreeProtoBuilderContext context, Stack<EvalNode> stack, FieldEval field) {
-    PlanProto.EvalNode.Builder builder = createEvalBuilder(context, field);
-    builder.setField(field.getColumnRef().getProto());
-    context.treeBuilder.addNodes(builder);
-    return field;
-  }
-
-  public EvalNode visitBetween(EvalTreeProtoBuilderContext context, BetweenPredicateEval between,
-                               Stack<EvalNode> stack) {
-    // visiting and registering childs
-    super.visitBetween(context, between, stack);
-    int [] childIds = registerGetChildIds(context, between);
-    Preconditions.checkState(childIds.length == 3, "Between must have three childs, but there are " + childIds.length
-        + " child nodes");
-
-    // building itself
-    PlanProto.BetweenEval.Builder betweenBuilder = PlanProto.BetweenEval.newBuilder();
-    betweenBuilder.setNegative(between.isNot());
-    betweenBuilder.setSymmetric(between.isSymmetric());
-    betweenBuilder.setPredicand(childIds[0]);
-    betweenBuilder.setBegin(childIds[1]);
-    betweenBuilder.setEnd(childIds[2]);
-
-    // registering itself and building EvalNode
-    PlanProto.EvalNode.Builder builder = createEvalBuilder(context, between);
-    builder.setBetween(betweenBuilder);
-    context.treeBuilder.addNodes(builder);
-    return between;
-  }
-
-  public EvalNode visitCaseWhen(EvalTreeProtoBuilderContext context, CaseWhenEval caseWhen, Stack<EvalNode> stack) {
-    // visiting and registering childs
-    super.visitCaseWhen(context, caseWhen, stack);
-    int [] childIds = registerGetChildIds(context, caseWhen);
-    Preconditions.checkState(childIds.length > 0, "Case When must have at least one child, but there is no child");
-
-    // building itself
-    PlanProto.CaseWhenEval.Builder caseWhenBuilder = PlanProto.CaseWhenEval.newBuilder();
-    int ifCondsNum = childIds.length - (caseWhen.hasElse() ? 1 : 0);
-    for (int i = 0; i < ifCondsNum; i++) {
-      caseWhenBuilder.addIfConds(childIds[i]);
-    }
-    if (caseWhen.hasElse()) {
-      caseWhenBuilder.setElse(childIds[childIds.length - 1]);
-    }
-
-    // registering itself and building EvalNode
-    PlanProto.EvalNode.Builder builder = createEvalBuilder(context, caseWhen);
-    builder.setCasewhen(caseWhenBuilder);
-    context.treeBuilder.addNodes(builder);
-
-    return caseWhen;
-  }
-
-  public EvalNode visitIfThen(EvalTreeProtoBuilderContext context, CaseWhenEval.IfThenEval ifCond,
-                              Stack<EvalNode> stack) {
-    // visiting and registering childs
-    super.visitIfThen(context, ifCond, stack);
-    int [] childIds = registerGetChildIds(context, ifCond);
-
-    // building itself
-    PlanProto.IfCondEval.Builder ifCondBuilder = PlanProto.IfCondEval.newBuilder();
-    ifCondBuilder.setCondition(childIds[0]);
-    ifCondBuilder.setThen(childIds[1]);
-
-    // registering itself and building EvalNode
-    PlanProto.EvalNode.Builder builder = createEvalBuilder(context, ifCond);
-    builder.setIfCond(ifCondBuilder);
-    context.treeBuilder.addNodes(builder);
-
-    return ifCond;
-  }
-
-  public EvalNode visitFuncCall(EvalTreeProtoBuilderContext context, FunctionEval function, Stack<EvalNode> stack) {
-    // visiting and registering childs
-    super.visitFuncCall(context, function, stack);
-    int [] childIds = registerGetChildIds(context, function);
-
-    // building itself
-    PlanProto.FunctionEval.Builder funcBuilder = PlanProto.FunctionEval.newBuilder();
-    funcBuilder.setFuncion(function.getFuncDesc().getProto());
-    for (int i = 0; i < childIds.length; i++) {
-      funcBuilder.addParamIds(childIds[i]);
-    }
-
-    // registering itself and building EvalNode
-    PlanProto.EvalNode.Builder builder = createEvalBuilder(context, function);
-    builder.setFunction(funcBuilder);
-    context.treeBuilder.addNodes(builder);
-
-    return function;
-  }
-
-  public static PlanProto.Datum serialize(Datum datum) {
-    PlanProto.Datum.Builder builder = PlanProto.Datum.newBuilder();
-
-    builder.setType(datum.type());
-
-    switch (datum.type()) {
-    case NULL_TYPE:
-      break;
-    case BOOLEAN:
-      builder.setBoolean(datum.asBool());
-      break;
-    case INT1:
-    case INT2:
-    case INT4:
-    case DATE:
-      builder.setInt4(datum.asInt4());
-      break;
-    case INT8:
-    case TIMESTAMP:
-    case TIME:
-      builder.setInt8(datum.asInt8());
-      break;
-    case FLOAT4:
-      builder.setFloat4(datum.asFloat4());
-      break;
-    case FLOAT8:
-      builder.setFloat8(datum.asFloat8());
-      break;
-    case CHAR:
-    case VARCHAR:
-    case TEXT:
-      builder.setText(datum.asChars());
-      break;
-    case BINARY:
-    case BLOB:
-      builder.setBlob(ByteString.copyFrom(datum.asByteArray()));
-      break;
-    case INTERVAL:
-      IntervalDatum interval = (IntervalDatum) datum;
-      PlanProto.Interval.Builder intervalBuilder = PlanProto.Interval.newBuilder();
-      intervalBuilder.setMonth(interval.getMonths());
-      intervalBuilder.setMsec(interval.getMilliSeconds());
-      builder.setInterval(intervalBuilder);
-      break;
-    default:
-      throw new RuntimeException("Unknown data type: " + datum.type().name());
-    }
-
-    return builder.build();
-  }
-}


Mime
View raw message