diff --git a/CHANGES b/CHANGES
index 3dd67a9..055b3c9 100644
--- a/CHANGES
+++ b/CHANGES
@@ -4,6 +4,9 @@
 
   NEW FEATURES 
 
+    TAJO-774: Implement logical plan part and physical executor for window 
+    function. (hyunsik)
+
     TAJO-847: Supporting MariaDB-based Store, which is compatible with MySQL.
     (Jinhang Choi via jihoon)
 
@@ -68,10 +71,6 @@
     TAJO-789: Improve shuffle URI. (jinho)
 
     TAJO-769: A minor improvements for HCatalogStore (Fengdong Yu via hyunsik)
-    
-  SUB TASKS:
-
-    TAJO-757: Add parser rule and algebraic expression for window function. (hyunsik)
 
     TAJO-734: Arrange TajoCli output message. (hyoungjunkim via jihoon)
 
diff --git a/tajo-algebra/src/main/java/org/apache/tajo/algebra/Aggregation.java b/tajo-algebra/src/main/java/org/apache/tajo/algebra/Aggregation.java
index bf95e89..edb523b 100644
--- a/tajo-algebra/src/main/java/org/apache/tajo/algebra/Aggregation.java
+++ b/tajo-algebra/src/main/java/org/apache/tajo/algebra/Aggregation.java
@@ -67,7 +67,23 @@
     return a && b;
   }
 
-  public static class GroupElement implements JsonSerializable {
+  @Override
+  public Object clone() throws CloneNotSupportedException {
+    Aggregation aggregation = (Aggregation) super.clone();
+
+    aggregation.namedExprs = new NamedExpr[namedExprs.length];
+    for (int i = 0; i < namedExprs.length; i++) {
+      aggregation.namedExprs[i] = (NamedExpr) namedExprs[i].clone();
+    }
+
+    aggregation.groups = new GroupElement[groups.length];
+    for (int i = 0; i < groups.length; i++) {
+      aggregation.groups[i] = (GroupElement) groups[i].clone();
+    }
+    return aggregation;
+  }
+
+  public static class GroupElement implements JsonSerializable, Cloneable {
     @Expose @SerializedName("GroupType")
     private GroupType group_type;
     @Expose @SerializedName("Dimensions")
@@ -110,6 +126,19 @@
 
       return false;
     }
+
+    @Override
+    public Object clone() throws CloneNotSupportedException {
+      GroupElement element = (GroupElement) super.clone();
+      element.group_type = group_type;
+      if (element.grouping_sets != null) {
+        element.grouping_sets = new Expr[grouping_sets.length];
+        for (int i = 0; i < grouping_sets.length; i++) {
+          element.grouping_sets[i] = (Expr) grouping_sets[i].clone();
+        }
+      }
+      return element;
+    }
   }
 
   public static enum GroupType {
diff --git a/tajo-algebra/src/main/java/org/apache/tajo/algebra/AlterTable.java b/tajo-algebra/src/main/java/org/apache/tajo/algebra/AlterTable.java
index 0f56bc2..6d72472 100644
--- a/tajo-algebra/src/main/java/org/apache/tajo/algebra/AlterTable.java
+++ b/tajo-algebra/src/main/java/org/apache/tajo/algebra/AlterTable.java
@@ -115,5 +115,15 @@
         TUtil.checkEquals(alterTableOpType, another.alterTableOpType);
   }
 
-
+  @Override
+  public Object clone() throws CloneNotSupportedException {
+    AlterTable alter = (AlterTable) super.clone();
+    alter.tableName = tableName;
+    alter.newTableName = newTableName;
+    alter.columnName = columnName;
+    alter.newColumnName = newColumnName;
+    alter.addNewColumn = (ColumnDefinition) addNewColumn.clone();
+    alter.alterTableOpType = alterTableOpType;
+    return alter;
+  }
 }
diff --git a/tajo-algebra/src/main/java/org/apache/tajo/algebra/AlterTablespace.java b/tajo-algebra/src/main/java/org/apache/tajo/algebra/AlterTablespace.java
index b9cb0c7..b17b53e 100644
--- a/tajo-algebra/src/main/java/org/apache/tajo/algebra/AlterTablespace.java
+++ b/tajo-algebra/src/main/java/org/apache/tajo/algebra/AlterTablespace.java
@@ -68,4 +68,13 @@
         TUtil.checkEquals(setType, another.setType) &&
         TUtil.checkEquals(uri, another.uri);
   }
+
+  @Override
+  public Object clone() throws CloneNotSupportedException {
+    AlterTablespace alter = (AlterTablespace) super.clone();
+    alter.setType = setType;
+    alter.tablespaceName = tablespaceName;
+    alter.uri = uri;
+    return alter;
+  }
 }
diff --git a/tajo-algebra/src/main/java/org/apache/tajo/algebra/BetweenPredicate.java b/tajo-algebra/src/main/java/org/apache/tajo/algebra/BetweenPredicate.java
index 4d4f09e..4b17bdb 100644
--- a/tajo-algebra/src/main/java/org/apache/tajo/algebra/BetweenPredicate.java
+++ b/tajo-algebra/src/main/java/org/apache/tajo/algebra/BetweenPredicate.java
@@ -75,4 +75,14 @@
     return symmetric == another.symmetric && predicand.equals(another.predicand) && begin.equals(another.begin) &&
         end.equals(another.end);
   }
+
+  public Object clone() throws CloneNotSupportedException {
+    BetweenPredicate between = (BetweenPredicate) super.clone();
+    between.not = not;
+    between.symmetric = symmetric;
+    between.predicand = (Expr) predicand.clone();
+    between.begin = (Expr) between.clone();
+    between.end = (Expr) end.clone();
+    return between;
+  }
 }
diff --git a/tajo-algebra/src/main/java/org/apache/tajo/algebra/BinaryOperator.java b/tajo-algebra/src/main/java/org/apache/tajo/algebra/BinaryOperator.java
index 0cc6376..4ca3788 100644
--- a/tajo-algebra/src/main/java/org/apache/tajo/algebra/BinaryOperator.java
+++ b/tajo-algebra/src/main/java/org/apache/tajo/algebra/BinaryOperator.java
@@ -67,4 +67,12 @@
   public String toString() {
     return left.toString() + " " + opType.toString() + " " + right.toString();
   }
+
+  @Override
+  public Object clone() throws CloneNotSupportedException {
+    BinaryOperator binaryOperator = (BinaryOperator) super.clone();
+    binaryOperator.left = (Expr) left.clone();
+    binaryOperator.right = (Expr) right.clone();
+    return binaryOperator;
+  }
 }
diff --git a/tajo-algebra/src/main/java/org/apache/tajo/algebra/CaseWhenPredicate.java b/tajo-algebra/src/main/java/org/apache/tajo/algebra/CaseWhenPredicate.java
index 41eba6e..8e97c31 100644
--- a/tajo-algebra/src/main/java/org/apache/tajo/algebra/CaseWhenPredicate.java
+++ b/tajo-algebra/src/main/java/org/apache/tajo/algebra/CaseWhenPredicate.java
@@ -69,11 +69,22 @@
   }
 
   @Override
+  public Object clone() throws CloneNotSupportedException {
+    CaseWhenPredicate caseWhen = (CaseWhenPredicate) super.clone();
+    caseWhen.whens = new ArrayList<WhenExpr>();
+    for (int i = 0; i < whens.size(); i++) {
+      caseWhen.whens.add((WhenExpr) whens.get(i).clone());
+    }
+    caseWhen.elseResult = elseResult != null ? (Expr) elseResult.clone() : null;
+    return caseWhen;
+  }
+
+  @Override
   public String toJson() {
     return JsonHelper.toJson(this);
   }
 
-  public static class WhenExpr {
+  public static class WhenExpr implements Cloneable {
     @Expose @SerializedName("Condition")
     Expr condition;
     @Expose @SerializedName("Result")
@@ -112,5 +123,13 @@
 
       return false;
     }
+
+    @Override
+    public Object clone() throws CloneNotSupportedException {
+      WhenExpr when = (WhenExpr) super.clone();
+      when.condition = (Expr) condition.clone();
+      when.result = (Expr) result.clone();
+      return when;
+    }
   }
 }
diff --git a/tajo-algebra/src/main/java/org/apache/tajo/algebra/ColumnDefinition.java b/tajo-algebra/src/main/java/org/apache/tajo/algebra/ColumnDefinition.java
index c20d409..e6e05d4 100644
--- a/tajo-algebra/src/main/java/org/apache/tajo/algebra/ColumnDefinition.java
+++ b/tajo-algebra/src/main/java/org/apache/tajo/algebra/ColumnDefinition.java
@@ -22,11 +22,11 @@
 
 public  class ColumnDefinition extends DataTypeExpr {
   @Expose @SerializedName("ColumnDefName")
-  String col_name;
+  String columnName;
 
   public ColumnDefinition(String columnName, String dataType) {
     super(dataType);
-    this.col_name = columnName;
+    this.columnName = columnName;
   }
 
   public ColumnDefinition(String columnName, DataTypeExpr dataType) {
@@ -37,17 +37,17 @@
         setScale(dataType.scale);
       }
     }
-    this.col_name = columnName;
+    this.columnName = columnName;
   }
 
   public String getColumnName() {
-    return this.col_name;
+    return this.columnName;
   }
 
   @Override
   public int hashCode() {
     int hash = super.hashCode();
-    return hash * 89 * col_name.hashCode();
+    return hash * 89 * columnName.hashCode();
 
   }
 
@@ -55,9 +55,16 @@
   public boolean equalsTo(Expr expr) {
     if (expr instanceof ColumnDefinition) {
       ColumnDefinition another = (ColumnDefinition) expr;
-      return col_name.equals(another.col_name) && super.equalsTo(another);
+      return columnName.equals(another.columnName) && super.equalsTo(another);
     }
 
     return false;
   }
+
+  @Override
+  public Object clone() throws CloneNotSupportedException {
+    ColumnDefinition column = (ColumnDefinition) super.clone();
+    column.columnName = columnName;
+    return column;
+  }
 }
diff --git a/tajo-algebra/src/main/java/org/apache/tajo/algebra/ColumnReferenceExpr.java b/tajo-algebra/src/main/java/org/apache/tajo/algebra/ColumnReferenceExpr.java
index 97bf332..456b570 100644
--- a/tajo-algebra/src/main/java/org/apache/tajo/algebra/ColumnReferenceExpr.java
+++ b/tajo-algebra/src/main/java/org/apache/tajo/algebra/ColumnReferenceExpr.java
@@ -91,4 +91,12 @@
   public String toString() {
     return qualifier != null ? qualifier + "." + name : name;
   }
+
+  @Override
+  public Object clone() throws CloneNotSupportedException {
+    ColumnReferenceExpr column = (ColumnReferenceExpr) super.clone();
+    column.qualifier = qualifier;
+    column.name = name;
+    return column;
+  }
 }
diff --git a/tajo-algebra/src/main/java/org/apache/tajo/algebra/CreateDatabase.java b/tajo-algebra/src/main/java/org/apache/tajo/algebra/CreateDatabase.java
index 2ff0ce7..1144b6e 100644
--- a/tajo-algebra/src/main/java/org/apache/tajo/algebra/CreateDatabase.java
+++ b/tajo-algebra/src/main/java/org/apache/tajo/algebra/CreateDatabase.java
@@ -83,4 +83,13 @@
         ifNotExists == another.ifNotExists;
 
   }
+
+  @Override
+  public Object clone() throws CloneNotSupportedException {
+    CreateDatabase createDb = (CreateDatabase) super.clone();
+    createDb.databaseName = databaseName;
+    createDb.tablespaceName = tablespaceName;
+    createDb.ifNotExists = ifNotExists;
+    return createDb;
+  }
 }
diff --git a/tajo-algebra/src/main/java/org/apache/tajo/algebra/CreateTable.java b/tajo-algebra/src/main/java/org/apache/tajo/algebra/CreateTable.java
index c74677d..bd04a91 100644
--- a/tajo-algebra/src/main/java/org/apache/tajo/algebra/CreateTable.java
+++ b/tajo-algebra/src/main/java/org/apache/tajo/algebra/CreateTable.java
@@ -24,6 +24,8 @@
 import com.google.gson.annotations.SerializedName;
 import org.apache.tajo.util.TUtil;
 
+import java.util.ArrayList;
+import java.util.HashMap;
 import java.lang.reflect.Type;
 import java.util.List;
 import java.util.Map;
@@ -179,6 +181,26 @@
         ifNotExists == another.ifNotExists;
   }
 
+  @Override
+  public Object clone() throws CloneNotSupportedException {
+    CreateTable createTable = (CreateTable) super.clone();
+    createTable.external = external;
+    createTable.tableName = tableName;
+    if (tableElements != null) {
+      createTable.tableElements = new ColumnDefinition[tableElements.length];
+      for (int i = 0; i < tableElements.length; i++) {
+        createTable.tableElements[i] = (ColumnDefinition) tableElements[i].clone();
+      }
+    }
+    createTable.storageType = storageType;
+    createTable.location = location;
+    createTable.subquery = subquery;
+    createTable.params = new HashMap<String, String>(params);
+    createTable.partition = (PartitionMethodDescExpr) partition.clone();
+    createTable.ifNotExists = ifNotExists;
+    return createTable;
+  }
+
   public static enum PartitionType {
     RANGE,
     HASH,
@@ -186,7 +208,7 @@
     COLUMN
   }
 
-  public static abstract class PartitionMethodDescExpr {
+  public static abstract class PartitionMethodDescExpr implements Cloneable {
     @Expose @SerializedName("PartitionType")
     PartitionType type;
 
@@ -198,6 +220,13 @@
       return type;
     }
 
+    @Override
+    public Object clone() throws CloneNotSupportedException {
+      PartitionMethodDescExpr partition = (PartitionMethodDescExpr) super.clone();
+      partition.type = type;
+      return partition;
+    }
+
     static class JsonSerDer implements JsonSerializer<PartitionMethodDescExpr>,
         JsonDeserializer<PartitionMethodDescExpr> {
 
@@ -237,7 +266,7 @@
     }
   }
 
-  public static class RangePartition extends PartitionMethodDescExpr {
+  public static class RangePartition extends PartitionMethodDescExpr implements Cloneable {
     @Expose @SerializedName("Columns")
     ColumnReferenceExpr [] columns;
     @Expose @SerializedName("Specifiers")
@@ -270,9 +299,25 @@
         return false;
       }
     }
+
+    @Override
+    public Object clone() throws CloneNotSupportedException {
+      RangePartition range = (RangePartition) super.clone();
+      range.columns = new ColumnReferenceExpr[columns.length];
+      for (int i = 0; i < columns.length; i++) {
+        range.columns[i] = (ColumnReferenceExpr) columns[i].clone();
+      }
+      if (range.specifiers != null) {
+        range.specifiers = new ArrayList<RangePartitionSpecifier>();
+        for (int i = 0; i < specifiers.size(); i++) {
+          range.specifiers.add(specifiers.get(i));
+        }
+      }
+      return range;
+    }
   }
 
-  public static class HashPartition extends PartitionMethodDescExpr {
+  public static class HashPartition extends PartitionMethodDescExpr implements Cloneable {
     @Expose @SerializedName("Columns")
     ColumnReferenceExpr [] columns;
     @Expose @SerializedName("Quantity")
@@ -325,6 +370,23 @@
         return false;
       }
     }
+
+    @Override
+    public Object clone() throws CloneNotSupportedException {
+      HashPartition hash = (HashPartition) super.clone();
+      hash.columns = new ColumnReferenceExpr[columns.length];
+      for (int i = 0; i < columns.length; i++) {
+        hash.columns[i] = (ColumnReferenceExpr) columns[i].clone();
+      }
+      hash.quantity = quantity;
+      if (specifiers != null) {
+        hash.specifiers = new ArrayList<PartitionSpecifier>();
+        for (PartitionSpecifier specifier : specifiers) {
+          hash.specifiers.add(specifier);
+        }
+      }
+      return hash;
+    }
   }
 
   public static class ListPartition extends PartitionMethodDescExpr {
@@ -360,6 +422,22 @@
         return false;
       }
     }
+
+    @Override
+    public Object clone() throws CloneNotSupportedException {
+      ListPartition listPartition = (ListPartition) super.clone();
+      listPartition.columns = new ColumnReferenceExpr[columns.length];
+      for (int i = 0; i < columns.length; i++) {
+        listPartition.columns[i] = (ColumnReferenceExpr) columns[i].clone();
+      }
+      if (specifiers != null) {
+        listPartition.specifiers = new ArrayList<ListPartitionSpecifier>();
+        for (ListPartitionSpecifier specifier : specifiers) {
+          listPartition.specifiers.add(specifier);
+        }
+      }
+      return listPartition;
+    }
   }
 
   public static class ColumnPartition extends PartitionMethodDescExpr {
@@ -368,35 +446,39 @@
     @Expose @SerializedName("IsOmitValues")
     private boolean isOmitValues;
 
-    public ColumnPartition(ColumnDefinition [] columns, boolean isOmitValues) {
+    public ColumnPartition(ColumnDefinition [] columns) {
       super(PartitionType.COLUMN);
       this.columns = columns;
-      this.isOmitValues = isOmitValues;
     }
 
     public ColumnDefinition [] getColumns() {
       return columns;
     }
 
-    public boolean isOmitValues() {
-      return isOmitValues;
-    }
-
     public int hashCode() {
-      return Objects.hashCode(Objects.hashCode(columns), isOmitValues);
+      return Objects.hashCode(columns);
     }
 
     public boolean equals(Object object) {
       if (object instanceof ColumnPartition) {
         ColumnPartition another = (ColumnPartition) object;
-        return type == another.type && TUtil.checkEquals(columns, another.columns) &&
-            TUtil.checkEquals(isOmitValues, another.isOmitValues);
+        return type == another.type && TUtil.checkEquals(columns, another.columns);
       }
       return false;
     }
+
+    @Override
+    public Object clone() throws CloneNotSupportedException {
+      ColumnPartition columnPartition = (ColumnPartition) super.clone();
+      columnPartition.columns = new ColumnDefinition[columns.length];
+      for (int i = 0; i < columns.length; i++) {
+        columnPartition.columns[i] = (ColumnDefinition) columns[i].clone();
+      }
+      return columnPartition;
+    }
   }
 
-  public static class RangePartitionSpecifier extends PartitionSpecifier {
+  public static class RangePartitionSpecifier extends PartitionSpecifier implements Cloneable {
     @Expose @SerializedName("End")
     Expr end;
     @Expose @SerializedName("IsMaxValue")
@@ -436,9 +518,17 @@
 
       return true;
     }
+
+    @Override
+    public Object clone() throws CloneNotSupportedException {
+      RangePartitionSpecifier specifier = (RangePartitionSpecifier) super.clone();
+      specifier.end = (Expr) end.clone();
+      specifier.maxValue = maxValue;
+      return specifier;
+    }
   }
 
-  public static class ListPartitionSpecifier extends PartitionSpecifier {
+  public static class ListPartitionSpecifier extends PartitionSpecifier implements Cloneable {
     @Expose @SerializedName("ValueList")
     ValueListExpr valueList;
 
@@ -465,9 +555,16 @@
 
       return valueList.equals(that.valueList);
     }
+
+    @Override
+    public Object clone() throws CloneNotSupportedException {
+      ListPartitionSpecifier specifier = (ListPartitionSpecifier) super.clone();
+      specifier.valueList = (ValueListExpr) valueList.clone();
+      return specifier;
+    }
   }
 
-  public static class PartitionSpecifier {
+  public static class PartitionSpecifier implements Cloneable {
     @Expose @SerializedName("PartitionSpecName")
     private String name;
 
@@ -490,5 +587,12 @@
         return false;
       }
     }
+
+    @Override
+    public Object clone() throws CloneNotSupportedException {
+      PartitionSpecifier specifier = (PartitionSpecifier) super.clone();
+      specifier.name = name;
+      return specifier;
+    }
   }
 }
diff --git a/tajo-algebra/src/main/java/org/apache/tajo/algebra/DataTypeExpr.java b/tajo-algebra/src/main/java/org/apache/tajo/algebra/DataTypeExpr.java
index 6978b84..9dc795b 100644
--- a/tajo-algebra/src/main/java/org/apache/tajo/algebra/DataTypeExpr.java
+++ b/tajo-algebra/src/main/java/org/apache/tajo/algebra/DataTypeExpr.java
@@ -76,4 +76,13 @@
         TUtil.checkEquals(lengthOrPrecision, another.lengthOrPrecision) &&
         TUtil.checkEquals(scale, another.scale);
   }
+
+  @Override
+  public Object clone() throws CloneNotSupportedException {
+    DataTypeExpr dataType = (DataTypeExpr) super.clone();
+    dataType.typeName = typeName;
+    dataType.lengthOrPrecision = lengthOrPrecision;
+    dataType.scale = scale;
+    return dataType;
+  }
 }
diff --git a/tajo-algebra/src/main/java/org/apache/tajo/algebra/DateLiteral.java b/tajo-algebra/src/main/java/org/apache/tajo/algebra/DateLiteral.java
index fafa8f8..297a44e 100644
--- a/tajo-algebra/src/main/java/org/apache/tajo/algebra/DateLiteral.java
+++ b/tajo-algebra/src/main/java/org/apache/tajo/algebra/DateLiteral.java
@@ -51,4 +51,11 @@
     }
     return false;
   }
+
+  @Override
+  public Object clone() throws CloneNotSupportedException {
+    DateLiteral newDate = (DateLiteral) super.clone();
+    newDate.date = date;
+    return newDate;
+  }
 }
diff --git a/tajo-algebra/src/main/java/org/apache/tajo/algebra/DateValue.java b/tajo-algebra/src/main/java/org/apache/tajo/algebra/DateValue.java
index 8424ab4..5878f92 100644
--- a/tajo-algebra/src/main/java/org/apache/tajo/algebra/DateValue.java
+++ b/tajo-algebra/src/main/java/org/apache/tajo/algebra/DateValue.java
@@ -22,7 +22,7 @@
 import com.google.gson.annotations.Expose;
 import com.google.gson.annotations.SerializedName;
 
-public class DateValue {
+public class DateValue implements Cloneable {
   @Expose @SerializedName("Year")
   private String years;
   @Expose @SerializedName("Month")
@@ -65,4 +65,13 @@
     }
     return false;
   }
+
+  @Override
+  public Object clone() throws CloneNotSupportedException {
+    DateValue date = (DateValue) super.clone();
+    date.years = years;
+    date.months = months;
+    date.days = days;
+    return date;
+  }
 }
diff --git a/tajo-algebra/src/main/java/org/apache/tajo/algebra/DropDatabase.java b/tajo-algebra/src/main/java/org/apache/tajo/algebra/DropDatabase.java
index fc949fd..e66b59d 100644
--- a/tajo-algebra/src/main/java/org/apache/tajo/algebra/DropDatabase.java
+++ b/tajo-algebra/src/main/java/org/apache/tajo/algebra/DropDatabase.java
@@ -52,4 +52,12 @@
     DropDatabase another = (DropDatabase) expr;
     return databaseName.equals(another.databaseName) && ifExists == another.ifExists;
   }
+
+  @Override
+  public Object clone() throws CloneNotSupportedException {
+    DropDatabase drop = (DropDatabase) super.clone();
+    drop.databaseName = databaseName;
+    drop.ifExists = ifExists;
+    return drop;
+  }
 }
diff --git a/tajo-algebra/src/main/java/org/apache/tajo/algebra/DropTable.java b/tajo-algebra/src/main/java/org/apache/tajo/algebra/DropTable.java
index c6e69dc..2c55b7f 100644
--- a/tajo-algebra/src/main/java/org/apache/tajo/algebra/DropTable.java
+++ b/tajo-algebra/src/main/java/org/apache/tajo/algebra/DropTable.java
@@ -24,11 +24,11 @@
 
 public class DropTable extends Expr {
   @Expose @SerializedName("TableName")
-  private final String tableName;
+  private String tableName;
   @Expose @SerializedName("IfExists")
-  private final boolean ifExists;
+  private boolean ifExists;
   @Expose @SerializedName("IsPurge")
-  private final boolean purge;
+  private boolean purge;
 
   public DropTable(String tableName, boolean ifExists, boolean purge) {
     super(OpType.DropTable);
@@ -64,4 +64,12 @@
     }
     return false;
   }
+
+  public Object clone() throws CloneNotSupportedException {
+    DropTable drop = (DropTable) super.clone();
+    drop.tableName = tableName;
+    drop.ifExists = ifExists;
+    drop.purge = purge;
+    return drop;
+  }
 }
diff --git a/tajo-algebra/src/main/java/org/apache/tajo/algebra/ExistsPredicate.java b/tajo-algebra/src/main/java/org/apache/tajo/algebra/ExistsPredicate.java
index 6bf2ddd..389fd74 100644
--- a/tajo-algebra/src/main/java/org/apache/tajo/algebra/ExistsPredicate.java
+++ b/tajo-algebra/src/main/java/org/apache/tajo/algebra/ExistsPredicate.java
@@ -50,4 +50,11 @@
     ExistsPredicate another = (ExistsPredicate) expr;
     return not == another.not;
   }
+
+  @Override
+  public Object clone() throws CloneNotSupportedException {
+    ExistsPredicate exists = (ExistsPredicate) super.clone();
+    exists.not = not;
+    return exists;
+  }
 }
diff --git a/tajo-algebra/src/main/java/org/apache/tajo/algebra/Expr.java b/tajo-algebra/src/main/java/org/apache/tajo/algebra/Expr.java
index 26e0831..76af393 100644
--- a/tajo-algebra/src/main/java/org/apache/tajo/algebra/Expr.java
+++ b/tajo-algebra/src/main/java/org/apache/tajo/algebra/Expr.java
@@ -25,7 +25,7 @@
 
 import java.lang.reflect.Type;
 
-public abstract class Expr implements JsonSerializable {
+public abstract class Expr implements JsonSerializable, Cloneable {
   @Expose @SerializedName("type")
   private static final String SERIALIZED_NAME_OF_OP_TYPE = "OpType";
   @Expose @SerializedName(SERIALIZED_NAME_OF_OP_TYPE)
@@ -85,6 +85,13 @@
 	}
 
   @Override
+  public Object clone() throws CloneNotSupportedException {
+    Expr newExpr = (Expr) super.clone();
+    newExpr.opType = opType;
+    return newExpr;
+  }
+
+  @Override
   public String toString() {
     return toJson();
   }
diff --git a/tajo-algebra/src/main/java/org/apache/tajo/algebra/FunctionExpr.java b/tajo-algebra/src/main/java/org/apache/tajo/algebra/FunctionExpr.java
index b776cd2..333c845 100644
--- a/tajo-algebra/src/main/java/org/apache/tajo/algebra/FunctionExpr.java
+++ b/tajo-algebra/src/main/java/org/apache/tajo/algebra/FunctionExpr.java
@@ -82,4 +82,17 @@
     return signature.equals(another.signature) &&
         TUtil.checkEquals(params, another.params);
   }
+
+  @Override
+  public Object clone() throws CloneNotSupportedException {
+    FunctionExpr func = (FunctionExpr) super.clone();
+    func.signature = signature;
+    if (params != null) {
+      func.params = new Expr[params.length];
+      for (int i = 0; i < params.length; i++) {
+        func.params[i] = (Expr) params[i].clone();
+      }
+    }
+    return func;
+  }
 }
diff --git a/tajo-algebra/src/main/java/org/apache/tajo/algebra/GeneralSetFunctionExpr.java b/tajo-algebra/src/main/java/org/apache/tajo/algebra/GeneralSetFunctionExpr.java
index 3d812e7..c10bd76 100644
--- a/tajo-algebra/src/main/java/org/apache/tajo/algebra/GeneralSetFunctionExpr.java
+++ b/tajo-algebra/src/main/java/org/apache/tajo/algebra/GeneralSetFunctionExpr.java
@@ -34,7 +34,6 @@
    */
   protected GeneralSetFunctionExpr(OpType type, String signature, boolean distinct, Expr [] params) {
     super(type, signature, params);
-    Preconditions.checkArgument(OpType.isAggregationFunction(type));
     this.distinct = distinct;
   }
 
@@ -66,4 +65,11 @@
   public boolean equalsTo(Expr expr) {
     return distinct == ((GeneralSetFunctionExpr)expr).distinct;
   }
+
+  @Override
+  public Object clone() throws CloneNotSupportedException {
+    GeneralSetFunctionExpr newSetFunc = (GeneralSetFunctionExpr) super.clone();
+    newSetFunc.distinct = distinct;
+    return newSetFunc;
+  }
 }
diff --git a/tajo-algebra/src/main/java/org/apache/tajo/algebra/Having.java b/tajo-algebra/src/main/java/org/apache/tajo/algebra/Having.java
index dbe6f21..5723496 100644
--- a/tajo-algebra/src/main/java/org/apache/tajo/algebra/Having.java
+++ b/tajo-algebra/src/main/java/org/apache/tajo/algebra/Having.java
@@ -62,4 +62,11 @@
     }
     return false;
   }
+
+  @Override
+  public Object clone() throws CloneNotSupportedException {
+    Having having = (Having) super.clone();
+    having.qual = (Expr) qual.clone();
+    return having;
+  }
 }
diff --git a/tajo-algebra/src/main/java/org/apache/tajo/algebra/InPredicate.java b/tajo-algebra/src/main/java/org/apache/tajo/algebra/InPredicate.java
index 52eb92a..51d89b7 100644
--- a/tajo-algebra/src/main/java/org/apache/tajo/algebra/InPredicate.java
+++ b/tajo-algebra/src/main/java/org/apache/tajo/algebra/InPredicate.java
@@ -41,4 +41,11 @@
   public Expr getInValue() {
     return right;
   }
+
+  @Override
+  public Object clone() throws CloneNotSupportedException {
+    InPredicate inPredicate = (InPredicate) super.clone();
+    inPredicate.not = not;
+    return inPredicate;
+  }
 }
diff --git a/tajo-algebra/src/main/java/org/apache/tajo/algebra/Insert.java b/tajo-algebra/src/main/java/org/apache/tajo/algebra/Insert.java
index 3f15957..ce9b703 100644
--- a/tajo-algebra/src/main/java/org/apache/tajo/algebra/Insert.java
+++ b/tajo-algebra/src/main/java/org/apache/tajo/algebra/Insert.java
@@ -23,6 +23,7 @@
 import com.google.gson.annotations.SerializedName;
 import org.apache.tajo.util.TUtil;
 
+import java.util.HashMap;
 import java.util.Map;
 
 public class Insert extends Expr {
@@ -140,4 +141,17 @@
         TUtil.checkEquals(location, another.location) &&
         TUtil.checkEquals(params, another.params);
   }
+
+  @Override
+  public Object clone() throws CloneNotSupportedException {
+    Insert insert = (Insert) super.clone();
+    insert.overwrite = overwrite;
+    insert.tableName = tableName;
+    insert.targetColumns = targetColumns != null ? targetColumns.clone() : null;
+    insert.storageType = storageType;
+    insert.location = location;
+    insert.subquery = (Expr) subquery.clone();
+    insert.params = new HashMap<String, String>(params);
+    return insert;
+  }
 }
diff --git a/tajo-algebra/src/main/java/org/apache/tajo/algebra/IsNullPredicate.java b/tajo-algebra/src/main/java/org/apache/tajo/algebra/IsNullPredicate.java
index f9f6e11..caa62c7 100644
--- a/tajo-algebra/src/main/java/org/apache/tajo/algebra/IsNullPredicate.java
+++ b/tajo-algebra/src/main/java/org/apache/tajo/algebra/IsNullPredicate.java
@@ -24,7 +24,7 @@
 
 public class IsNullPredicate extends UnaryOperator {
   @Expose @SerializedName("IsNot")
-  private final boolean not;
+  private boolean not;
 
   public IsNullPredicate(boolean not, Expr column) {
     super(OpType.IsNullPredicate);
@@ -49,4 +49,11 @@
     IsNullPredicate nullPredicate = (IsNullPredicate) expr;
     return not == nullPredicate.not;
   }
+
+  @Override
+  public Object clone() throws CloneNotSupportedException {
+    IsNullPredicate nullPredicate = (IsNullPredicate) super.clone();
+    nullPredicate.not = not;
+    return nullPredicate;
+  }
 }
diff --git a/tajo-algebra/src/main/java/org/apache/tajo/algebra/Join.java b/tajo-algebra/src/main/java/org/apache/tajo/algebra/Join.java
index 2acdde3..2b1f344 100644
--- a/tajo-algebra/src/main/java/org/apache/tajo/algebra/Join.java
+++ b/tajo-algebra/src/main/java/org/apache/tajo/algebra/Join.java
@@ -91,4 +91,19 @@
   public String toJson() {
     return JsonHelper.toJson(this);
   }
+
+  @Override
+  public Object clone() throws CloneNotSupportedException {
+    Join join = (Join) super.clone();
+    join.joinType = joinType;
+    join.joinQual = (Expr) joinQual.clone();
+    if (joinColumns != null) {
+      join.joinColumns = new ColumnReferenceExpr[joinColumns.length];
+      for (ColumnReferenceExpr colume : joinColumns) {
+        join.joinColumns = (ColumnReferenceExpr[]) colume.clone();
+      }
+    }
+    join.natural = natural;
+    return join;
+  }
 }
diff --git a/tajo-algebra/src/main/java/org/apache/tajo/algebra/Limit.java b/tajo-algebra/src/main/java/org/apache/tajo/algebra/Limit.java
index 20dad6b..3a49ac7 100644
--- a/tajo-algebra/src/main/java/org/apache/tajo/algebra/Limit.java
+++ b/tajo-algebra/src/main/java/org/apache/tajo/algebra/Limit.java
@@ -55,4 +55,11 @@
     Limit another = (Limit) expr;
     return fetchFirstNum.equals(another.fetchFirstNum);
   }
+
+  @Override
+  public Object clone() throws CloneNotSupportedException {
+    Limit limit = (Limit) super.clone();
+    limit.fetchFirstNum = (Expr) fetchFirstNum.clone();
+    return limit;
+  }
 }
diff --git a/tajo-algebra/src/main/java/org/apache/tajo/algebra/LiteralValue.java b/tajo-algebra/src/main/java/org/apache/tajo/algebra/LiteralValue.java
index 7747ff9..d829f70 100644
--- a/tajo-algebra/src/main/java/org/apache/tajo/algebra/LiteralValue.java
+++ b/tajo-algebra/src/main/java/org/apache/tajo/algebra/LiteralValue.java
@@ -87,4 +87,12 @@
       return LiteralType.String;
     }
   }
+
+  @Override
+  public LiteralValue clone() throws CloneNotSupportedException {
+    LiteralValue literal = (LiteralValue) super.clone();
+    literal.valueType = valueType;
+    literal.value = value;
+    return literal;
+  }
 }
diff --git a/tajo-algebra/src/main/java/org/apache/tajo/algebra/NamedExpr.java b/tajo-algebra/src/main/java/org/apache/tajo/algebra/NamedExpr.java
index 31e835f..7a0af05 100644
--- a/tajo-algebra/src/main/java/org/apache/tajo/algebra/NamedExpr.java
+++ b/tajo-algebra/src/main/java/org/apache/tajo/algebra/NamedExpr.java
@@ -89,4 +89,11 @@
   public String toJson() {
     return JsonHelper.toJson(this);
   }
+
+  @Override
+  public NamedExpr clone() throws CloneNotSupportedException {
+    NamedExpr namedExpr = (NamedExpr) super.clone();
+    namedExpr.alias = alias;
+    return namedExpr;
+  }
 }
diff --git a/tajo-algebra/src/main/java/org/apache/tajo/algebra/OpType.java b/tajo-algebra/src/main/java/org/apache/tajo/algebra/OpType.java
index bc6e89b..19c4ab5 100644
--- a/tajo-algebra/src/main/java/org/apache/tajo/algebra/OpType.java
+++ b/tajo-algebra/src/main/java/org/apache/tajo/algebra/OpType.java
@@ -94,11 +94,10 @@
   Column(ColumnReferenceExpr.class),
   Target(NamedExpr.class),
   Function(FunctionExpr.class),
-  WindowFunction(WindowFunctionExpr.class),
-  WindowSpec(WindowSpecExpr.class),
   Asterisk(QualifiedAsteriskExpr.class),
 
   // Set Functions
+  WindowFunction(WindowFunctionExpr.class),
   CountRowsFunction(CountRowsFunctionExpr.class),
   GeneralSetFunction(GeneralSetFunctionExpr.class),
 
@@ -142,13 +141,36 @@
     }
   }
 
+  public static boolean isLogicalType(OpType type) {
+    return type == Not || type == And || type == Or;
+  }
+
+  public static boolean isComparisonType(OpType type) {
+    return
+        type == OpType.Equals ||
+        type == OpType.NotEquals ||
+        type == OpType.LessThan ||
+        type == OpType.GreaterThan ||
+        type == OpType.LessThanOrEquals ||
+        type == OpType.GreaterThanOrEquals;
+  }
+
+  public static boolean isArithmeticType(OpType type) {
+    return
+        type == Plus ||
+        type == Minus ||
+        type == Multiply ||
+        type == Divide ||
+        type == Modular;
+  }
+
   /**
    * Check if it is one of the literal types.
    *
    * @param type The type to be checked
    * @return True if it is one of the literal types. Otherwise, it returns False.
    */
-  public static boolean isLiteral(OpType type) {
+  public static boolean isLiteralType(OpType type) {
     return  type == Literal ||
             type == NullLiteral ||
             type == TimeLiteral ||
@@ -163,7 +185,7 @@
    * @return True if it is aggregation function type. Otherwise, it returns False.
    */
   public static boolean isFunction(OpType type) {
-    return type == Function || isAggregationFunction(type);
+    return type == Function || isAggregationFunction(type) || isWindowFunction(type);
   }
 
   /**
@@ -175,4 +197,14 @@
   public static boolean isAggregationFunction(OpType type) {
     return type == GeneralSetFunction || type == CountRowsFunction;
   }
+
+  /**
+   * Check if it is an window function type.
+   *
+   * @param type The type to be checked
+   * @return True if it is window function type. Otherwise, it returns False.
+   */
+  public static boolean isWindowFunction(OpType type) {
+    return type == WindowFunction;
+  }
 }
diff --git a/tajo-algebra/src/main/java/org/apache/tajo/algebra/PatternMatchPredicate.java b/tajo-algebra/src/main/java/org/apache/tajo/algebra/PatternMatchPredicate.java
index 2399e03..14c7d41 100644
--- a/tajo-algebra/src/main/java/org/apache/tajo/algebra/PatternMatchPredicate.java
+++ b/tajo-algebra/src/main/java/org/apache/tajo/algebra/PatternMatchPredicate.java
@@ -65,4 +65,12 @@
     return opType == another.opType &&
         not == another.not && caseInsensitive == another.caseInsensitive;
   }
+
+  @Override
+  public PatternMatchPredicate clone() throws CloneNotSupportedException {
+    PatternMatchPredicate patternMatchPredicate = (PatternMatchPredicate) super.clone();
+    patternMatchPredicate.not = not;
+    patternMatchPredicate.caseInsensitive = caseInsensitive;
+    return patternMatchPredicate;
+  }
 }
diff --git a/tajo-algebra/src/main/java/org/apache/tajo/algebra/Projection.java b/tajo-algebra/src/main/java/org/apache/tajo/algebra/Projection.java
index 6ffc502..24c4399 100644
--- a/tajo-algebra/src/main/java/org/apache/tajo/algebra/Projection.java
+++ b/tajo-algebra/src/main/java/org/apache/tajo/algebra/Projection.java
@@ -69,10 +69,16 @@
     return JsonHelper.toJson(this);
   }
 
+  @Override
   public Projection clone() throws CloneNotSupportedException {
     Projection projection = (Projection)super.clone();
     projection.distinct = distinct;
-    projection.targets = targets;
+    if (targets != null) {
+      projection.targets = new NamedExpr[targets.length];
+      for (int i = 0; i < targets.length; i++) {
+        projection.targets[i] = targets[i].clone();
+      }
+    }
 
     return projection;
   }
diff --git a/tajo-algebra/src/main/java/org/apache/tajo/algebra/QualifiedAsteriskExpr.java b/tajo-algebra/src/main/java/org/apache/tajo/algebra/QualifiedAsteriskExpr.java
index f067e59..e7ab208 100644
--- a/tajo-algebra/src/main/java/org/apache/tajo/algebra/QualifiedAsteriskExpr.java
+++ b/tajo-algebra/src/main/java/org/apache/tajo/algebra/QualifiedAsteriskExpr.java
@@ -64,4 +64,11 @@
   public String toString() {
     return hasQualifier() ? qualifier + "." + ASTERISK : ASTERISK;
   }
+
+  @Override
+  public QualifiedAsteriskExpr clone() throws CloneNotSupportedException {
+    QualifiedAsteriskExpr qualifiedAsteriskExpr = (QualifiedAsteriskExpr) super.clone();
+    qualifiedAsteriskExpr.qualifier = qualifier;
+    return qualifiedAsteriskExpr;
+  }
 }
diff --git a/tajo-algebra/src/main/java/org/apache/tajo/algebra/Relation.java b/tajo-algebra/src/main/java/org/apache/tajo/algebra/Relation.java
index c865454..2092b67 100644
--- a/tajo-algebra/src/main/java/org/apache/tajo/algebra/Relation.java
+++ b/tajo-algebra/src/main/java/org/apache/tajo/algebra/Relation.java
@@ -74,4 +74,12 @@
     return TUtil.checkEquals(tableName, other.tableName) &&
         TUtil.checkEquals(alias, other.alias);
   }
+
+  @Override
+  public Object clone() throws CloneNotSupportedException {
+    Relation relation = (Relation) super.clone();
+    relation.tableName = tableName;
+    relation.alias = alias;
+    return relation;
+  }
 }
diff --git a/tajo-algebra/src/main/java/org/apache/tajo/algebra/RelationList.java b/tajo-algebra/src/main/java/org/apache/tajo/algebra/RelationList.java
index fdf6c94..ad7315b 100644
--- a/tajo-algebra/src/main/java/org/apache/tajo/algebra/RelationList.java
+++ b/tajo-algebra/src/main/java/org/apache/tajo/algebra/RelationList.java
@@ -68,4 +68,14 @@
     Set<Expr> anotherSet = TUtil.newHashSet(another.relations);
     return thisSet.equals(anotherSet);
   }
+
+  @Override
+  public Object clone() throws CloneNotSupportedException {
+    RelationList relationList = (RelationList) super.clone();
+    relationList.relations = new Expr[relations.length];
+    for (int i = 0; i < relations.length; i++) {
+      relationList.relations[i] = (Expr) relations[i].clone();
+    }
+    return relationList;
+  }
 }
diff --git a/tajo-algebra/src/main/java/org/apache/tajo/algebra/Selection.java b/tajo-algebra/src/main/java/org/apache/tajo/algebra/Selection.java
index 9cc245e..a2e6504 100644
--- a/tajo-algebra/src/main/java/org/apache/tajo/algebra/Selection.java
+++ b/tajo-algebra/src/main/java/org/apache/tajo/algebra/Selection.java
@@ -61,4 +61,11 @@
     }
     return false;
   }
+
+  @Override
+  public Object clone() throws CloneNotSupportedException {
+    Selection selection = (Selection) super.clone();
+    selection.qual = (Expr) qual.clone();
+    return selection;
+  }
 }
diff --git a/tajo-algebra/src/main/java/org/apache/tajo/algebra/SetOperation.java b/tajo-algebra/src/main/java/org/apache/tajo/algebra/SetOperation.java
index c20bf06..71ee1c8 100644
--- a/tajo-algebra/src/main/java/org/apache/tajo/algebra/SetOperation.java
+++ b/tajo-algebra/src/main/java/org/apache/tajo/algebra/SetOperation.java
@@ -52,4 +52,11 @@
     SetOperation another = (SetOperation) expr;
     return distinct == another.distinct;
   }
+
+  @Override
+  public Object clone() throws CloneNotSupportedException {
+    SetOperation setOperation = (SetOperation) super.clone();
+    setOperation.distinct = distinct;
+    return setOperation;
+  }
 }
diff --git a/tajo-algebra/src/main/java/org/apache/tajo/algebra/SignedExpr.java b/tajo-algebra/src/main/java/org/apache/tajo/algebra/SignedExpr.java
index 0c04341..c82b0f3 100644
--- a/tajo-algebra/src/main/java/org/apache/tajo/algebra/SignedExpr.java
+++ b/tajo-algebra/src/main/java/org/apache/tajo/algebra/SignedExpr.java
@@ -44,4 +44,11 @@
   boolean equalsTo(Expr expr) {
     return negative == ((SignedExpr)expr).negative;
   }
+
+  @Override
+  public Object clone() throws CloneNotSupportedException {
+    SignedExpr signedExpr = (SignedExpr) super.clone();
+    signedExpr.negative = negative;
+    return signedExpr;
+  }
 }
diff --git a/tajo-algebra/src/main/java/org/apache/tajo/algebra/Sort.java b/tajo-algebra/src/main/java/org/apache/tajo/algebra/Sort.java
index 0632b2e..7d5cb55 100644
--- a/tajo-algebra/src/main/java/org/apache/tajo/algebra/Sort.java
+++ b/tajo-algebra/src/main/java/org/apache/tajo/algebra/Sort.java
@@ -56,7 +56,17 @@
     return JsonHelper.toJson(this);
   }
 
-  public static class SortSpec {
+  @Override
+  public Object clone() throws CloneNotSupportedException {
+    Sort sort = (Sort) super.clone();
+    sort.sortSpecs = new SortSpec[sortSpecs.length];
+    for (int i = 0; i < sortSpecs.length; i++) {
+      sort.sortSpecs[i] = (SortSpec) sortSpecs[i].clone();
+    }
+    return sort;
+  }
+
+  public static class SortSpec implements Cloneable {
     @Expose @SerializedName("SortKey")
     private Expr key;
     @Expose @SerializedName("IsAsc")
@@ -98,6 +108,10 @@
       this.nullFirst = true;
     }
 
+    public void setKey(Expr expr) {
+      this.key = expr;
+    }
+
     public final Expr getKey() {
       return this.key;
     }
@@ -122,5 +136,14 @@
     public String toString() {
       return key + " " + (asc ? "asc" : "desc") + " " + (nullFirst ? "null first" :"");
     }
+
+    @Override
+    public Object clone() throws CloneNotSupportedException {
+      SortSpec sortSpec = (SortSpec) super.clone();
+      sortSpec.key = (Expr) key.clone();
+      sortSpec.asc = asc;
+      sortSpec.nullFirst = nullFirst;
+      return sortSpec;
+    }
   }
 }
diff --git a/tajo-algebra/src/main/java/org/apache/tajo/algebra/TablePrimarySubQuery.java b/tajo-algebra/src/main/java/org/apache/tajo/algebra/TablePrimarySubQuery.java
index 5c55164..6f08b0d 100644
--- a/tajo-algebra/src/main/java/org/apache/tajo/algebra/TablePrimarySubQuery.java
+++ b/tajo-algebra/src/main/java/org/apache/tajo/algebra/TablePrimarySubQuery.java
@@ -63,4 +63,14 @@
   public String toJson() {
     return JsonHelper.toJson(this);
   }
+
+  @Override
+  public Object clone() throws CloneNotSupportedException {
+    TablePrimarySubQuery subQuery = (TablePrimarySubQuery) super.clone();
+    subQuery.subquery = (Expr) subquery.clone();
+    if (columnNames != null) {
+      subQuery.columnNames = columnNames.clone();
+    }
+    return subQuery;
+  }
 }
diff --git a/tajo-algebra/src/main/java/org/apache/tajo/algebra/TimeLiteral.java b/tajo-algebra/src/main/java/org/apache/tajo/algebra/TimeLiteral.java
index 7baccd8..cba2293 100644
--- a/tajo-algebra/src/main/java/org/apache/tajo/algebra/TimeLiteral.java
+++ b/tajo-algebra/src/main/java/org/apache/tajo/algebra/TimeLiteral.java
@@ -51,4 +51,11 @@
     }
     return false;
   }
+
+  @Override
+  public Object clone() throws CloneNotSupportedException {
+    TimeLiteral timeLiteral = (TimeLiteral) super.clone();
+    timeLiteral.time = (TimeValue) timeLiteral.time.clone();
+    return timeLiteral;
+  }
 }
diff --git a/tajo-algebra/src/main/java/org/apache/tajo/algebra/TimeValue.java b/tajo-algebra/src/main/java/org/apache/tajo/algebra/TimeValue.java
index 6ad75bc..cc8cc88 100644
--- a/tajo-algebra/src/main/java/org/apache/tajo/algebra/TimeValue.java
+++ b/tajo-algebra/src/main/java/org/apache/tajo/algebra/TimeValue.java
@@ -23,7 +23,7 @@
 import com.google.gson.annotations.SerializedName;
 import org.apache.commons.lang.StringUtils;
 
-public class TimeValue {
+public class TimeValue implements Cloneable {
   @Expose @SerializedName("Hour")
   private String hours;
   @Expose @SerializedName("Minute")
@@ -82,4 +82,14 @@
   public int hashCode() {
     return Objects.hashCode(hours, minutes, seconds);
   }
+
+  @Override
+  public Object clone() throws CloneNotSupportedException {
+    TimeValue timeValue = (TimeValue) super.clone();
+    timeValue.hours = hours;
+    timeValue.minutes = minutes;
+    timeValue.seconds = seconds;
+    timeValue.secondsFraction = secondsFraction;
+    return timeValue;
+  }
 }
diff --git a/tajo-algebra/src/main/java/org/apache/tajo/algebra/TimestampLiteral.java b/tajo-algebra/src/main/java/org/apache/tajo/algebra/TimestampLiteral.java
index 5ee20d3..a9ed6da 100644
--- a/tajo-algebra/src/main/java/org/apache/tajo/algebra/TimestampLiteral.java
+++ b/tajo-algebra/src/main/java/org/apache/tajo/algebra/TimestampLiteral.java
@@ -22,7 +22,7 @@
 import com.google.gson.annotations.Expose;
 import com.google.gson.annotations.SerializedName;
 
-public class TimestampLiteral extends Expr {
+public class TimestampLiteral extends Expr implements Cloneable {
   @Expose @SerializedName("Date")
   private DateValue date;
   @Expose @SerializedName("Time")
@@ -58,4 +58,12 @@
     }
     return false;
   }
+
+  @Override
+  public Object clone() throws CloneNotSupportedException {
+    TimestampLiteral timestampLiteral = (TimestampLiteral) super.clone();
+    timestampLiteral.date = date;
+    timestampLiteral.time = time;
+    return timestampLiteral;
+  }
 }
diff --git a/tajo-algebra/src/main/java/org/apache/tajo/algebra/UnaryOperator.java b/tajo-algebra/src/main/java/org/apache/tajo/algebra/UnaryOperator.java
index e266393..b85d58c 100644
--- a/tajo-algebra/src/main/java/org/apache/tajo/algebra/UnaryOperator.java
+++ b/tajo-algebra/src/main/java/org/apache/tajo/algebra/UnaryOperator.java
@@ -45,4 +45,11 @@
   public int hashCode() {
     return Objects.hashCode(getType(), child.hashCode());
   }
+
+  @Override
+  public Object clone() throws CloneNotSupportedException {
+    UnaryOperator unaryOperator = (UnaryOperator) super.clone();
+    unaryOperator.child = (Expr) child.clone();
+    return unaryOperator;
+  }
 }
diff --git a/tajo-algebra/src/main/java/org/apache/tajo/algebra/ValueListExpr.java b/tajo-algebra/src/main/java/org/apache/tajo/algebra/ValueListExpr.java
index c5eafb6..32a5e82 100644
--- a/tajo-algebra/src/main/java/org/apache/tajo/algebra/ValueListExpr.java
+++ b/tajo-algebra/src/main/java/org/apache/tajo/algebra/ValueListExpr.java
@@ -46,4 +46,14 @@
     ValueListExpr valueListExpr = (ValueListExpr) expr;
     return TUtil.checkEquals(values, valueListExpr.values);
   }
+
+  @Override
+  public Object clone() throws CloneNotSupportedException {
+    ValueListExpr valueListExpr = (ValueListExpr) super.clone();
+    valueListExpr.values = new Expr[values.length];
+    for (int i = 0; i < values.length; i++) {
+      valueListExpr.values = (Expr[]) values[i].clone();
+    }
+    return valueListExpr;
+  }
 }
diff --git a/tajo-algebra/src/main/java/org/apache/tajo/algebra/Window.java b/tajo-algebra/src/main/java/org/apache/tajo/algebra/Window.java
index e9cd21c..8ac90a4 100644
--- a/tajo-algebra/src/main/java/org/apache/tajo/algebra/Window.java
+++ b/tajo-algebra/src/main/java/org/apache/tajo/algebra/Window.java
@@ -46,9 +46,9 @@
 
   public static class WindowDefinition {
     private String windowName;
-    private WindowSpecExpr windowSpec;
+    private WindowSpec windowSpec;
 
-    public WindowDefinition(String windowName, WindowSpecExpr spec) {
+    public WindowDefinition(String windowName, WindowSpec spec) {
       this.windowName = windowName;
       this.windowSpec = spec;
     }
@@ -57,7 +57,7 @@
       return windowName;
     }
 
-    public WindowSpecExpr getWindowSpec() {
+    public WindowSpec getWindowSpec() {
       return windowSpec;
     }
   }
diff --git a/tajo-algebra/src/main/java/org/apache/tajo/algebra/WindowFunctionExpr.java b/tajo-algebra/src/main/java/org/apache/tajo/algebra/WindowFunctionExpr.java
index 517ec0f..14ff930 100644
--- a/tajo-algebra/src/main/java/org/apache/tajo/algebra/WindowFunctionExpr.java
+++ b/tajo-algebra/src/main/java/org/apache/tajo/algebra/WindowFunctionExpr.java
@@ -19,19 +19,20 @@
 package org.apache.tajo.algebra;
 
 import com.google.common.base.Objects;
+import com.google.gson.annotations.Expose;
+import com.google.gson.annotations.SerializedName;
 import org.apache.tajo.util.TUtil;
 
-public class WindowFunctionExpr extends Expr {
-  // set function
-  GeneralSetFunctionExpr function;
+public class WindowFunctionExpr extends GeneralSetFunctionExpr {
 
   // over clause - only one of both is used.
+  @Expose @SerializedName("WindowName")
   private String windowName;
-  private WindowSpecExpr windowSpec;
+  @Expose @SerializedName("WindowSpec")
+  private WindowSpec windowSpec;
 
   public WindowFunctionExpr(GeneralSetFunctionExpr function) {
-    super(OpType.WindowFunction);
-    this.function = function;
+    super(OpType.WindowFunction, function.getSignature(), function.isDistinct(), function.getParams());
   }
 
   public boolean hasWindowName() {
@@ -50,11 +51,11 @@
     return windowSpec != null;
   }
 
-  public void setWindowSpec(WindowSpecExpr windowSpec) {
+  public void setWindowSpec(WindowSpec windowSpec) {
     this.windowSpec = windowSpec;
   }
 
-  public WindowSpecExpr getWindowSpec() {
+  public WindowSpec getWindowSpec() {
     return this.windowSpec;
   }
 
@@ -64,7 +65,7 @@
   }
 
   @Override
-  boolean equalsTo(Expr expr) {
+  public boolean equalsTo(Expr expr) {
     return TUtil.checkEquals(windowName, windowSpec);
   }
 }
diff --git a/tajo-algebra/src/main/java/org/apache/tajo/algebra/WindowSpec.java b/tajo-algebra/src/main/java/org/apache/tajo/algebra/WindowSpec.java
new file mode 100644
index 0000000..da10b91
--- /dev/null
+++ b/tajo-algebra/src/main/java/org/apache/tajo/algebra/WindowSpec.java
@@ -0,0 +1,242 @@
+/**
+ * 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.algebra;
+
+import com.google.common.base.Objects;
+import com.google.gson.annotations.Expose;
+import org.apache.tajo.util.TUtil;
+
+public class WindowSpec implements Cloneable {
+  @Expose private String windowName;
+  @Expose private Expr [] partitionKeys; // OVER (PARTITION BY ?,...,?)
+  @Expose private Sort.SortSpec [] sortSpecs; // OVER (... ORDER BY ?,...,?)
+  @Expose private WindowFrame windowFrame;
+
+  public boolean hasWindowName() {
+    return windowName != null;
+  }
+
+  public void setWindowName(String windowName) {
+    this.windowName = windowName;
+  }
+
+  public String getWindowName() {
+    return this.windowName;
+  }
+
+  public boolean hasPartitionBy() {
+    return this.partitionKeys != null;
+  }
+
+  public void setPartitionKeys(Expr[] partitionKeys) {
+    this.partitionKeys = partitionKeys;
+  }
+
+  public Expr [] getPartitionKeys() {
+    return this.partitionKeys;
+  }
+
+  public boolean hasOrderBy() {
+    return this.sortSpecs != null;
+  }
+
+  public void setSortSpecs(Sort.SortSpec [] sortSpecs) {
+    this.sortSpecs = sortSpecs;
+  }
+
+  public Sort.SortSpec [] getSortSpecs() {
+    return this.sortSpecs;
+  }
+
+  public boolean hasWindowFrame() {
+    return windowFrame != null;
+  }
+
+  public void setWindowFrame(WindowFrame frame) {
+    this.windowFrame = frame;
+  }
+
+  public WindowFrame getWindowFrame() {
+    return windowFrame;
+  }
+
+  public Object clone() throws CloneNotSupportedException {
+    WindowSpec windowSpec = (WindowSpec) super.clone();
+    windowSpec.windowName = windowName;
+    if (hasPartitionBy()) {
+      windowSpec.partitionKeys = new Expr[windowSpec.partitionKeys.length];
+      for (int i = 0; i < partitionKeys.length; i++) {
+        windowSpec.partitionKeys[i] = (Expr) partitionKeys[i].clone();
+      }
+    }
+    if (hasOrderBy()) {
+      windowSpec.sortSpecs = new Sort.SortSpec[sortSpecs.length];
+      for (int i = 0; i < sortSpecs.length; i++) {
+        windowSpec.sortSpecs[i] = (Sort.SortSpec) sortSpecs[i].clone();
+      }
+    }
+    if (hasWindowFrame()) {
+      windowSpec.windowFrame = (WindowFrame) windowFrame.clone();
+    }
+    return windowSpec;
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hashCode(windowName, partitionKeys, sortSpecs);
+  }
+
+  @Override
+  public boolean equals(Object obj) {
+    if (obj == this) {
+      return true;
+    }
+
+    if (obj instanceof WindowSpec) {
+      WindowSpec another = (WindowSpec) obj;
+      return TUtil.checkEquals(windowName, another.windowName) &&
+          TUtil.checkEquals(partitionKeys, another.partitionKeys) &&
+          TUtil.checkEquals(sortSpecs, another.sortSpecs) &&
+          TUtil.checkEquals(windowFrame, another.windowFrame);
+    } else {
+      return false;
+    }
+
+  }
+
+  public static enum WindowFrameUnit {
+    ROW,
+    RANGE
+  }
+
+  public static enum WindowFrameStartBoundType {
+    UNBOUNDED_PRECEDING,
+    CURRENT_ROW,
+    PRECEDING
+  }
+
+  public static enum WindowFrameEndBoundType {
+    UNBOUNDED_FOLLOWING,
+    CURRENT_ROW,
+    FOLLOWING
+  }
+
+  public static class WindowFrame implements Cloneable {
+    @Expose private WindowFrameUnit unit;
+    @Expose private WindowStartBound startBound;
+    @Expose private WindowEndBound endBound;
+
+    public WindowFrame(WindowFrameUnit unit, WindowStartBound startBound) {
+      this.unit = unit;
+      this.startBound = startBound;
+    }
+
+    public WindowFrame(WindowFrameUnit unit, WindowStartBound startBound, WindowEndBound endBound) {
+      this(unit, startBound);
+      this.endBound = endBound;
+    }
+
+    public WindowStartBound getStartBound() {
+      return startBound;
+    }
+
+    public boolean hasEndBound() {
+      return endBound != null;
+    }
+
+    public WindowEndBound getEndBound() {
+      return endBound;
+    }
+
+    @Override
+    public Object clone() throws CloneNotSupportedException {
+      WindowFrame frame = (WindowFrame) super.clone();
+      frame.unit = unit;
+      frame.startBound = (WindowStartBound) startBound.clone();
+      frame.endBound = (WindowEndBound) endBound.clone();
+      return frame;
+    }
+  }
+
+  public static class WindowStartBound implements Cloneable {
+    @Expose private WindowFrameStartBoundType boundType;
+    @Expose private Expr number;
+
+    public WindowStartBound(WindowFrameStartBoundType type) {
+      this.boundType = type;
+    }
+
+    public WindowFrameStartBoundType getBoundType() {
+      return boundType;
+    }
+
+    public boolean hasNumber() {
+      return this.number != null;
+    }
+
+    public void setNumber(Expr number) {
+      this.number = number;
+    }
+
+    public Expr getNumber() {
+      return number;
+    }
+
+    @Override
+    public Object clone() throws CloneNotSupportedException {
+      WindowStartBound start = (WindowStartBound) super.clone();
+      start.boundType = boundType;
+      start.number = (Expr) number.clone();
+      return start;
+    }
+  }
+
+  public static class WindowEndBound implements Cloneable {
+    @Expose private WindowFrameEndBoundType boundType;
+    @Expose private Expr number;
+
+    public WindowEndBound(WindowFrameEndBoundType type) {
+      this.boundType = type;
+    }
+
+    public WindowFrameEndBoundType getBoundType() {
+      return boundType;
+    }
+
+    public boolean hasNumber() {
+      return this.number != null;
+    }
+
+    public void setNumber(Expr number) {
+      this.number = number;
+    }
+
+    public Expr getNumber() {
+      return number;
+    }
+
+    @Override
+    public Object clone() throws CloneNotSupportedException {
+      WindowEndBound end = (WindowEndBound) super.clone();
+      end.boundType = boundType;
+      end.number = (Expr) number.clone();
+      return end;
+    }
+  }
+}
diff --git a/tajo-algebra/src/main/java/org/apache/tajo/algebra/WindowSpecExpr.java b/tajo-algebra/src/main/java/org/apache/tajo/algebra/WindowSpecExpr.java
deleted file mode 100644
index 37c2ff1..0000000
--- a/tajo-algebra/src/main/java/org/apache/tajo/algebra/WindowSpecExpr.java
+++ /dev/null
@@ -1,181 +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.algebra;
-
-import com.google.common.base.Objects;
-import org.apache.tajo.util.TUtil;
-
-public class WindowSpecExpr extends Expr {
-  private String windowName;
-  private Expr [] partitionKeys; // OVER (PARTITION BY ?,...,?)
-  private Sort.SortSpec [] sortSpecs; // OVER (... ORDER BY ?,...,?)
-  private WindowFrame windowFrame;
-
-  public WindowSpecExpr() {
-    super(OpType.WindowSpec);
-  }
-
-  public boolean hasWindowName() {
-    return windowName != null;
-  }
-
-  public void setWindowName(String windowName) {
-    this.windowName = windowName;
-  }
-
-  public String getWindowName() {
-    return this.windowName;
-  }
-
-  public boolean hasPartitionBy() {
-    return this.partitionKeys != null;
-  }
-
-  public void setPartitionKeys(Expr[] partitionKeys) {
-    this.partitionKeys = partitionKeys;
-  }
-
-  public Expr [] getPartitionKeys() {
-    return this.partitionKeys;
-  }
-
-  public boolean hasOrderBy() {
-    return this.sortSpecs != null;
-  }
-
-  public void setSortSpecs(Sort.SortSpec [] sortSpecs) {
-    this.sortSpecs = sortSpecs;
-  }
-
-  public Sort.SortSpec [] getSortSpecs() {
-    return this.sortSpecs;
-  }
-
-  public boolean hasWindowFrame() {
-    return windowFrame != null;
-  }
-
-  public void setWindowFrame(WindowFrame frame) {
-    this.windowFrame = frame;
-  }
-
-  public WindowFrame getWindowFrame() {
-    return windowFrame;
-  }
-
-  @Override
-  public int hashCode() {
-    return Objects.hashCode(windowName, partitionKeys, sortSpecs);
-  }
-
-  @Override
-  boolean equalsTo(Expr expr) {
-    WindowSpecExpr another = (WindowSpecExpr) expr;
-    return TUtil.checkEquals(windowName, another.windowName) &&
-        TUtil.checkEquals(partitionKeys, another.partitionKeys) &&
-        TUtil.checkEquals(sortSpecs, another.sortSpecs);
-  }
-
-  public static enum WindowFrameUnit {
-    ROW,
-    RANGE
-  }
-
-  public static enum WindowFrameStartBoundType {
-    UNBOUNDED_PRECEDING,
-    CURRENT_ROW,
-    PRECEDING
-  }
-
-  public static enum WindowFrameEndBoundType {
-    UNBOUNDED_FOLLOWING,
-    CURRENT_ROW,
-    FOLLOWING
-  }
-
-  public static class WindowFrame {
-    WindowFrameUnit unit;
-    private WindowStartBound startBound;
-    private WindowEndBound endBound;
-
-    public WindowFrame(WindowFrameUnit unit, WindowStartBound startBound) {
-      this.unit = unit;
-      this.startBound = startBound;
-    }
-
-    public WindowFrame(WindowFrameUnit unit, WindowStartBound startBound, WindowEndBound endBound) {
-      this(unit, startBound);
-      this.endBound = endBound;
-    }
-
-    public WindowStartBound getStartBound() {
-      return startBound;
-    }
-
-    public boolean hasEndBound() {
-      return endBound != null;
-    }
-
-    public WindowEndBound getEndBound() {
-      return endBound;
-    }
-  }
-
-  public static class WindowStartBound {
-    private WindowFrameStartBoundType boundType;
-    private Expr number;
-
-    public WindowStartBound(WindowFrameStartBoundType type) {
-      this.boundType = type;
-    }
-
-    public WindowFrameStartBoundType getBoundType() {
-      return boundType;
-    }
-
-    public void setNumber(Expr number) {
-      this.number = number;
-    }
-
-    public Expr getNumber() {
-      return number;
-    }
-  }
-
-  public static class WindowEndBound {
-    private WindowFrameEndBoundType boundType;
-    private Expr number;
-
-    public WindowEndBound(WindowFrameEndBoundType type) {
-      this.boundType = type;
-    }
-
-    public WindowFrameEndBoundType getBoundType() {
-      return boundType;
-    }
-
-    public Expr setNumber(Expr number) {
-      return number;
-    }
-
-    public Expr getNumber() {
-      return number;
-    }
-  }
-}
diff --git a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/exception/NoSuchFunctionException.java b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/exception/NoSuchFunctionException.java
index 78df08f..d109470 100644
--- a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/exception/NoSuchFunctionException.java
+++ b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/exception/NoSuchFunctionException.java
@@ -26,6 +26,10 @@
 public class NoSuchFunctionException extends RuntimeException {
 	private static final long serialVersionUID = 5062193018697228028L;
 
+  public NoSuchFunctionException(String message) {
+    super(message);
+  }
+
   public NoSuchFunctionException(String funcName, TajoDataTypes.DataType [] parameters) {
     super("function " + CatalogUtil.getCanonicalSignature(funcName, parameters) + " does not exist");
   }
diff --git a/tajo-catalog/tajo-catalog-common/src/main/proto/CatalogProtos.proto b/tajo-catalog/tajo-catalog-common/src/main/proto/CatalogProtos.proto
index e70ed2b..367d0b8 100644
--- a/tajo-catalog/tajo-catalog-common/src/main/proto/CatalogProtos.proto
+++ b/tajo-catalog/tajo-catalog-common/src/main/proto/CatalogProtos.proto
@@ -119,9 +119,11 @@
   GENERAL = 0;
   AGGREGATION = 1;
   DISTINCT_AGGREGATION = 2;
-  UDF = 3;
-  UDA = 4;
-  DISTINCT_UDA = 5;
+  WINDOW = 3;
+  UDF = 4;
+  UDA = 5;
+  DISTINCT_UDA = 6;
+  WINDOW_UDA = 7;
 }
 
 message FunctionDescProto {
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/eval/AggregationFunctionCallEval.java b/tajo-core/src/main/java/org/apache/tajo/engine/eval/AggregationFunctionCallEval.java
index 10eadce..ab18aa9 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/eval/AggregationFunctionCallEval.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/eval/AggregationFunctionCallEval.java
@@ -33,6 +33,11 @@
   @Expose boolean firstPhase = false;
   private Tuple params;
 
+  protected AggregationFunctionCallEval(EvalType type, FunctionDesc desc, AggFunction instance, EvalNode[] givenArgs) {
+    super(type, desc, givenArgs);
+    this.instance = instance;
+  }
+
   public AggregationFunctionCallEval(FunctionDesc desc, AggFunction instance, EvalNode[] givenArgs) {
     super(EvalType.AGG_FUNCTION, desc, givenArgs);
     this.instance = instance;
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/eval/AlgebraicUtil.java b/tajo-core/src/main/java/org/apache/tajo/engine/eval/AlgebraicUtil.java
index d993b27..a72e2a8 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/eval/AlgebraicUtil.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/eval/AlgebraicUtil.java
@@ -37,7 +37,7 @@
    * @return Transposed expression
    */
   public static EvalNode transpose(EvalNode evalNode, Column target) {
-    BinaryEval commutated = null;
+    BinaryEval commutated;
 
     if (evalNode instanceof BinaryEval) { // if it is binary
       BinaryEval binaryEval = (BinaryEval) evalNode;
@@ -168,19 +168,22 @@
       return unaryEval;
     }
 
-    public EvalNode visitFuncCall(Object context, GeneralFunctionEval evalNode, Stack<EvalNode> stack) {
-      boolean constant = true;
+    @Override
+    public EvalNode visitFuncCall(Object context, FunctionEval evalNode, Stack<EvalNode> stack) {
+      boolean constantOfAllDescendents = true;
 
       if ("sleep".equals(evalNode.funcDesc.getSignature())) {
-        constant = false;
+        constantOfAllDescendents = false;
       } else {
-        for (EvalNode arg : evalNode.getArgs()) {
-          arg = visit(context, arg, stack);
-          constant &= (arg.getType() == EvalType.CONST);
+        if (evalNode.getArgs() != null) {
+          for (EvalNode arg : evalNode.getArgs()) {
+            arg = visit(context, arg, stack);
+            constantOfAllDescendents &= (arg.getType() == EvalType.CONST);
+          }
         }
       }
 
-      if (constant) {
+      if (constantOfAllDescendents && evalNode.getType() == EvalType.FUNCTION) {
         return new ConstEval(evalNode.eval(null, null));
       } else {
         return evalNode;
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/eval/BasicEvalNodeVisitor.java b/tajo-core/src/main/java/org/apache/tajo/engine/eval/BasicEvalNodeVisitor.java
index 6e83c70..3b94cc9 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/eval/BasicEvalNodeVisitor.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/eval/BasicEvalNodeVisitor.java
@@ -125,6 +125,9 @@
       case AGG_FUNCTION:
         result = visitAggrFuncCall(context, (AggregationFunctionCallEval) evalNode, stack);
         break;
+      case WINDOW_FUNCTION:
+        result = visitWindowFunc(context, (WindowFunctionEval) evalNode, stack);
+      break;
 
       case SIGNED:
         result = visitSigned(context, (SignedEval) evalNode, stack);
@@ -326,6 +329,11 @@
   }
 
   @Override
+  public RESULT visitWindowFunc(CONTEXT context, WindowFunctionEval evalNode, Stack<EvalNode> stack) {
+    return visitDefaultFunctionEval(context, evalNode, stack);
+  }
+
+  @Override
   public RESULT visitSigned(CONTEXT context, SignedEval signedEval, Stack<EvalNode> stack) {
     return visitDefaultUnaryEval(context, signedEval, stack);
   }
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/eval/BinaryEval.java b/tajo-core/src/main/java/org/apache/tajo/engine/eval/BinaryEval.java
index bafce91..70d6bb1 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/eval/BinaryEval.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/eval/BinaryEval.java
@@ -26,6 +26,7 @@
 import org.apache.tajo.datum.Datum;
 import org.apache.tajo.datum.DatumFactory;
 import org.apache.tajo.datum.NullDatum;
+import org.apache.tajo.engine.utils.DataTypeUtil;
 import org.apache.tajo.exception.InvalidOperationException;
 import org.apache.tajo.storage.Tuple;
 
@@ -60,7 +61,7 @@
             type == EvalType.MULTIPLY ||
             type == EvalType.DIVIDE ||
             type == EvalType.MODULAR ) {
-      this.returnType = determineType(left.getValueType(), right.getValueType());
+      this.returnType = DataTypeUtil.determineType(left.getValueType(), right.getValueType());
 
     } else if (type == EvalType.CONCATENATE) {
       this.returnType = CatalogUtil.newSimpleDataType(Type.TEXT);
@@ -97,102 +98,6 @@
     }
   }
 
-  /**
-   * This is verified by ExprsVerifier.checkArithmeticOperand().
-   */
-  private DataType determineType(DataType left, DataType right) throws InvalidEvalException {
-    switch (left.getType()) {
-
-    case INT1:
-    case INT2:
-    case INT4: {
-      switch(right.getType()) {
-      case INT1:
-      case INT2:
-      case INT4: return CatalogUtil.newSimpleDataType(Type.INT4);
-      case INT8: return CatalogUtil.newSimpleDataType(Type.INT8);
-      case FLOAT4: return CatalogUtil.newSimpleDataType(Type.FLOAT4);
-      case FLOAT8: return CatalogUtil.newSimpleDataType(Type.FLOAT8);
-      case DATE: return CatalogUtil.newSimpleDataType(Type.DATE);
-      case INTERVAL: return CatalogUtil.newSimpleDataType(Type.INTERVAL);
-      }
-    }
-
-    case INT8: {
-      switch(right.getType()) {
-      case INT1:
-      case INT2:
-      case INT4:
-      case INT8: return CatalogUtil.newSimpleDataType(Type.INT8);
-      case FLOAT4: return CatalogUtil.newSimpleDataType(Type.FLOAT4);
-      case FLOAT8: return CatalogUtil.newSimpleDataType(Type.FLOAT8);
-      case DATE: return CatalogUtil.newSimpleDataType(Type.DATE);
-      case INTERVAL: return CatalogUtil.newSimpleDataType(Type.INTERVAL);
-      }
-    }
-
-    case FLOAT4: {
-      switch(right.getType()) {
-      case INT1:
-      case INT2:
-      case INT4: return CatalogUtil.newSimpleDataType(Type.FLOAT4);
-      case INT8: return CatalogUtil.newSimpleDataType(Type.FLOAT4);
-      case FLOAT4: return CatalogUtil.newSimpleDataType(Type.FLOAT4);
-      case FLOAT8: return CatalogUtil.newSimpleDataType(Type.FLOAT8);
-      case INTERVAL: return CatalogUtil.newSimpleDataType(Type.INTERVAL);
-      }
-    }
-
-    case FLOAT8: {
-      switch(right.getType()) {
-      case INT1:
-      case INT2:
-      case INT4:
-      case INT8:
-      case FLOAT4:
-      case FLOAT8: return CatalogUtil.newSimpleDataType(Type.FLOAT8);
-      case INTERVAL: return CatalogUtil.newSimpleDataType(Type.INTERVAL);
-      }
-    }
-
-    case DATE: {
-      switch(right.getType()) {
-      case INT2:
-      case INT4:
-      case INT8: return CatalogUtil.newSimpleDataType(Type.DATE);
-      case INTERVAL:
-      case TIME: return CatalogUtil.newSimpleDataType(Type.TIMESTAMP);
-      case DATE: return CatalogUtil.newSimpleDataType(Type.INT4);
-      }
-    }
-
-    case TIME: {
-      switch(right.getType()) {
-      case INTERVAL: return CatalogUtil.newSimpleDataType(Type.TIME);
-      case TIME: return CatalogUtil.newSimpleDataType(Type.INTERVAL);
-      case DATE: return CatalogUtil.newSimpleDataType(Type.INT4);
-      }
-    }
-
-    case TIMESTAMP: {
-      switch (right.getType()) {
-      case INTERVAL: return CatalogUtil.newSimpleDataType(Type.TIMESTAMP);
-      case TIMESTAMP: return CatalogUtil.newSimpleDataType(Type.INTERVAL);
-      }
-    }
-
-    case INTERVAL: {
-      switch (right.getType()) {
-      case INTERVAL:
-      case FLOAT4:
-      case FLOAT8: return CatalogUtil.newSimpleDataType(Type.INTERVAL);
-      }
-    }
-
-    default: return left;
-    }
-  }
-
   @Override
   public Datum eval(Schema schema, Tuple tuple) {
     Datum lhs = leftExpr.eval(schema, tuple);
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/eval/EvalNodeVisitor2.java b/tajo-core/src/main/java/org/apache/tajo/engine/eval/EvalNodeVisitor2.java
index 024a988..e85984e 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/eval/EvalNodeVisitor2.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/eval/EvalNodeVisitor2.java
@@ -64,6 +64,7 @@
   // Functions
   RESULT visitFuncCall(CONTEXT context, GeneralFunctionEval evalNode, Stack<EvalNode> stack);
   RESULT visitAggrFuncCall(CONTEXT context, AggregationFunctionCallEval evalNode, Stack<EvalNode> stack);
+  RESULT visitWindowFunc(CONTEXT context, WindowFunctionEval evalNode, Stack<EvalNode> stack);
 
   RESULT visitSigned(CONTEXT context, SignedEval signedEval, Stack<EvalNode> stack);
 
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/eval/EvalType.java b/tajo-core/src/main/java/org/apache/tajo/engine/eval/EvalType.java
index 14367f4..549f8d0 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/eval/EvalType.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/eval/EvalType.java
@@ -44,6 +44,7 @@
   BIT_XOR(BinaryEval.class, "|"),
 
   // Function
+  WINDOW_FUNCTION(WindowFunctionEval.class),
   AGG_FUNCTION(AggregationFunctionCallEval.class),
   FUNCTION(GeneralFunctionEval.class),
 
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/eval/FunctionEval.java b/tajo-core/src/main/java/org/apache/tajo/engine/eval/FunctionEval.java
index 6781c34..0cc8d98 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/eval/FunctionEval.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/eval/FunctionEval.java
@@ -19,6 +19,7 @@
 package org.apache.tajo.engine.eval;
 
 import com.google.common.base.Objects;
+import com.google.common.base.Preconditions;
 import com.google.gson.annotations.Expose;
 import org.apache.tajo.catalog.FunctionDesc;
 import org.apache.tajo.catalog.Schema;
@@ -42,7 +43,8 @@
 	public FunctionEval(EvalType type, FunctionDesc funcDesc, EvalNode[] argEvals) {
 		super(type);
 		this.funcDesc = funcDesc;
-		this.argEvals = argEvals;
+    Preconditions.checkArgument(argEvals != null, "argEvals cannot be null");
+    this.argEvals = argEvals;
 	}
 
   public ParamType [] getParamType() {
@@ -85,6 +87,10 @@
 		return funcDesc.getSignature();
 	}
 
+  public FunctionDesc getFuncDesc() {
+    return this.funcDesc;
+  }
+
   @Override
 	public String toString() {
 		StringBuilder sb = new StringBuilder();
@@ -119,26 +125,32 @@
   public Object clone() throws CloneNotSupportedException {
     FunctionEval eval = (FunctionEval) super.clone();
     eval.funcDesc = (FunctionDesc) funcDesc.clone();
-    eval.argEvals = new EvalNode[argEvals.length];
-    for (int i = 0; i < argEvals.length; i++) {
-      eval.argEvals[i] = (EvalNode) argEvals[i].clone();
-    }    
+    if (argEvals != null) {
+      eval.argEvals = new EvalNode[argEvals.length];
+      for (int i = 0; i < argEvals.length; i++) {
+        eval.argEvals[i] = (EvalNode) argEvals[i].clone();
+      }
+    }
     return eval;
   }
 	
 	@Override
   public void preOrder(EvalNodeVisitor visitor) {
-    for (EvalNode eval : argEvals) {
-      eval.postOrder(visitor);
+    if (argEvals != null) {
+      for (EvalNode eval : argEvals) {
+        eval.postOrder(visitor);
+      }
     }
     visitor.visit(this);
   }
 	
 	@Override
 	public void postOrder(EvalNodeVisitor visitor) {
-	  for (EvalNode eval : argEvals) {
-	    eval.postOrder(visitor);
-	  }
+    if (argEvals != null) {
+      for (EvalNode eval : argEvals) {
+        eval.postOrder(visitor);
+      }
+    }
 	  visitor.visit(this);
 	}
 }
\ No newline at end of file
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/eval/SimpleEvalNodeVisitor.java b/tajo-core/src/main/java/org/apache/tajo/engine/eval/SimpleEvalNodeVisitor.java
index e4503e2..93f1f74 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/eval/SimpleEvalNodeVisitor.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/eval/SimpleEvalNodeVisitor.java
@@ -66,10 +66,13 @@
 
       // Functions
       case FUNCTION:
-        result = visitFuncCall(context, (GeneralFunctionEval) evalNode, stack);
+        result = visitFuncCall(context, (FunctionEval) evalNode, stack);
         break;
       case AGG_FUNCTION:
-        result = visitAggrFuncCall(context, (AggregationFunctionCallEval) evalNode, stack);
+        result = visitFuncCall(context, (FunctionEval) evalNode, stack);
+        break;
+      case WINDOW_FUNCTION:
+        result = visitFuncCall(context, (FunctionEval) evalNode, stack);
         break;
 
       default:
@@ -163,11 +166,7 @@
   // Functions
   ///////////////////////////////////////////////////////////////////////////////////////////////
 
-  public EvalNode visitFuncCall(CONTEXT context, GeneralFunctionEval evalNode, Stack<EvalNode> stack) {
-    return visitDefaultFunctionEval(context, stack, evalNode);
-  }
-
-  public EvalNode visitAggrFuncCall(CONTEXT context, AggregationFunctionCallEval evalNode, Stack<EvalNode> stack) {
+  public EvalNode visitFuncCall(CONTEXT context, FunctionEval evalNode, Stack<EvalNode> stack) {
     return visitDefaultFunctionEval(context, stack, evalNode);
   }
 }
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/eval/WindowFunctionEval.java b/tajo-core/src/main/java/org/apache/tajo/engine/eval/WindowFunctionEval.java
new file mode 100644
index 0000000..fb4eede
--- /dev/null
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/eval/WindowFunctionEval.java
@@ -0,0 +1,117 @@
+/**
+ * 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.eval;
+
+import com.google.gson.annotations.Expose;
+import org.apache.tajo.catalog.FunctionDesc;
+import org.apache.tajo.catalog.Schema;
+import org.apache.tajo.catalog.SortSpec;
+import org.apache.tajo.common.TajoDataTypes.DataType;
+import org.apache.tajo.datum.Datum;
+import org.apache.tajo.engine.function.AggFunction;
+import org.apache.tajo.engine.function.FunctionContext;
+import org.apache.tajo.engine.planner.logical.WindowSpec;
+import org.apache.tajo.storage.Tuple;
+import org.apache.tajo.storage.VTuple;
+import org.apache.tajo.util.TUtil;
+
+public class WindowFunctionEval extends AggregationFunctionCallEval implements Cloneable {
+  @Expose private SortSpec [] sortSpecs;
+  @Expose WindowSpec.WindowFrame windowFrame;
+  private Tuple params;
+
+  public WindowFunctionEval(FunctionDesc desc, AggFunction instance, EvalNode[] givenArgs,
+                            WindowSpec.WindowFrame windowFrame) {
+    super(EvalType.WINDOW_FUNCTION, desc, instance, givenArgs);
+    this.windowFrame = windowFrame;
+  }
+
+  public boolean hasSortSpecs() {
+    return sortSpecs != null;
+  }
+
+  public void setSortSpecs(SortSpec [] sortSpecs) {
+    this.sortSpecs = sortSpecs;
+  }
+
+  public SortSpec [] getSortSpecs() {
+    return sortSpecs;
+  }
+
+  public WindowSpec.WindowFrame getWindowFrame() {
+    return windowFrame;
+  }
+
+  @Override
+  public Datum eval(Schema schema, Tuple tuple) {
+    throw new UnsupportedOperationException("Cannot execute eval() of aggregation function");
+  }
+
+  public void merge(FunctionContext context, Schema schema, Tuple tuple) {
+    if (params == null) {
+      this.params = new VTuple(argEvals.length);
+    }
+
+    if (argEvals != null) {
+      for (int i = 0; i < argEvals.length; i++) {
+        params.put(i, argEvals[i].eval(schema, tuple));
+      }
+    }
+
+    instance.eval(context, params);
+  }
+
+  public Datum terminate(FunctionContext context) {
+    return instance.terminate(context);
+  }
+
+  @Override
+  public DataType getValueType() {
+    return funcDesc.getReturnType();
+  }
+
+  @Override
+  public Object clone() throws CloneNotSupportedException {
+    WindowFunctionEval windowFunctionEval = (WindowFunctionEval) super.clone();
+    if (sortSpecs != null) {
+      windowFunctionEval.sortSpecs = new SortSpec[sortSpecs.length];
+      for (int i = 0; i < sortSpecs.length; i++) {
+        windowFunctionEval.sortSpecs[i] = (SortSpec) sortSpecs[i].clone();
+      }
+    }
+    return windowFunctionEval;
+  }
+
+  public String toString() {
+    StringBuilder sb = new StringBuilder();
+    if (argEvals != null) {
+      for(int i=0; i < argEvals.length; i++) {
+        sb.append(argEvals[i]);
+        if(i+1 < argEvals.length)
+          sb.append(",");
+      }
+    }
+    sb.append(funcDesc.getSignature()).append("(").append(isDistinct() ? " distinct" : "").append(sb)
+        .append(")");
+    if (hasSortSpecs()) {
+      sb.append("ORDER BY ").append(TUtil.arrayToString(sortSpecs));
+    }
+    return sb.toString();
+  }
+}
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/function/WindowAggFunc.java b/tajo-core/src/main/java/org/apache/tajo/engine/function/WindowAggFunc.java
new file mode 100644
index 0000000..164738a
--- /dev/null
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/function/WindowAggFunc.java
@@ -0,0 +1,62 @@
+/**
+ * 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.function;
+
+import org.apache.tajo.catalog.Column;
+import org.apache.tajo.catalog.json.CatalogGsonHelper;
+import org.apache.tajo.catalog.proto.CatalogProtos;
+import org.apache.tajo.common.TajoDataTypes.DataType;
+import org.apache.tajo.datum.Datum;
+import org.apache.tajo.exception.InvalidOperationException;
+import org.apache.tajo.storage.Tuple;
+
+public abstract class WindowAggFunc<T extends Datum> extends AggFunction<T> {
+
+  public WindowAggFunc(Column[] definedArgs) {
+    super(definedArgs);
+  }
+
+  public abstract FunctionContext newContext();
+
+  public abstract void eval(FunctionContext ctx, Tuple params);
+
+  public void merge(FunctionContext ctx, Tuple part) {
+    throw new InvalidOperationException("Window function does not support getPartialResult()");
+  }
+
+  public Datum getPartialResult(FunctionContext ctx) {
+    throw new InvalidOperationException("Window function does not support getPartialResult()");
+  }
+
+  public DataType getPartialResultType() {
+    throw new InvalidOperationException("Window function does not support getPartialResultType()");
+  }
+
+  public abstract T terminate(FunctionContext ctx);
+
+  @Override
+  public String toJson() {
+    return CatalogGsonHelper.toJson(this, WindowAggFunc.class);
+  }
+
+  @Override
+  public CatalogProtos.FunctionType getFunctionType() {
+    return CatalogProtos.FunctionType.WINDOW;
+  }
+}
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/function/window/Rank.java b/tajo-core/src/main/java/org/apache/tajo/engine/function/window/Rank.java
new file mode 100644
index 0000000..e461f9d
--- /dev/null
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/function/window/Rank.java
@@ -0,0 +1,90 @@
+/**
+ * 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.function.window;
+
+import org.apache.tajo.catalog.Column;
+import org.apache.tajo.catalog.proto.CatalogProtos;
+import org.apache.tajo.common.TajoDataTypes;
+import org.apache.tajo.datum.Datum;
+import org.apache.tajo.datum.DatumFactory;
+import org.apache.tajo.datum.Int8Datum;
+import org.apache.tajo.engine.function.FunctionContext;
+import org.apache.tajo.engine.function.WindowAggFunc;
+import org.apache.tajo.engine.function.annotation.Description;
+import org.apache.tajo.engine.function.annotation.ParamTypes;
+import org.apache.tajo.storage.Tuple;
+
+@Description(
+    functionName = "rank",
+    description = " The number of rows for "
+        + "which the supplied expressions are unique and non-NULL.",
+    example = "> SELECT rank() OVER (ORDER BY x) FROM ...;",
+    returnType = TajoDataTypes.Type.INT8,
+    paramTypes = {@ParamTypes(paramTypes = {})}
+)
+public final class Rank extends WindowAggFunc {
+
+  public Rank() {
+    super(new Column[] {
+        new Column("expr", TajoDataTypes.Type.ANY)
+    });
+  }
+
+  public static boolean checkIfDistinctValue(RankContext context, Tuple params) {
+    for (int i = 0; i < context.latest.length; i++) {
+      if (!context.latest[i].equalsTo(params.get(i)).isTrue()) {
+        return true;
+      }
+    }
+
+    return false;
+  }
+
+  @Override
+  public void eval(FunctionContext context, Tuple params) {
+    RankContext ctx = (RankContext) context;
+
+    if ((ctx.latest == null || checkIfDistinctValue(ctx, params))) {
+      ctx.rank = ctx.accumulatedCount;
+      ctx.latest = params.getValues().clone();
+    }
+    ctx.accumulatedCount++;
+  }
+
+  @Override
+  public Int8Datum terminate(FunctionContext ctx) {
+    return DatumFactory.createInt8(((RankContext) ctx).rank);
+  }
+
+  @Override
+  public FunctionContext newContext() {
+    return new RankContext();
+  }
+
+  private class RankContext implements FunctionContext {
+    long rank = 0;
+    long accumulatedCount = 1;
+    Datum [] latest = null;
+  }
+
+  @Override
+  public CatalogProtos.FunctionType getFunctionType() {
+    return CatalogProtos.FunctionType.WINDOW;
+  }
+}
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/function/window/RowNumber.java b/tajo-core/src/main/java/org/apache/tajo/engine/function/window/RowNumber.java
new file mode 100644
index 0000000..8b0943f
--- /dev/null
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/function/window/RowNumber.java
@@ -0,0 +1,67 @@
+/**
+ * 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.function.window;
+
+import org.apache.tajo.catalog.Column;
+import org.apache.tajo.common.TajoDataTypes.Type;
+import org.apache.tajo.datum.Datum;
+import org.apache.tajo.datum.DatumFactory;
+import org.apache.tajo.datum.Int8Datum;
+import org.apache.tajo.engine.function.FunctionContext;
+import org.apache.tajo.engine.function.WindowAggFunc;
+import org.apache.tajo.engine.function.annotation.Description;
+import org.apache.tajo.engine.function.annotation.ParamTypes;
+import org.apache.tajo.storage.Tuple;
+
+@Description(
+  functionName = "row_number",
+  description = "the total number of retrieved rows",
+  example = "> SELECT row_number() OVER ();",
+  returnType = Type.INT8,
+  paramTypes = {@ParamTypes(paramTypes = {})}
+)
+public class RowNumber extends WindowAggFunc<Datum> {
+
+  public RowNumber() {
+    super(NoArgs);
+  }
+
+  protected RowNumber(Column[] columns) {
+    super(columns);
+  }
+
+  @Override
+  public FunctionContext newContext() {
+    return new RowNumberContext();
+  }
+
+  @Override
+  public void eval(FunctionContext ctx, Tuple params) {
+    ((RowNumberContext) ctx).count++;
+  }
+
+  @Override
+  public Int8Datum terminate(FunctionContext ctx) {
+    return DatumFactory.createInt8(((RowNumberContext) ctx).count);
+  }
+
+  protected class RowNumberContext implements FunctionContext {
+    long count = 0;
+  }
+}
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/parser/SQLAnalyzer.java b/tajo-core/src/main/java/org/apache/tajo/engine/parser/SQLAnalyzer.java
index 59a9b46..580ec61 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/parser/SQLAnalyzer.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/parser/SQLAnalyzer.java
@@ -36,8 +36,8 @@
 
 import static org.apache.tajo.algebra.Aggregation.GroupElement;
 import static org.apache.tajo.algebra.CreateTable.*;
-import static org.apache.tajo.algebra.WindowSpecExpr.WindowFrameEndBoundType;
-import static org.apache.tajo.algebra.WindowSpecExpr.WindowFrameStartBoundType;
+import static org.apache.tajo.algebra.WindowSpec.WindowFrameEndBoundType;
+import static org.apache.tajo.algebra.WindowSpec.WindowFrameStartBoundType;
 import static org.apache.tajo.common.TajoDataTypes.Type;
 import static org.apache.tajo.engine.parser.SQLParser.*;
 
@@ -347,7 +347,7 @@
     if (checkIfExist(windowNameOrSpec.window_name())) {
       windowFunction.setWindowName(windowNameOrSpec.window_name().getText());
     } else {
-      windowFunction.setWindowSpec(visitWindow_specification(windowNameOrSpec.window_specification()));
+      windowFunction.setWindowSpec(buildWindowSpec(windowNameOrSpec.window_specification()));
     }
 
     return windowFunction;
@@ -360,14 +360,14 @@
     for (int i = 0; i < definitions.length; i++) {
       Window_definitionContext windowDefinitionContext = ctx.window_definition_list().window_definition(i);
       String windowName = windowDefinitionContext.window_name().identifier().getText();
-      WindowSpecExpr windowSpec = visitWindow_specification(windowDefinitionContext.window_specification());
+      WindowSpec windowSpec = buildWindowSpec(windowDefinitionContext.window_specification());
       definitions[i] = new Window.WindowDefinition(windowName, windowSpec);
     }
     return new Window(definitions);
   }
 
-  @Override public WindowSpecExpr visitWindow_specification(@NotNull SQLParser.Window_specificationContext ctx) {
-    WindowSpecExpr windowSpec = new WindowSpecExpr();
+  public WindowSpec buildWindowSpec(SQLParser.Window_specificationContext ctx) {
+    WindowSpec windowSpec = new WindowSpec();
     if (checkIfExist(ctx.window_specification_details())) {
       Window_specification_detailsContext windowSpecDetail = ctx.window_specification_details();
 
@@ -388,26 +388,26 @@
       if (checkIfExist(windowSpecDetail.window_frame_clause())) {
         Window_frame_clauseContext frameContext = windowSpecDetail.window_frame_clause();
 
-        WindowSpecExpr.WindowFrameUnit unit;
+        WindowSpec.WindowFrameUnit unit;
         // frame unit - there are only two cases: RANGE and ROW
         if (checkIfExist(frameContext.window_frame_units().RANGE())) {
-          unit = WindowSpecExpr.WindowFrameUnit.RANGE;
+          unit = WindowSpec.WindowFrameUnit.RANGE;
         } else {
-          unit = WindowSpecExpr.WindowFrameUnit.ROW;
+          unit = WindowSpec.WindowFrameUnit.ROW;
         }
 
-        WindowSpecExpr.WindowFrame windowFrame;
+        WindowSpec.WindowFrame windowFrame;
 
         if (checkIfExist(frameContext.window_frame_extent().window_frame_between())) { // when 'between' is given
           Window_frame_betweenContext between = frameContext.window_frame_extent().window_frame_between();
-          WindowSpecExpr.WindowStartBound startBound = buildWindowStartBound(between.window_frame_start_bound());
-          WindowSpecExpr.WindowEndBound endBound = buildWindowEndBound(between.window_frame_end_bound());
+          WindowSpec.WindowStartBound startBound = buildWindowStartBound(between.window_frame_start_bound());
+          WindowSpec.WindowEndBound endBound = buildWindowEndBound(between.window_frame_end_bound());
 
-          windowFrame = new WindowSpecExpr.WindowFrame(unit, startBound, endBound);
+          windowFrame = new WindowSpec.WindowFrame(unit, startBound, endBound);
         } else { // if there is only start bound
-          WindowSpecExpr.WindowStartBound startBound =
+          WindowSpec.WindowStartBound startBound =
               buildWindowStartBound(frameContext.window_frame_extent().window_frame_start_bound());
-          windowFrame = new WindowSpecExpr.WindowFrame(unit, startBound);
+          windowFrame = new WindowSpec.WindowFrame(unit, startBound);
         }
 
         windowSpec.setWindowFrame(windowFrame);
@@ -416,7 +416,7 @@
     return windowSpec;
   }
 
-  public WindowSpecExpr.WindowStartBound buildWindowStartBound(Window_frame_start_boundContext context) {
+  public WindowSpec.WindowStartBound buildWindowStartBound(Window_frame_start_boundContext context) {
     WindowFrameStartBoundType boundType = null;
     if (checkIfExist(context.UNBOUNDED())) {
       boundType = WindowFrameStartBoundType.UNBOUNDED_PRECEDING;
@@ -426,7 +426,7 @@
       boundType = WindowFrameStartBoundType.CURRENT_ROW;
     }
 
-    WindowSpecExpr.WindowStartBound bound = new WindowSpecExpr.WindowStartBound(boundType);
+    WindowSpec.WindowStartBound bound = new WindowSpec.WindowStartBound(boundType);
     if (boundType == WindowFrameStartBoundType.PRECEDING) {
       bound.setNumber(visitUnsigned_value_specification(context.unsigned_value_specification()));
     }
@@ -434,7 +434,7 @@
     return bound;
   }
 
-  public WindowSpecExpr.WindowEndBound buildWindowEndBound(Window_frame_end_boundContext context) {
+  public WindowSpec.WindowEndBound buildWindowEndBound(Window_frame_end_boundContext context) {
     WindowFrameEndBoundType boundType;
     if (checkIfExist(context.UNBOUNDED())) {
       boundType = WindowFrameEndBoundType.UNBOUNDED_FOLLOWING;
@@ -444,7 +444,7 @@
       boundType = WindowFrameEndBoundType.CURRENT_ROW;
     }
 
-    WindowSpecExpr.WindowEndBound endBound = new WindowSpecExpr.WindowEndBound(boundType);
+    WindowSpec.WindowEndBound endBound = new WindowSpec.WindowEndBound(boundType);
     if (boundType == WindowFrameEndBoundType.FOLLOWING) {
       endBound.setNumber(visitUnsigned_value_specification(context.unsigned_value_specification()));
     }
@@ -1275,7 +1275,7 @@
       return new ListPartition(buildColumnReferenceList(ctx.list_partitions().column_reference_list()), specifiers);
 
     } else if (checkIfExist(ctx.column_partitions())) { // For Column Partition (Hive Style)
-      return new CreateTable.ColumnPartition(getDefinitions(ctx.column_partitions().table_elements()), true);
+      return new CreateTable.ColumnPartition(getDefinitions(ctx.column_partitions().table_elements()));
     } else {
       throw new SQLSyntaxError("Invalid Partition Type: " + ctx.toStringTree());
     }
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/AlgebraVisitor.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/AlgebraVisitor.java
index 5811d36..8c3e606 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/AlgebraVisitor.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/AlgebraVisitor.java
@@ -91,13 +91,14 @@
   RESULT visitSign(CONTEXT ctx, Stack<Expr> stack, SignedExpr expr) throws PlanningException;
   RESULT visitColumnReference(CONTEXT ctx, Stack<Expr> stack, ColumnReferenceExpr expr) throws PlanningException;
   RESULT visitTargetExpr(CONTEXT ctx, Stack<Expr> stack, NamedExpr expr) throws PlanningException;
-  RESULT visitFunction(CONTEXT ctx, Stack<Expr> stack, FunctionExpr expr) throws PlanningException;
   RESULT visitQualifiedAsterisk(CONTEXT ctx, Stack<Expr> stack, QualifiedAsteriskExpr expr) throws PlanningException;
 
-  // set functions
-  RESULT visitCountRowsFunction(CONTEXT ctx, Stack<Expr> stack, CountRowsFunctionExpr expr) throws PlanningException;
+  // functions
+  RESULT visitFunction(CONTEXT ctx, Stack<Expr> stack, FunctionExpr expr) throws PlanningException;
   RESULT visitGeneralSetFunction(CONTEXT ctx, Stack<Expr> stack, GeneralSetFunctionExpr expr)
       throws PlanningException;
+  RESULT visitCountRowsFunction(CONTEXT ctx, Stack<Expr> stack, CountRowsFunctionExpr expr) throws PlanningException;
+  RESULT visitWindowFunction(CONTEXT ctx, Stack<Expr> stack, WindowFunctionExpr expr) throws PlanningException;
 
   // Literal
   RESULT visitCastExpr(CONTEXT ctx, Stack<Expr> stack, CastExpr expr) throws PlanningException;
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
index 907042a..24ff2e4 100644
--- 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
@@ -230,6 +230,10 @@
     case GeneralSetFunction:
       current = visitGeneralSetFunction(ctx, stack, (GeneralSetFunctionExpr) expr);
       break;
+    case WindowFunction:
+      current = visitWindowFunction(ctx, stack, (WindowFunctionExpr) expr);
+      break;
+
 
     case DataType:
       current = visitDataType(ctx, stack, (DataTypeExpr) expr);
@@ -703,6 +707,39 @@
     return result;
   }
 
+  @Override
+  public RESULT visitWindowFunction(CONTEXT ctx, Stack<Expr> stack, WindowFunctionExpr expr) throws PlanningException {
+    stack.push(expr);
+    RESULT result = null;
+    for (Expr param : expr.getParams()) {
+      result = visit(ctx, stack, param);
+    }
+
+    WindowSpec windowSpec = expr.getWindowSpec();
+
+    if (windowSpec.hasPartitionBy()) {
+      for (Expr partitionKey : windowSpec.getPartitionKeys()) {
+        visit(ctx, stack, partitionKey);
+      }
+    }
+    if (windowSpec.hasOrderBy()) {
+      for (Sort.SortSpec sortKey : windowSpec.getSortSpecs()) {
+        visit(ctx, stack, sortKey.getKey());
+      }
+    }
+    if (windowSpec.hasWindowFrame()) {
+      if (windowSpec.getWindowFrame().getStartBound().hasNumber()) {
+        visit(ctx, stack, windowSpec.getWindowFrame().getStartBound().getNumber());
+      }
+      if (windowSpec.getWindowFrame().getEndBound().hasNumber()) {
+        visit(ctx, stack, windowSpec.getWindowFrame().getEndBound().getNumber());
+      }
+    }
+
+    stack.pop();
+    return result;
+  }
+
   ///////////////////////////////////////////////////////////////////////////////////////////////////////////
   // Literal Section
   ///////////////////////////////////////////////////////////////////////////////////////////////////////////
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
index 2112615..a4e90b4 100644
--- 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
@@ -74,6 +74,9 @@
       case GROUP_BY:
         current = visitGroupBy(context, plan, block, (GroupbyNode) node, stack);
         break;
+      case WINDOW_AGG:
+        current = visitWindowAgg(context, plan, block, (WindowAggNode) node, stack);
+        break;
       case DISTINCT_GROUP_BY:
         current = visitDistinct(context, plan, block, (DistinctGroupbyNode) node, stack);
         break;
@@ -191,6 +194,14 @@
   }
 
   @Override
+  public RESULT visitWindowAgg(CONTEXT context, LogicalPlan plan, LogicalPlan.QueryBlock block, WindowAggNode node,
+                               Stack<LogicalNode> stack) throws PlanningException {
+    stack.push(node);
+    RESULT result = visit(context, plan, block, node.getChild(), stack);
+    stack.pop();
+    return result;
+  }
+
   public RESULT visitDistinct(CONTEXT context, LogicalPlan plan, LogicalPlan.QueryBlock block, DistinctGroupbyNode node,
                              Stack<LogicalNode> stack) throws PlanningException {
     stack.push(node);
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
index ad9bdf1..a7e5375 100644
--- 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
@@ -109,6 +109,11 @@
   }
 
   @Override
+  public LogicalNode visitWindowAgg(Context context, LogicalPlan plan, LogicalPlan.QueryBlock block, WindowAggNode node,
+                                    Stack<LogicalNode> stack) throws PlanningException {
+    return visitUnaryNode(context, plan, block, node, stack);
+  }
+
   public LogicalNode visitDistinct(Context context, LogicalPlan plan, LogicalPlan.QueryBlock block, DistinctGroupbyNode node,
                                   Stack<LogicalNode> stack) throws PlanningException {
     return visitUnaryNode(context, plan, block, node, stack);
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
index e143823..2c386b2 100644
--- 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
@@ -19,13 +19,14 @@
 package org.apache.tajo.engine.planner;
 
 import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
 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;
@@ -38,10 +39,17 @@
 import org.apache.tajo.util.datetime.TimeMeta;
 
 import java.util.Map;
+import java.util.Set;
 import java.util.Stack;
 
+import static org.apache.tajo.algebra.WindowSpec.WindowFrameEndBoundType;
+import static org.apache.tajo.algebra.WindowSpec.WindowFrameStartBoundType;
+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.engine.planner.logical.WindowSpec.WindowEndBound;
+import static org.apache.tajo.engine.planner.logical.WindowSpec.WindowFrame;
+import static org.apache.tajo.engine.planner.logical.WindowSpec.WindowStartBound;
 
 /**
  * <code>ExprAnnotator</code> makes an annotated expression called <code>EvalNode</code> from an
@@ -122,7 +130,7 @@
    * @return The widest DataType
    * @throws PlanningException when types are not compatible, it throws the exception.
    */
-  static DataType getWidestType(DataType...types) throws PlanningException {
+  public static DataType getWidestType(DataType...types) throws PlanningException {
     DataType widest = types[0];
     for (int i = 1; i < types.length; i++) {
 
@@ -541,6 +549,10 @@
     throw new PlanningException("ExprAnnotator cannot take NamedExpr");
   }
 
+  ///////////////////////////////////////////////////////////////////////////////////////////////////////////
+  // Functions and General Set Functions Section
+  ///////////////////////////////////////////////////////////////////////////////////////////////////////////
+
   @Override
   public EvalNode visitFunction(Context ctx, Stack<Expr> stack, FunctionExpr expr) throws PlanningException {
     stack.push(expr); // <--- Push
@@ -589,18 +601,18 @@
 
 
     try {
-      CatalogProtos.FunctionType functionType = funcDesc.getFuncType();
-      if (functionType == CatalogProtos.FunctionType.GENERAL
-          || functionType == CatalogProtos.FunctionType.UDF) {
+      FunctionType functionType = funcDesc.getFuncType();
+      if (functionType == FunctionType.GENERAL
+          || functionType == FunctionType.UDF) {
         return new GeneralFunctionEval(funcDesc, (GeneralFunction) funcDesc.newInstance(), givenArgs);
-      } else if (functionType == CatalogProtos.FunctionType.AGGREGATION
-          || functionType == CatalogProtos.FunctionType.UDA) {
+      } else if (functionType == FunctionType.AGGREGATION
+          || functionType == 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) {
+      } else if (functionType == FunctionType.DISTINCT_AGGREGATION
+          || functionType == FunctionType.DISTINCT_UDA) {
         throw new PlanningException("Unsupported function: " + funcDesc.toString());
       } else {
         throw new PlanningException("Unsupported Function Type: " + functionType.name());
@@ -610,14 +622,10 @@
     }
   }
 
-  ///////////////////////////////////////////////////////////////////////////////////////////////////////////
-  // General Set Section
-  ///////////////////////////////////////////////////////////////////////////////////////////////////////////
-
   @Override
   public EvalNode visitCountRowsFunction(Context ctx, Stack<Expr> stack, CountRowsFunctionExpr expr)
       throws PlanningException {
-    FunctionDesc countRows = catalog.getFunction("count", CatalogProtos.FunctionType.AGGREGATION,
+    FunctionDesc countRows = catalog.getFunction("count", FunctionType.AGGREGATION,
         new DataType[] {});
     if (countRows == null) {
       throw new NoSuchFunctionException(expr.getSignature(), new DataType[]{});
@@ -641,8 +649,8 @@
     EvalNode[] givenArgs = new EvalNode[params.length];
     DataType[] paramTypes = new DataType[params.length];
 
-    CatalogProtos.FunctionType functionType = setFunction.isDistinct() ?
-        CatalogProtos.FunctionType.DISTINCT_AGGREGATION : CatalogProtos.FunctionType.AGGREGATION;
+    FunctionType functionType = setFunction.isDistinct() ?
+        FunctionType.DISTINCT_AGGREGATION : FunctionType.AGGREGATION;
     givenArgs[0] = visit(ctx, stack, params[0]);
     if (setFunction.getSignature().equalsIgnoreCase("count")) {
       paramTypes[0] = CatalogUtil.newSimpleDataType(Type.ANY);
@@ -666,6 +674,91 @@
     }
   }
 
+  public static final Set<String> WINDOW_FUNCTIONS =
+      Sets.newHashSet("row_number", "rank", "dense_rank", "percent_rank", "cume_dist");
+
+  public EvalNode visitWindowFunction(Context ctx, Stack<Expr> stack, WindowFunctionExpr windowFunc)
+      throws PlanningException {
+
+    WindowSpec windowSpec = windowFunc.getWindowSpec();
+
+    Expr key;
+    if (windowSpec.hasPartitionBy()) {
+      for (int i = 0; i < windowSpec.getPartitionKeys().length; i++) {
+        key = windowSpec.getPartitionKeys()[i];
+        visit(ctx, stack, key);
+      }
+    }
+
+    EvalNode [] sortKeys = null;
+    if (windowSpec.hasOrderBy()) {
+      sortKeys = new EvalNode[windowSpec.getSortSpecs().length];
+      for (int i = 0; i < windowSpec.getSortSpecs().length; i++) {
+        key = windowSpec.getSortSpecs()[i].getKey();
+        sortKeys[i] = visit(ctx, stack, key);
+      }
+    }
+
+    String funcName = windowFunc.getSignature();
+    boolean distinct = windowFunc.isDistinct();
+    Expr[] params = windowFunc.getParams();
+    EvalNode[] givenArgs = new EvalNode[params.length];
+    TajoDataTypes.DataType[] paramTypes = new TajoDataTypes.DataType[params.length];
+    FunctionType functionType;
+
+    WindowFrame frame = null;
+
+    if (params.length > 0) {
+      givenArgs[0] = visit(ctx, stack, params[0]);
+      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);
+      } else {
+        paramTypes[0] = givenArgs[0].getValueType();
+      }
+    } else {
+      if (windowFunc.getSignature().equalsIgnoreCase("rank")) {
+        givenArgs = sortKeys != null ? sortKeys : new EvalNode[0];
+      }
+    }
+
+    if (frame == null) {
+      if (windowSpec.hasOrderBy()) {
+        frame = new WindowFrame(new WindowStartBound(WindowFrameStartBoundType.UNBOUNDED_PRECEDING),
+            new WindowEndBound(WindowFrameEndBoundType.CURRENT_ROW));
+      } else if (windowFunc.getSignature().equalsIgnoreCase("row_number")) {
+        frame = new WindowFrame(new WindowStartBound(WindowFrameStartBoundType.UNBOUNDED_PRECEDING),
+            new WindowEndBound(WindowFrameEndBoundType.UNBOUNDED_FOLLOWING));
+      } else {
+        frame = new WindowFrame();
+      }
+    }
+
+    // TODO - containFunction and getFunction should support the function type mask which provides ORing multiple types.
+    // the below checking against WINDOW_FUNCTIONS is a workaround code for the above problem.
+    if (WINDOW_FUNCTIONS.contains(funcName.toLowerCase())) {
+      if (distinct) {
+        throw new NoSuchFunctionException("row_number() does not support distinct keyword.");
+      }
+      functionType = FunctionType.WINDOW;
+    } else {
+      functionType = distinct ? FunctionType.DISTINCT_AGGREGATION : FunctionType.AGGREGATION;
+    }
+
+    if (!catalog.containFunction(windowFunc.getSignature(), functionType, paramTypes)) {
+      throw new NoSuchFunctionException(funcName, paramTypes);
+    }
+
+    FunctionDesc funcDesc = catalog.getFunction(funcName, functionType, paramTypes);
+
+    try {
+      return new WindowFunctionEval(funcDesc, (AggFunction) funcDesc.newInstance(), givenArgs, frame);
+    } catch (InternalException e) {
+      throw new PlanningException(e);
+    }
+  }
+
   ///////////////////////////////////////////////////////////////////////////////////////////////////////////
   // Literal Section
   ///////////////////////////////////////////////////////////////////////////////////////////////////////////
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/ExprNormalizer.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/ExprNormalizer.java
index 75b2b95..81bbd41 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/ExprNormalizer.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/ExprNormalizer.java
@@ -18,12 +18,16 @@
 
 package org.apache.tajo.engine.planner;
 
+import com.google.common.collect.Sets;
+import com.google.common.collect.Sets;
 import org.apache.tajo.algebra.*;
 import org.apache.tajo.catalog.CatalogUtil;
 import org.apache.tajo.engine.exception.NoSuchColumnException;
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Set;
+import java.util.Set;
 import java.util.Stack;
 
 /**
@@ -87,6 +91,8 @@
                    // function.
     List<NamedExpr> aggExprs = new ArrayList<NamedExpr>(); // aggregation functions
     List<NamedExpr> scalarExprs = new ArrayList<NamedExpr>(); // scalar expressions which can be referred
+    List<NamedExpr> windowAggExprs = new ArrayList<NamedExpr>(); // window expressions which can be referred
+    Set<WindowSpecReferences> windowSpecs = Sets.newLinkedHashSet();
 
     private ExprNormalizedResult(LogicalPlanner.PlanContext context, boolean tryBinaryCommonTermsElimination) {
       this.plan = context.plan;
@@ -248,7 +254,7 @@
 
       // If parameters are all constants, we don't need to dissect an aggregation expression into two parts:
       // function and parameter parts.
-      if (!OpType.isLiteral(param.getType()) && param.getType() != OpType.Column) {
+      if (!OpType.isLiteralType(param.getType()) && param.getType() != OpType.Column) {
         String referenceName = ctx.block.namedExprsMgr.addExpr(param);
         ctx.scalarExprs.add(new NamedExpr(param, referenceName));
         expr.getParams()[i] = new ColumnReferenceExpr(referenceName);
@@ -258,6 +264,54 @@
     return expr;
   }
 
+  public Expr visitWindowFunction(ExprNormalizedResult ctx, Stack<Expr> stack, WindowFunctionExpr expr)
+      throws PlanningException {
+    stack.push(expr);
+
+    WindowSpec windowSpec = expr.getWindowSpec();
+    Expr key;
+
+    WindowSpecReferences windowSpecReferences;
+    if (windowSpec.hasWindowName()) {
+      windowSpecReferences = new WindowSpecReferences(windowSpec.getWindowName());
+    } else {
+      String [] partitionKeyReferenceNames = null;
+      if (windowSpec.hasPartitionBy()) {
+        partitionKeyReferenceNames = new String [windowSpec.getPartitionKeys().length];
+        for (int i = 0; i < windowSpec.getPartitionKeys().length; i++) {
+          key = windowSpec.getPartitionKeys()[i];
+          visit(ctx, stack, key);
+          partitionKeyReferenceNames[i] = ctx.block.namedExprsMgr.addExpr(key);
+        }
+      }
+
+      String [] orderKeyReferenceNames = null;
+      if (windowSpec.hasOrderBy()) {
+        orderKeyReferenceNames = new String[windowSpec.getSortSpecs().length];
+        for (int i = 0; i < windowSpec.getSortSpecs().length; i++) {
+          key = windowSpec.getSortSpecs()[i].getKey();
+          visit(ctx, stack, key);
+          String referenceName = ctx.block.namedExprsMgr.addExpr(key);
+          if (OpType.isAggregationFunction(key.getType())) {
+            ctx.aggExprs.add(new NamedExpr(key, referenceName));
+            windowSpec.getSortSpecs()[i].setKey(new ColumnReferenceExpr(referenceName));
+          }
+          orderKeyReferenceNames[i] = referenceName;
+        }
+      }
+      windowSpecReferences =
+          new WindowSpecReferences(partitionKeyReferenceNames,orderKeyReferenceNames);
+    }
+    ctx.windowSpecs.add(windowSpecReferences);
+
+    String funcExprRef = ctx.block.namedExprsMgr.addExpr(expr);
+    ctx.windowAggExprs.add(new NamedExpr(expr, funcExprRef));
+    stack.pop();
+
+    ctx.block.setHasWindowFunction();
+    return expr;
+  }
+
   ///////////////////////////////////////////////////////////////////////////////////////////////////////////
   // Literal Section
   ///////////////////////////////////////////////////////////////////////////////////////////////////////////
@@ -288,4 +342,40 @@
     }
     return expr;
   }
+
+  public static class WindowSpecReferences {
+    String windowName;
+
+    String [] partitionKeys;
+    String [] orderKeys;
+
+    public WindowSpecReferences(String windowName) {
+      this.windowName = windowName;
+    }
+
+    public WindowSpecReferences(String [] partitionKeys, String [] orderKeys) {
+      this.partitionKeys = partitionKeys;
+      this.orderKeys = orderKeys;
+    }
+
+    public String getWindowName() {
+      return windowName;
+    }
+
+    public boolean hasPartitionKeys() {
+      return partitionKeys != null;
+    }
+
+    public String [] getPartitionKeys() {
+      return partitionKeys;
+    }
+
+    public boolean hasOrderBy() {
+      return orderKeys != null;
+    }
+
+    public String [] getOrderKeys() {
+      return orderKeys;
+    }
+  }
 }
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/LogicalPlan.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/LogicalPlan.java
index 92df760..4e1d313 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/LogicalPlan.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/LogicalPlan.java
@@ -19,6 +19,7 @@
 package org.apache.tajo.engine.planner;
 
 import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
 import org.apache.commons.lang.ObjectUtils;
 import org.apache.tajo.algebra.*;
 import org.apache.tajo.annotation.NotThreadSafe;
@@ -48,6 +49,8 @@
   /** the prefix character for virtual tables */
   public static final char VIRTUAL_TABLE_PREFIX='#';
   public static final char NONAMED_COLUMN_PREFIX='?';
+  public static final char NONAMED_WINDOW_PREFIX='^';
+
   /** it indicates the root block */
   public static final String ROOT_BLOCK = VIRTUAL_TABLE_PREFIX + "ROOT";
   public static final String NONAME_BLOCK_PREFIX = VIRTUAL_TABLE_PREFIX + "QB_";
@@ -55,6 +58,7 @@
   private int nextPid = 0;
   private Integer noNameBlockId = 0;
   private Integer noNameColumnId = 0;
+  private Integer noNameWindowId = 0;
 
   /** a map from between a block name to a block plan */
   private Map<String, QueryBlock> queryBlocks = new LinkedHashMap<String, QueryBlock>();
@@ -134,6 +138,10 @@
     return newAndGetBlock(NONAME_BLOCK_PREFIX + (noNameBlockId++));
   }
 
+  public void resetGeneratedId() {
+    noNameColumnId = 0;
+  }
+
   /**
    * It generates an unique column name from EvalNode. It is usually used for an expression or predicate without
    * a specified name (i.e., alias).
@@ -596,6 +604,8 @@
     private final Map<String, List<String>> aliasMap = TUtil.newHashMap();
     private final Map<OpType, List<Expr>> operatorToExprMap = TUtil.newHashMap();
     private final List<RelationNode> relationList = TUtil.newList();
+    private boolean hasWindowFunction = false;
+
     /**
      * It's a map between nodetype and node. node types can be duplicated. So, latest node type is only kept.
      */
@@ -779,6 +789,14 @@
       return (T) exprToNodeMap.get(ObjectUtils.identityToString(expr));
     }
 
+    public void setHasWindowFunction() {
+      hasWindowFunction = true;
+    }
+
+    public boolean hasWindowSpecs() {
+      return hasWindowFunction;
+    }
+
     /**
      * This flag can be changed as a plan is generated.
      *
@@ -823,4 +841,4 @@
       return blockName;
     }
   }
-}
\ No newline at end of file
+}
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/LogicalPlanPreprocessor.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/LogicalPlanPreprocessor.java
index 2de96c4..4f1218f 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/LogicalPlanPreprocessor.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/LogicalPlanPreprocessor.java
@@ -20,8 +20,8 @@
 
 import org.apache.tajo.algebra.*;
 import org.apache.tajo.catalog.*;
+import org.apache.tajo.common.TajoDataTypes;
 import org.apache.tajo.engine.eval.EvalNode;
-import org.apache.tajo.engine.eval.EvalType;
 import org.apache.tajo.engine.eval.FieldEval;
 import org.apache.tajo.engine.exception.NoSuchColumnException;
 import org.apache.tajo.engine.planner.LogicalPlan.QueryBlock;
@@ -35,13 +35,14 @@
 /**
  * It finds all relations for each block and builds base schema information.
  */
-class LogicalPlanPreprocessor extends BaseAlgebraVisitor<LogicalPlanPreprocessor.PreprocessContext, LogicalNode> {
+public class LogicalPlanPreprocessor extends BaseAlgebraVisitor<LogicalPlanPreprocessor.PreprocessContext, LogicalNode> {
+  private TypeDeterminant typeDeterminant;
   private ExprAnnotator annotator;
 
-  static class PreprocessContext {
-    Session session;
-    LogicalPlan plan;
-    LogicalPlan.QueryBlock currentBlock;
+  public static class PreprocessContext {
+    public Session session;
+    public LogicalPlan plan;
+    public LogicalPlan.QueryBlock currentBlock;
 
     public PreprocessContext(Session session, LogicalPlan plan, LogicalPlan.QueryBlock currentBlock) {
       this.session = session;
@@ -62,6 +63,7 @@
   LogicalPlanPreprocessor(CatalogService catalog, ExprAnnotator annotator) {
     this.catalog = catalog;
     this.annotator = annotator;
+    this.typeDeterminant = new TypeDeterminant(catalog);
   }
 
   @Override
@@ -201,16 +203,13 @@
 
     for (int i = 0; i < expr.getNamedExprs().length; i++) {
       NamedExpr namedExpr = expr.getNamedExprs()[i];
-      EvalNode evalNode = annotator.createEvalNode(ctx.plan, ctx.currentBlock, namedExpr.getExpr());
+      TajoDataTypes.DataType dataType = typeDeterminant.determineDataType(ctx, namedExpr.getExpr());
 
       if (namedExpr.hasAlias()) {
-        targets[i] = new Target(evalNode, namedExpr.getAlias());
-      } else if (evalNode.getType() == EvalType.FIELD) {
-        targets[i] = new Target((FieldEval) evalNode);
+        targets[i] = new Target(new FieldEval(new Column(namedExpr.getAlias(), dataType)));
       } else {
         String generatedName = ctx.plan.generateUniqueColumnName(namedExpr.getExpr());
-        targets[i] = new Target(evalNode, generatedName);
-        namedExpr.setAlias(generatedName);
+        targets[i] = new Target(new FieldEval(new Column(generatedName, dataType)));
       }
     }
     stack.pop(); // <--- Pop
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/LogicalPlanVisitor.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/LogicalPlanVisitor.java
index 963e9f1..0a36610 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/LogicalPlanVisitor.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/LogicalPlanVisitor.java
@@ -40,7 +40,8 @@
 
   RESULT visitGroupBy(CONTEXT context, LogicalPlan plan, LogicalPlan.QueryBlock block, GroupbyNode node,
                       Stack<LogicalNode> stack) throws PlanningException;
-
+  RESULT visitWindowAgg(CONTEXT context, LogicalPlan plan, LogicalPlan.QueryBlock block, WindowAggNode node,
+                      Stack<LogicalNode> stack) throws PlanningException;
   RESULT visitDistinct(CONTEXT context, LogicalPlan plan, LogicalPlan.QueryBlock block, DistinctGroupbyNode node,
                                 Stack<LogicalNode> stack) throws PlanningException;
 
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/LogicalPlanner.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/LogicalPlanner.java
index be7bce6..80390d3 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/LogicalPlanner.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/LogicalPlanner.java
@@ -30,6 +30,7 @@
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.tajo.algebra.*;
+import org.apache.tajo.algebra.WindowSpec;
 import org.apache.tajo.catalog.*;
 import org.apache.tajo.catalog.partition.PartitionMethodDesc;
 import org.apache.tajo.catalog.proto.CatalogProtos;
@@ -43,6 +44,7 @@
 import org.apache.tajo.engine.utils.SchemaUtil;
 import org.apache.tajo.master.session.Session;
 import org.apache.tajo.storage.StorageUtil;
+import org.apache.tajo.util.Pair;
 import org.apache.tajo.util.KeyValueSet;
 import org.apache.tajo.util.TUtil;
 
@@ -117,6 +119,7 @@
     QueryBlock rootBlock = plan.newAndGetBlock(LogicalPlan.ROOT_BLOCK);
     PreprocessContext preProcessorCtx = new PreprocessContext(session, plan, rootBlock);
     preprocessor.visit(preProcessorCtx, new Stack<Expr>(), expr);
+    plan.resetGeneratedId();
 
     PlanContext context = new PlanContext(session, plan, plan.getRootBlock(), debug);
     LogicalNode topMostNode = this.visit(context, new Stack<Expr>(), expr);
@@ -196,7 +199,8 @@
     String [] referenceNames;
     // in prephase, insert all target list into NamedExprManagers.
     // Then it gets reference names, each of which points an expression in target list.
-    referenceNames = doProjectionPrephase(context, projection);
+    Pair<String [], ExprNormalizer.WindowSpecReferences []> referencesPair = doProjectionPrephase(context, projection);
+    referenceNames = referencesPair.getFirst();
 
     ////////////////////////////////////////////////////////
     // Visit and Build Child Plan
@@ -208,6 +212,14 @@
     if (block.isAggregationRequired()) {
       child = insertGroupbyNode(context, child, stack);
     }
+
+    if (block.hasWindowSpecs()) {
+      LogicalNode windowAggNode =
+          insertWindowAggNode(context, child, stack, referenceNames, referencesPair.getSecond());
+      if (windowAggNode != null) {
+        child = windowAggNode;
+      }
+    }
     stack.pop();
     ////////////////////////////////////////////////////////
 
@@ -274,30 +286,73 @@
     projectionNode.setInSchema(dupRemoval.getOutSchema());
   }
 
-  private String [] doProjectionPrephase(PlanContext context, Projection projection) throws PlanningException {
+  private Pair<String [], ExprNormalizer.WindowSpecReferences []> doProjectionPrephase(PlanContext context,
+                                                                                    Projection projection)
+      throws PlanningException {
+
     QueryBlock block = context.queryBlock;
 
     int finalTargetNum = projection.size();
     String [] referenceNames = new String[finalTargetNum];
     ExprNormalizedResult [] normalizedExprList = new ExprNormalizedResult[finalTargetNum];
-    NamedExpr namedExpr;
-    for (int i = 0; i < finalTargetNum; i++) {
-      namedExpr = projection.getNamedExprs()[i];
 
-      if (PlannerUtil.existsAggregationFunction(namedExpr)) {
-        block.setAggregationRequire();
+    List<ExprNormalizer.WindowSpecReferences> windowSpecReferencesList = TUtil.newList();
+
+    List<Integer> targetsIds = normalize(context, projection, normalizedExprList, new Matcher() {
+      @Override
+      public boolean isMatch(Expr expr) {
+        return ExprFinder.finds(expr, OpType.WindowFunction).size() == 0;
       }
-      // dissect an expression into multiple parts (at most dissected into three parts)
-      normalizedExprList[i] = normalizer.normalize(context, namedExpr.getExpr());
-    }
+    });
 
     // Note: Why separate normalization and add(Named)Expr?
     //
     // ExprNormalizer internally makes use of the named exprs in NamedExprsManager.
     // If we don't separate normalization work and addExprWithName, addExprWithName will find named exprs evaluated
     // the same logical node. It will cause impossible evaluation in physical executors.
-    for (int i = 0; i < finalTargetNum; i++) {
-      namedExpr = projection.getNamedExprs()[i];
+    addNamedExprs(block, referenceNames, normalizedExprList, windowSpecReferencesList, projection, targetsIds);
+
+    targetsIds = normalize(context, projection, normalizedExprList, new Matcher() {
+      @Override
+      public boolean isMatch(Expr expr) {
+        return ExprFinder.finds(expr, OpType.WindowFunction).size() > 0;
+      }
+    });
+    addNamedExprs(block, referenceNames, normalizedExprList, windowSpecReferencesList, projection, targetsIds);
+
+    return new Pair<String[], ExprNormalizer.WindowSpecReferences []>(referenceNames,
+        windowSpecReferencesList.toArray(new ExprNormalizer.WindowSpecReferences[windowSpecReferencesList.size()]));
+  }
+
+  private interface Matcher {
+    public boolean isMatch(Expr expr);
+  }
+
+  public List<Integer> normalize(PlanContext context, Projection projection, ExprNormalizedResult [] normalizedExprList,
+                                 Matcher matcher) throws PlanningException {
+    List<Integer> targetIds = new ArrayList<Integer>();
+    for (int i = 0; i < projection.size(); i++) {
+      NamedExpr namedExpr = projection.getNamedExprs()[i];
+
+      if (PlannerUtil.existsAggregationFunction(namedExpr)) {
+        context.queryBlock.setAggregationRequire();
+      }
+
+      if (matcher.isMatch(namedExpr.getExpr())) {
+        // dissect an expression into multiple parts (at most dissected into three parts)
+        normalizedExprList[i] = normalizer.normalize(context, namedExpr.getExpr());
+        targetIds.add(i);
+      }
+    }
+
+    return targetIds;
+  }
+
+  private void addNamedExprs(QueryBlock block, String [] referenceNames, ExprNormalizedResult [] normalizedExprList,
+                             List<ExprNormalizer.WindowSpecReferences> windowSpecReferencesList, Projection projection,
+                             List<Integer> targetIds) throws PlanningException {
+    for (int i : targetIds) {
+      NamedExpr namedExpr = projection.getNamedExprs()[i];
       // Get all projecting references
       if (namedExpr.hasAlias()) {
         NamedExpr aliasedExpr = new NamedExpr(normalizedExprList[i].baseExpr, namedExpr.getAlias());
@@ -309,9 +364,10 @@
       // Add sub-expressions (i.e., aggregation part and scalar part) from dissected parts.
       block.namedExprsMgr.addNamedExprArray(normalizedExprList[i].aggExprs);
       block.namedExprsMgr.addNamedExprArray(normalizedExprList[i].scalarExprs);
-    }
+      block.namedExprsMgr.addNamedExprArray(normalizedExprList[i].windowAggExprs);
 
-    return referenceNames;
+      windowSpecReferencesList.addAll(normalizedExprList[i].windowSpecs);
+    }
   }
 
   /**
@@ -359,64 +415,246 @@
     return targets;
   }
 
+  /**
+   * It checks if all targets of Projectable plan node can be evaluated from the child node.
+   * It can avoid potential errors which possibly occur in physical operators.
+   *
+   * @param block QueryBlock which includes the Projectable node
+   * @param projectable Projectable node to be valid
+   * @throws PlanningException
+   */
   public static void verifyProjectedFields(QueryBlock block, Projectable projectable) throws PlanningException {
-    if (projectable instanceof ProjectionNode && block.hasNode(NodeType.GROUP_BY)) {
-      for (Target target : projectable.getTargets()) {
-        Set<Column> columns = EvalTreeUtil.findUniqueColumns(target.getEvalTree());
-        for (Column c : columns) {
-          if (!projectable.getInSchema().contains(c)) {
-            throw new PlanningException(c.getQualifiedName()
-                + " must appear in the GROUP BY clause or be used in an aggregate function at node ("
-                + projectable.getPID() + ")" );
-          }
-        }
-      }
-    } else  if (projectable instanceof GroupbyNode) {
+    if (projectable instanceof GroupbyNode) {
       GroupbyNode groupbyNode = (GroupbyNode) projectable;
-      // It checks if all column references within each target can be evaluated with the input schema.
-      int groupingColumnNum = groupbyNode.getGroupingColumns().length;
-      for (int i = 0; i < groupingColumnNum; i++) {
-        Set<Column> columns = EvalTreeUtil.findUniqueColumns(groupbyNode.getTargets()[i].getEvalTree());
-        if (!projectable.getInSchema().containsAll(columns)) {
-          throw new PlanningException(String.format("Cannot get the field(s) \"%s\" at node (%d)",
-              TUtil.collectionToString(columns), projectable.getPID()));
-        }
-      }
-      if (groupbyNode.hasAggFunctions()) {
-        for (AggregationFunctionCallEval f : groupbyNode.getAggFunctions()) {
-          Set<Column> columns = EvalTreeUtil.findUniqueColumns(f);
-          for (Column c : columns) {
-            if (!projectable.getInSchema().contains(c)) {
-              throw new PlanningException(String.format("Cannot get the field \"%s\" at node (%d)",
-                  c, projectable.getPID()));
+
+      if (!groupbyNode.isEmptyGrouping()) { // it should be targets instead of
+        int groupingKeyNum = groupbyNode.getGroupingColumns().length;
+
+        for (int i = 0; i < groupingKeyNum; i++) {
+          Target target = groupbyNode.getTargets()[i];
+          if (groupbyNode.getTargets()[i].getEvalTree().getType() == EvalType.FIELD) {
+            FieldEval grpKeyEvalNode = target.getEvalTree();
+            if (!groupbyNode.getInSchema().contains(grpKeyEvalNode.getColumnRef())) {
+              throwCannotEvaluateException(projectable, grpKeyEvalNode.getName());
             }
           }
         }
       }
+
+      if (groupbyNode.hasAggFunctions()) {
+        verifyIfEvalNodesCanBeEvaluated(projectable, groupbyNode.getAggFunctions());
+      }
+
+    } else if (projectable instanceof WindowAggNode) {
+      WindowAggNode windowAggNode = (WindowAggNode) projectable;
+
+      if (windowAggNode.hasPartitionKeys()) {
+        verifyIfColumnCanBeEvaluated(projectable.getInSchema(), projectable, windowAggNode.getPartitionKeys());
+      }
+
+      if (windowAggNode.hasAggFunctions()) {
+        verifyIfEvalNodesCanBeEvaluated(projectable, windowAggNode.getWindowFunctions());
+      }
+
+      if (windowAggNode.hasSortSpecs()) {
+        Column [] sortKeys = PlannerUtil.sortSpecsToSchema(windowAggNode.getSortSpecs()).toArray();
+        verifyIfColumnCanBeEvaluated(projectable.getInSchema(), projectable, sortKeys);
+      }
+
+      // verify targets except for function slots
+      for (int i = 0; i < windowAggNode.getTargets().length - windowAggNode.getWindowFunctions().length; i++) {
+        Target target = windowAggNode.getTargets()[i];
+        Set<Column> columns = EvalTreeUtil.findUniqueColumns(target.getEvalTree());
+        for (Column c : columns) {
+          if (!windowAggNode.getInSchema().contains(c)) {
+            throwCannotEvaluateException(projectable, c.getQualifiedName());
+          }
+        }
+      }
+
     } else if (projectable instanceof RelationNode) {
       RelationNode relationNode = (RelationNode) projectable;
-      for (Target target : projectable.getTargets()) {
-        Set<Column> columns = EvalTreeUtil.findUniqueColumns(target.getEvalTree());
-        for (Column c : columns) {
-          if (!relationNode.getTableSchema().contains(c)) {
-            throw new PlanningException(String.format("Cannot get the field \"%s\" at node (%d)",
-                c, projectable.getPID()));
-          }
-        }
-      }
+      verifyIfTargetsCanBeEvaluated(relationNode.getTableSchema(), (Projectable) relationNode);
+
     } else {
-      for (Target target : projectable.getTargets()) {
-        Set<Column> columns = EvalTreeUtil.findUniqueColumns(target.getEvalTree());
-        for (Column c : columns) {
-          if (!projectable.getInSchema().contains(c)) {
-            throw new PlanningException(String.format("Cannot get the field \"%s\" at node (%d)",
-                c, projectable.getPID()));
-          }
+      verifyIfTargetsCanBeEvaluated(projectable.getInSchema(), projectable);
+    }
+  }
+
+  public static void verifyIfEvalNodesCanBeEvaluated(Projectable projectable, EvalNode[] evalNodes)
+      throws PlanningException {
+    for (EvalNode e : evalNodes) {
+      Set<Column> columns = EvalTreeUtil.findUniqueColumns(e);
+      for (Column c : columns) {
+        if (!projectable.getInSchema().contains(c)) {
+          throwCannotEvaluateException(projectable, c.getQualifiedName());
         }
       }
     }
   }
 
+  public static void verifyIfTargetsCanBeEvaluated(Schema baseSchema, Projectable projectable)
+      throws PlanningException {
+    for (Target target : projectable.getTargets()) {
+      Set<Column> columns = EvalTreeUtil.findUniqueColumns(target.getEvalTree());
+      for (Column c : columns) {
+        if (!baseSchema.contains(c)) {
+          throwCannotEvaluateException(projectable, c.getQualifiedName());
+        }
+      }
+    }
+  }
+
+  public static void verifyIfColumnCanBeEvaluated(Schema baseSchema, Projectable projectable, Column [] columns)
+      throws PlanningException {
+    for (Column c : columns) {
+      if (!baseSchema.contains(c)) {
+        throwCannotEvaluateException(projectable, c.getQualifiedName());
+      }
+    }
+  }
+
+  public static void throwCannotEvaluateException(Projectable projectable, String columnName) throws PlanningException {
+    if (projectable instanceof UnaryNode && ((UnaryNode) projectable).getChild().getType() == NodeType.GROUP_BY) {
+      throw new PlanningException(columnName
+          + " must appear in the GROUP BY clause or be used in an aggregate function at node ("
+          + projectable.getPID() + ")");
+    } else {
+      throw new PlanningException(String.format("Cannot evaluate the field \"%s\" at node (%d)",
+          columnName, projectable.getPID()));
+    }
+  }
+
+  private LogicalNode insertWindowAggNode(PlanContext context, LogicalNode child, Stack<Expr> stack,
+                                          String [] referenceNames,
+                                          ExprNormalizer.WindowSpecReferences [] windowSpecReferenceses)
+      throws PlanningException {
+    LogicalPlan plan = context.plan;
+    QueryBlock block = context.queryBlock;
+    WindowAggNode windowAggNode = context.plan.createNode(WindowAggNode.class);
+    if (child.getType() == NodeType.LIMIT) {
+      LimitNode limitNode = (LimitNode) child;
+      windowAggNode.setChild(limitNode.getChild());
+      windowAggNode.setInSchema(limitNode.getChild().getOutSchema());
+      limitNode.setChild(windowAggNode);
+    } else if (child.getType() == NodeType.SORT) {
+      SortNode sortNode = (SortNode) child;
+      windowAggNode.setChild(sortNode.getChild());
+      windowAggNode.setInSchema(sortNode.getChild().getOutSchema());
+      sortNode.setChild(windowAggNode);
+    } else {
+      windowAggNode.setChild(child);
+      windowAggNode.setInSchema(child.getOutSchema());
+    }
+
+    List<String> winFuncRefs = new ArrayList<String>();
+    List<WindowFunctionEval> winFuncs = new ArrayList<WindowFunctionEval>();
+    List<WindowSpec> rawWindowSpecs = Lists.newArrayList();
+    for (Iterator<NamedExpr> it = block.namedExprsMgr.getIteratorForUnevaluatedExprs(); it.hasNext();) {
+      NamedExpr rawTarget = it.next();
+      try {
+        EvalNode evalNode = exprAnnotator.createEvalNode(context.plan, context.queryBlock, rawTarget.getExpr());
+        if (evalNode.getType() == EvalType.WINDOW_FUNCTION) {
+          winFuncRefs.add(rawTarget.getAlias());
+          winFuncs.add((WindowFunctionEval) evalNode);
+          block.namedExprsMgr.markAsEvaluated(rawTarget.getAlias(), evalNode);
+
+          // TODO - Later, we also consider the possibility that a window function contains only a window name.
+          rawWindowSpecs.add(((WindowFunctionExpr) (rawTarget.getExpr())).getWindowSpec());
+        }
+      } catch (VerifyException ve) {
+      }
+    }
+
+    // we only consider one window definition.
+    if (windowSpecReferenceses[0].hasPartitionKeys()) {
+      Column [] partitionKeyColumns = new Column[windowSpecReferenceses[0].getPartitionKeys().length];
+      int i = 0;
+      for (String partitionKey : windowSpecReferenceses[0].getPartitionKeys()) {
+        if (block.namedExprsMgr.isEvaluated(partitionKey)) {
+          partitionKeyColumns[i++] = block.namedExprsMgr.getTarget(partitionKey).getNamedColumn();
+        } else {
+          throw new PlanningException("Each grouping column expression must be a scalar expression.");
+        }
+      }
+      windowAggNode.setPartitionKeys(partitionKeyColumns);
+    }
+
+    SortSpec [][] sortGroups = new SortSpec[rawWindowSpecs.size()][];
+
+    for (int winSpecIdx = 0; winSpecIdx < rawWindowSpecs.size(); winSpecIdx++) {
+      WindowSpec spec = rawWindowSpecs.get(winSpecIdx);
+      if (spec.hasOrderBy()) {
+        Sort.SortSpec [] sortSpecs = spec.getSortSpecs();
+        int sortNum = sortSpecs.length;
+        String [] sortKeyRefNames = windowSpecReferenceses[winSpecIdx].getOrderKeys();
+        SortSpec [] annotatedSortSpecs = new SortSpec[sortNum];
+
+        Column column;
+        for (int i = 0; i < sortNum; i++) {
+          if (block.namedExprsMgr.isEvaluated(sortKeyRefNames[i])) {
+            column = block.namedExprsMgr.getTarget(sortKeyRefNames[i]).getNamedColumn();
+          } else {
+            throw new IllegalStateException("Unexpected State: " + TUtil.arrayToString(sortSpecs));
+          }
+          annotatedSortSpecs[i] = new SortSpec(column, sortSpecs[i].isAscending(), sortSpecs[i].isNullFirst());
+        }
+
+        sortGroups[winSpecIdx] = annotatedSortSpecs;
+      } else {
+        sortGroups[winSpecIdx] = null;
+      }
+    }
+
+    for (int i = 0; i < winFuncRefs.size(); i++) {
+      WindowFunctionEval winFunc = winFuncs.get(i);
+      if (sortGroups[i] != null) {
+        winFunc.setSortSpecs(sortGroups[i]);
+      }
+    }
+
+    Target [] targets = new Target[referenceNames.length];
+    List<Integer> windowFuncIndices = Lists.newArrayList();
+    Projection projection = (Projection) stack.peek();
+    int windowFuncIdx = 0;
+    for (NamedExpr expr : projection.getNamedExprs()) {
+      if (expr.getExpr().getType() == OpType.WindowFunction) {
+        windowFuncIndices.add(windowFuncIdx);
+      }
+      windowFuncIdx++;
+    }
+    windowAggNode.setWindowFunctions(winFuncs.toArray(new WindowFunctionEval[winFuncs.size()]));
+
+    int targetIdx = 0;
+    for (int i = 0; i < referenceNames.length ; i++) {
+      if (!windowFuncIndices.contains(i)) {
+        targets[targetIdx++] = block.namedExprsMgr.getTarget(referenceNames[i]);
+      }
+    }
+    for (int i = 0; i < winFuncRefs.size(); i++) {
+      targets[targetIdx++] = block.namedExprsMgr.getTarget(winFuncRefs.get(i));
+    }
+    windowAggNode.setTargets(targets);
+    verifyProjectedFields(block, windowAggNode);
+
+    block.registerNode(windowAggNode);
+    postHook(context, stack, null, windowAggNode);
+
+    if (child.getType() == NodeType.LIMIT) {
+      LimitNode limitNode = (LimitNode) child;
+      limitNode.setInSchema(windowAggNode.getOutSchema());
+      limitNode.setOutSchema(windowAggNode.getOutSchema());
+      return null;
+    } else if (child.getType() == NodeType.SORT) {
+      SortNode sortNode = (SortNode) child;
+      sortNode.setInSchema(windowAggNode.getOutSchema());
+      sortNode.setOutSchema(windowAggNode.getOutSchema());
+      return null;
+    } else {
+      return windowAggNode;
+    }
+  }
+
   /**
    * Insert a group-by operator before a sort or a projection operator.
    * It is used only when a group-by clause is not given.
@@ -458,6 +696,8 @@
     // this inserted group-by node doesn't pass through preprocessor. So manually added.
     block.registerNode(groupbyNode);
     postHook(context, stack, null, groupbyNode);
+
+    verifyProjectedFields(block, groupbyNode);
     return groupbyNode;
   }
 
@@ -1541,7 +1781,7 @@
     return new Column(columnDefinition.getColumnName(), convertDataType(columnDefinition));
   }
 
-  static TajoDataTypes.DataType convertDataType(DataTypeExpr dataType) {
+  public static TajoDataTypes.DataType convertDataType(DataTypeExpr dataType) {
     TajoDataTypes.Type type = TajoDataTypes.Type.valueOf(dataType.getTypeName());
 
     TajoDataTypes.DataType.Builder builder = TajoDataTypes.DataType.newBuilder();
@@ -1600,6 +1840,20 @@
     Util SECTION
   ===============================================================================================*/
 
+  public static boolean checkIfBeEvaluatedAtWindowAgg(EvalNode evalNode, WindowAggNode node) {
+    Set<Column> columnRefs = EvalTreeUtil.findUniqueColumns(evalNode);
+
+    if (columnRefs.size() > 0 && !node.getInSchema().containsAll(columnRefs)) {
+      return false;
+    }
+
+    if (EvalTreeUtil.findDistinctAggFunction(evalNode).size() > 0) {
+      return false;
+    }
+
+    return true;
+  }
+
   public static boolean checkIfBeEvaluatedAtGroupBy(EvalNode evalNode, GroupbyNode node) {
     Set<Column> columnRefs = EvalTreeUtil.findUniqueColumns(evalNode);
 
@@ -1607,6 +1861,10 @@
       return false;
     }
 
+    if (EvalTreeUtil.findEvalsByType(evalNode, EvalType.WINDOW_FUNCTION).size() > 0) {
+      return false;
+    }
+
     return true;
   }
 
@@ -1618,6 +1876,10 @@
       return false;
     }
 
+    if (EvalTreeUtil.findEvalsByType(evalNode, EvalType.WINDOW_FUNCTION).size() > 0) {
+      return false;
+    }
+
     if (columnRefs.size() > 0 && !node.getInSchema().containsAll(columnRefs)) {
       return false;
     }
@@ -1658,6 +1920,11 @@
       return false;
     }
 
+    // aggregation functions cannot be evaluated in scan node
+    if (EvalTreeUtil.findEvalsByType(evalNode, EvalType.WINDOW_FUNCTION).size() > 0) {
+      return false;
+    }
+
     if (columnRefs.size() > 0 && !node.getTableSchema().containsAll(columnRefs)) {
       return false;
     }
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/PhysicalPlannerImpl.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/PhysicalPlannerImpl.java
index f41d61d..7b643c3 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/PhysicalPlannerImpl.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/PhysicalPlannerImpl.java
@@ -163,7 +163,6 @@
         leftExec = createPlanRecursive(ctx, subQueryNode.getSubQuery(), stack);
         stack.pop();
         return new ProjectionExec(ctx, subQueryNode, leftExec);
-
       }
 
       case PARTITIONS_SCAN:
@@ -178,6 +177,13 @@
         stack.pop();
         return createGroupByPlan(ctx, grpNode, leftExec);
 
+      case WINDOW_AGG:
+        WindowAggNode windowAggNode = (WindowAggNode) logicalNode;
+        stack.push(windowAggNode);
+        leftExec = createPlanRecursive(ctx, windowAggNode.getChild(), stack);
+        stack.pop();
+        return createWindowAgg(ctx, windowAggNode, leftExec);
+
       case DISTINCT_GROUP_BY:
         DistinctGroupbyNode distinctNode = (DistinctGroupbyNode) logicalNode;
         stack.push(distinctNode);
@@ -973,6 +979,27 @@
     }
   }
 
+  public PhysicalExec createWindowAgg(TaskAttemptContext context,WindowAggNode windowAggNode, PhysicalExec subOp)
+      throws IOException {
+    PhysicalExec child = subOp;
+    if (windowAggNode.hasPartitionKeys()) {
+      Column[] grpColumns = windowAggNode.getPartitionKeys();
+      SortSpec[] sortSpecs = new SortSpec[grpColumns.length];
+      for (int i = 0; i < grpColumns.length; i++) {
+        sortSpecs[i] = new SortSpec(grpColumns[i], true, false);
+      }
+
+      SortNode sortNode = LogicalPlan.createNodeWithoutPID(SortNode.class);
+      sortNode.setSortSpecs(sortSpecs);
+      sortNode.setInSchema(subOp.getSchema());
+      sortNode.setOutSchema(subOp.getSchema());
+      child = new ExternalSortExec(context, sm, sortNode, subOp);
+      LOG.info("The planner chooses [Sort Aggregation] in (" + TUtil.arrayToString(sortSpecs) + ")");
+    }
+
+    return new WindowAggExec(context, windowAggNode, child);
+  }
+
   public PhysicalExec createDistinctGroupByPlan(TaskAttemptContext context,
                                                 DistinctGroupbyNode distinctNode, PhysicalExec subOp)
       throws IOException {
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/PlannerUtil.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/PlannerUtil.java
index 1d8fd0f..55892ee 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/PlannerUtil.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/PlannerUtil.java
@@ -499,10 +499,10 @@
   }
 
   public static SortSpec[] schemaToSortSpecs(Schema schema) {
-    return schemaToSortSpecs(schema.toArray());
+    return columnsToSortSpecs(schema.toArray());
   }
 
-  public static SortSpec[] schemaToSortSpecs(Column[] columns) {
+  public static SortSpec[] columnsToSortSpecs(Column[] columns) {
     SortSpec[] specs = new SortSpec[columns.length];
 
     for (int i = 0; i < columns.length; i++) {
@@ -512,14 +512,8 @@
     return specs;
   }
 
-  public static SortSpec[] columnsToSortSpec(Collection<Column> columns) {
-    SortSpec[] specs = new SortSpec[columns.size()];
-    int i = 0;
-    for (Column column : columns) {
-      specs[i++] = new SortSpec(column, true, false);
-    }
-
-    return specs;
+  public static SortSpec[] columnsToSortSpecs(Collection<Column> columns) {
+    return columnsToSortSpecs(columns.toArray(new Column[columns.size()]));
   }
 
   public static Schema sortSpecsToSchema(SortSpec[] sortSpecs) {
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/SimpleAlgebraVisitor.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/SimpleAlgebraVisitor.java
index bae6e4a..8b34189 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/SimpleAlgebraVisitor.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/SimpleAlgebraVisitor.java
@@ -155,18 +155,13 @@
   }
 
   ///////////////////////////////////////////////////////////////////////////////////////////////////////////
-  // Other Expressions
+  // Functions and General Set Function Section
   ///////////////////////////////////////////////////////////////////////////////////////////////////////////
-
   @Override
   public RESULT visitFunction(CONTEXT ctx, Stack<Expr> stack, FunctionExpr expr) throws PlanningException {
     return super.visitFunction(ctx, stack, expr);
   }
 
-  ///////////////////////////////////////////////////////////////////////////////////////////////////////////
-  // General Set Section
-  ///////////////////////////////////////////////////////////////////////////////////////////////////////////
-
   @Override
   public RESULT visitCountRowsFunction(CONTEXT ctx, Stack<Expr> stack, CountRowsFunctionExpr expr)
       throws PlanningException {
@@ -179,6 +174,11 @@
     return super.visitGeneralSetFunction(ctx, stack, expr);
   }
 
+  @Override
+  public RESULT visitWindowFunction(CONTEXT ctx, Stack<Expr> stack, WindowFunctionExpr expr) throws PlanningException {
+    return super.visitWindowFunction(ctx, stack, expr);
+  }
+
   ///////////////////////////////////////////////////////////////////////////////////////////////////////////
   // Literal Section
   ///////////////////////////////////////////////////////////////////////////////////////////////////////////
@@ -189,11 +189,6 @@
   }
 
   @Override
-  public RESULT visitCastExpr(CONTEXT ctx, Stack<Expr> stack, CastExpr expr) throws PlanningException {
-    return super.visitCastExpr(ctx, stack, expr);
-  }
-
-  @Override
   public RESULT visitLiteral(CONTEXT ctx, Stack<Expr> stack, LiteralValue expr) throws PlanningException {
     return super.visitLiteral(ctx, stack, expr);
   }
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/TypeDeterminant.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/TypeDeterminant.java
new file mode 100644
index 0000000..2e3c1c1
--- /dev/null
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/TypeDeterminant.java
@@ -0,0 +1,301 @@
+/**
+ * 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.engine.utils.DataTypeUtil;
+
+import java.util.Stack;
+
+import static org.apache.tajo.common.TajoDataTypes.DataType;
+import static org.apache.tajo.common.TajoDataTypes.Type.BOOLEAN;
+import static org.apache.tajo.common.TajoDataTypes.Type.NULL_TYPE;
+import static org.apache.tajo.engine.planner.LogicalPlanPreprocessor.PreprocessContext;
+
+public class TypeDeterminant extends SimpleAlgebraVisitor<PreprocessContext, DataType> {
+  private DataType BOOL_TYPE = CatalogUtil.newSimpleDataType(BOOLEAN);
+  private CatalogService catalog;
+
+  public TypeDeterminant(CatalogService catalog) {
+    this.catalog = catalog;
+  }
+
+  public DataType determineDataType(PreprocessContext ctx, Expr expr) throws PlanningException {
+    return visit(ctx, new Stack<Expr>(), expr);
+  }
+
+  @Override
+  public DataType visitUnaryOperator(PreprocessContext ctx, Stack<Expr> stack, UnaryOperator expr) throws PlanningException {
+    stack.push(expr);
+    DataType dataType = null;
+    switch (expr.getType()) {
+    case IsNullPredicate:
+    case ExistsPredicate:
+      dataType = BOOL_TYPE;
+      break;
+    case Cast:
+      dataType = LogicalPlanner.convertDataType(((CastExpr)expr).getTarget());
+      break;
+    default:
+      dataType = visit(ctx, stack, expr.getChild());
+    }
+
+    return dataType;
+  }
+
+  @Override
+  public DataType visitBinaryOperator(PreprocessContext ctx, Stack<Expr> stack, BinaryOperator expr)
+      throws PlanningException {
+    stack.push(expr);
+    DataType lhsType = visit(ctx, stack, expr.getLeft());
+    DataType rhsType = visit(ctx, stack, expr.getRight());
+    stack.pop();
+    return computeBinaryType(expr.getType(), lhsType, rhsType);
+  }
+
+  public DataType computeBinaryType(OpType type, DataType lhsDataType, DataType rhsDataType) throws PlanningException {
+    if(OpType.isLogicalType(type) || OpType.isComparisonType(type)) {
+      return BOOL_TYPE;
+    } else if (OpType.isArithmeticType(type)) {
+      return DataTypeUtil.determineType(lhsDataType, rhsDataType);
+    } else if (type == OpType.Concatenate) {
+      return CatalogUtil.newSimpleDataType(TajoDataTypes.Type.TEXT);
+    } else if (type == OpType.InPredicate) {
+      return BOOL_TYPE;
+    } else if (type == OpType.LikePredicate || type == OpType.SimilarToPredicate || type == OpType.Regexp) {
+      return BOOL_TYPE;
+    } else {
+      throw new PlanningException(type.name() + "is not binary type");
+    }
+  }
+
+  ///////////////////////////////////////////////////////////////////////////////////////////////////////////
+  // Other Predicates Section
+  ///////////////////////////////////////////////////////////////////////////////////////////////////////////
+
+  @Override
+  public DataType visitBetween(PreprocessContext ctx, Stack<Expr> stack, BetweenPredicate expr)
+      throws PlanningException {
+    return CatalogUtil.newSimpleDataType(BOOLEAN);
+  }
+
+  @Override
+  public DataType visitCaseWhen(PreprocessContext ctx, Stack<Expr> stack, CaseWhenPredicate caseWhen)
+      throws PlanningException {
+    DataType lastDataType = null;
+
+    for (CaseWhenPredicate.WhenExpr when : caseWhen.getWhens()) {
+      DataType resultType = visit(ctx, stack, when.getResult());
+      if (lastDataType != null) {
+        lastDataType = ExprAnnotator.getWidestType(lastDataType, resultType);
+      } else {
+        lastDataType = resultType;
+      }
+    }
+
+    if (caseWhen.hasElseResult()) {
+      DataType elseResultType = visit(ctx, stack, caseWhen.getElseResult());
+      lastDataType = ExprAnnotator.getWidestType(lastDataType, elseResultType);
+    }
+
+    return lastDataType;
+  }
+
+  ///////////////////////////////////////////////////////////////////////////////////////////////////////////
+  // Other Expressions
+  ///////////////////////////////////////////////////////////////////////////////////////////////////////////
+
+  @Override
+  public DataType visitColumnReference(PreprocessContext ctx, Stack<Expr> stack, ColumnReferenceExpr expr)
+      throws PlanningException {
+    stack.push(expr);
+    Column column = ctx.plan.resolveColumn(ctx.currentBlock, expr);
+    stack.pop();
+    return column.getDataType();
+  }
+
+  ///////////////////////////////////////////////////////////////////////////////////////////////////////////
+  // General Set Section
+  ///////////////////////////////////////////////////////////////////////////////////////////////////////////
+
+  @Override
+  public DataType visitFunction(PreprocessContext 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];
+    }
+
+    DataType[] givenArgs = new DataType[params.length];
+    DataType[] paramTypes = new DataType[params.length];
+
+    for (int i = 0; i < params.length; i++) {
+      givenArgs[i] = visit(ctx, stack, params[i]);
+      paramTypes[i] = givenArgs[i];
+    }
+
+    stack.pop(); // <--- Pop
+
+    if (!catalog.containFunction(expr.getSignature(), paramTypes)) {
+      throw new NoSuchFunctionException(expr.getSignature(), paramTypes);
+    }
+
+    FunctionDesc funcDesc = catalog.getFunction(expr.getSignature(), paramTypes);
+    return funcDesc.getReturnType();
+  }
+
+  @Override
+  public DataType visitCountRowsFunction(PreprocessContext ctx, Stack<Expr> stack, CountRowsFunctionExpr expr)
+      throws PlanningException {
+    FunctionDesc countRows = catalog.getFunction("count", CatalogProtos.FunctionType.AGGREGATION,
+        new DataType[] {});
+    return countRows.getReturnType();
+  }
+
+  @Override
+  public DataType visitGeneralSetFunction(PreprocessContext ctx, Stack<Expr> stack, GeneralSetFunctionExpr setFunction)
+      throws PlanningException {
+    stack.push(setFunction);
+
+    Expr[] params = setFunction.getParams();
+    DataType[] givenArgs = new DataType[params.length];
+    DataType[] paramTypes = new 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];
+    }
+
+    stack.pop(); // <-- pop
+
+    if (!catalog.containFunction(setFunction.getSignature(), functionType, paramTypes)) {
+      throw new NoSuchFunctionException(setFunction.getSignature(), paramTypes);
+    }
+
+    FunctionDesc funcDesc = catalog.getFunction(setFunction.getSignature(), functionType, paramTypes);
+    return funcDesc.getReturnType();
+  }
+
+  @Override
+  public DataType visitWindowFunction(PreprocessContext ctx, Stack<Expr> stack, WindowFunctionExpr windowFunc)
+      throws PlanningException {
+    stack.push(windowFunc); // <--- Push
+
+    String funcName = windowFunc.getSignature();
+    boolean distinct = windowFunc.isDistinct();
+    Expr[] params = windowFunc.getParams();
+    DataType[] givenArgs = new DataType[params.length];
+    TajoDataTypes.DataType[] paramTypes = new TajoDataTypes.DataType[params.length];
+    CatalogProtos.FunctionType functionType;
+
+    // Rewrite parameters if necessary
+    if (params.length > 0) {
+      givenArgs[0] = visit(ctx, stack, params[0]);
+
+      if (windowFunc.getSignature().equalsIgnoreCase("count")) {
+        paramTypes[0] = CatalogUtil.newSimpleDataType(TajoDataTypes.Type.ANY);
+      } else if (windowFunc.getSignature().equalsIgnoreCase("row_number")) {
+        paramTypes[0] = CatalogUtil.newSimpleDataType(TajoDataTypes.Type.INT8);
+      } else {
+        paramTypes[0] = givenArgs[0];
+      }
+    }
+    stack.pop(); // <--- Pop
+
+    // TODO - containFunction and getFunction should support the function type mask which provides ORing multiple types.
+    // the below checking against WINDOW_FUNCTIONS is a workaround code for the above problem.
+    if (ExprAnnotator.WINDOW_FUNCTIONS.contains(funcName.toLowerCase())) {
+      if (distinct) {
+        throw new NoSuchFunctionException("row_number() does not support distinct keyword.");
+      }
+      functionType = CatalogProtos.FunctionType.WINDOW;
+    } else {
+      functionType = distinct ?
+          CatalogProtos.FunctionType.DISTINCT_AGGREGATION : CatalogProtos.FunctionType.AGGREGATION;
+    }
+
+    if (!catalog.containFunction(windowFunc.getSignature(), functionType, paramTypes)) {
+      throw new NoSuchFunctionException(funcName, paramTypes);
+    }
+
+    FunctionDesc funcDesc = catalog.getFunction(funcName, functionType, paramTypes);
+
+    return funcDesc.getReturnType();
+  }
+
+  ///////////////////////////////////////////////////////////////////////////////////////////////////////////
+  // Literal Section
+  ///////////////////////////////////////////////////////////////////////////////////////////////////////////
+
+  @Override
+  public DataType visitDataType(PreprocessContext ctx, Stack<Expr> stack, DataTypeExpr expr) throws PlanningException {
+    return LogicalPlanner.convertDataType(expr);
+  }
+
+  @Override
+  public DataType visitLiteral(PreprocessContext ctx, Stack<Expr> stack, LiteralValue expr) throws PlanningException {
+    // It must be the same to ExprAnnotator::visitLiteral.
+
+    switch (expr.getValueType()) {
+    case Boolean:
+      return CatalogUtil.newSimpleDataType(BOOLEAN);
+    case String:
+      return CatalogUtil.newSimpleDataType(TajoDataTypes.Type.TEXT);
+    case Unsigned_Integer:
+      return CatalogUtil.newSimpleDataType(TajoDataTypes.Type.INT4);
+    case Unsigned_Large_Integer:
+      return CatalogUtil.newSimpleDataType(TajoDataTypes.Type.INT8);
+    case Unsigned_Float:
+      return CatalogUtil.newSimpleDataType(TajoDataTypes.Type.FLOAT8);
+    default:
+      throw new RuntimeException("Unsupported type: " + expr.getValueType());
+    }
+  }
+
+  @Override
+  public DataType visitNullLiteral(PreprocessContext ctx, Stack<Expr> stack, NullLiteral expr)
+      throws PlanningException {
+    return CatalogUtil.newSimpleDataType(NULL_TYPE);
+  }
+
+  @Override
+  public DataType visitTimestampLiteral(PreprocessContext ctx, Stack<Expr> stack, TimestampLiteral expr)
+      throws PlanningException {
+    return CatalogUtil.newSimpleDataType(TajoDataTypes.Type.TIMESTAMP);
+  }
+
+  @Override
+  public DataType visitTimeLiteral(PreprocessContext ctx, Stack<Expr> stack, TimeLiteral expr)
+      throws PlanningException {
+    return CatalogUtil.newSimpleDataType(TajoDataTypes.Type.TIME);
+  }
+}
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/global/GlobalPlanner.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/global/GlobalPlanner.java
index edd5674..bb7f2e5 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/global/GlobalPlanner.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/global/GlobalPlanner.java
@@ -756,7 +756,7 @@
 
     ExecutionBlock secondStage = context.plan.newExecutionBlock();
     secondStage.setPlan(secondPhaseGroupby);
-    SortSpec [] sortSpecs = PlannerUtil.columnsToSortSpec(firstStageGroupingColumns);
+    SortSpec [] sortSpecs = PlannerUtil.columnsToSortSpecs(firstStageGroupingColumns);
     secondStage.getEnforcer().enforceSortAggregation(secondPhaseGroupby.getPID(), sortSpecs);
 
     // Create a data channel between the first and second stages
@@ -1220,6 +1220,59 @@
     }
 
     @Override
+    public LogicalNode visitWindowAgg(GlobalPlanContext context, LogicalPlan plan, LogicalPlan.QueryBlock block,
+                                    WindowAggNode node, Stack<LogicalNode> stack) throws PlanningException {
+      LogicalNode child = super.visitWindowAgg(context, plan, block, node, stack);
+
+      ExecutionBlock childBlock = context.execBlockMap.remove(child.getPID());
+      ExecutionBlock newExecBlock = buildWindowAgg(context, childBlock, node);
+      context.execBlockMap.put(newExecBlock.getPlan().getPID(), newExecBlock);
+
+      return newExecBlock.getPlan();
+    }
+
+    private ExecutionBlock buildWindowAgg(GlobalPlanContext context, ExecutionBlock lastBlock,
+                                        WindowAggNode windowAgg) throws PlanningException {
+      MasterPlan masterPlan = context.plan;
+
+      ExecutionBlock childBlock = lastBlock;
+      ExecutionBlock currentBlock = masterPlan.newExecutionBlock();
+      DataChannel channel;
+      if (windowAgg.hasPartitionKeys()) { // if there is at one distinct aggregation function
+        channel = new DataChannel(childBlock, currentBlock, RANGE_SHUFFLE, 32);
+        channel.setShuffleKeys(windowAgg.getPartitionKeys());
+      } else {
+        channel = new DataChannel(childBlock, currentBlock, HASH_SHUFFLE, 1);
+        channel.setShuffleKeys(null);
+      }
+      channel.setSchema(windowAgg.getInSchema());
+      channel.setStoreType(storeType);
+
+      LogicalNode childNode = windowAgg.getChild();
+      ScanNode scanNode = buildInputExecutor(masterPlan.getLogicalPlan(), channel);
+
+      if (windowAgg.hasPartitionKeys()) {
+        SortNode sortNode = masterPlan.getLogicalPlan().createNode(SortNode.class);
+        sortNode.setOutSchema(scanNode.getOutSchema());
+        sortNode.setInSchema(scanNode.getOutSchema());
+        sortNode.setSortSpecs(PlannerUtil.columnsToSortSpecs(windowAgg.getPartitionKeys()));
+        sortNode.setChild(childNode);
+        childBlock.setPlan(sortNode);
+
+        windowAgg.setChild(scanNode);
+      } else {
+        windowAgg.setInSchema(scanNode.getOutSchema());
+        windowAgg.setChild(scanNode);
+        childBlock.setPlan(childNode);
+      }
+
+      currentBlock.setPlan(windowAgg);
+      context.plan.addConnect(channel);
+
+      return currentBlock;
+    }
+
+    @Override
     public LogicalNode visitGroupBy(GlobalPlanContext context, LogicalPlan plan, LogicalPlan.QueryBlock block,
                                     GroupbyNode node, Stack<LogicalNode> stack) throws PlanningException {
       LogicalNode child = super.visitGroupBy(context, plan, block, node, stack);
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/logical/NodeType.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/logical/NodeType.java
index de79f93..fa1199b 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/logical/NodeType.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/logical/NodeType.java
@@ -35,6 +35,7 @@
   SORT(SortNode.class),
   HAVING(HavingNode.class),
   GROUP_BY(GroupbyNode.class),
+  WINDOW_AGG(WindowAggNode.class),
   SELECTION(SelectionNode.class),
   JOIN(JoinNode.class),
   UNION(UnionNode.class),
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/logical/WindowAggNode.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/logical/WindowAggNode.java
new file mode 100644
index 0000000..20b195f
--- /dev/null
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/logical/WindowAggNode.java
@@ -0,0 +1,238 @@
+/**
+ * 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.logical;
+
+import com.google.common.base.Objects;
+import com.google.gson.annotations.Expose;
+import org.apache.tajo.catalog.Column;
+import org.apache.tajo.catalog.SortSpec;
+import org.apache.tajo.engine.eval.WindowFunctionEval;
+import org.apache.tajo.engine.planner.PlanString;
+import org.apache.tajo.engine.planner.PlannerUtil;
+import org.apache.tajo.engine.planner.Target;
+import org.apache.tajo.util.TUtil;
+
+public class WindowAggNode extends UnaryNode implements Projectable, Cloneable {
+	/** partition key sets */
+  @Expose private Column [] partitionKeys;
+  /** order key sets */
+  @Expose private SortSpec [] sortSpecs;
+
+  /** Aggregation Functions */
+  @Expose private WindowFunctionEval[] windowFuncs;
+  /**
+   * It's a list of targets. The partition key columns should be followed by window functions.
+   * aggrFunctions keep actual aggregation functions, but it only contains field references.
+   * */
+  @Expose private Target [] targets;
+  @Expose private boolean hasDistinct = false;
+
+  public WindowAggNode(int pid) {
+    super(pid, NodeType.WINDOW_AGG);
+  }
+
+  public final boolean hasPartitionKeys() {
+    return partitionKeys != null && partitionKeys.length > 0;
+  }
+
+  public void setPartitionKeys(Column[] groupingColumns) {
+    this.partitionKeys = groupingColumns;
+  }
+
+	public final Column [] getPartitionKeys() {
+	  return this.partitionKeys;
+	}
+
+  public final boolean hasSortSpecs() {
+    return this.sortSpecs != null;
+  }
+
+  public void setSortSpecs(SortSpec [] sortSpecs) {
+    this.sortSpecs = sortSpecs;
+  }
+
+  public final SortSpec [] getSortSpecs() {
+    return this.sortSpecs;
+  }
+
+  public final boolean isDistinct() {
+    return hasDistinct;
+  }
+
+  public void setDistinct(boolean distinct) {
+    hasDistinct = distinct;
+  }
+
+  public boolean hasAggFunctions() {
+    return this.windowFuncs != null;
+  }
+
+  public WindowFunctionEval [] getWindowFunctions() {
+    return this.windowFuncs;
+  }
+
+  public void setWindowFunctions(WindowFunctionEval[] evals) {
+    this.windowFuncs = evals;
+  }
+
+  @Override
+  public boolean hasTargets() {
+    return this.targets != null;
+  }
+
+  @Override
+  public void setTargets(Target[] targets) {
+    this.targets = targets;
+    setOutSchema(PlannerUtil.targetToSchema(targets));
+  }
+
+  @Override
+  public Target[] getTargets() {
+    return this.targets;
+  }
+  
+  public void setChild(LogicalNode subNode) {
+    super.setChild(subNode);
+  }
+  
+  public String toString() {
+    StringBuilder sb = new StringBuilder("WinAgg (");
+    if (hasPartitionKeys()) {
+      sb.append("partition keys=").append(TUtil.arrayToString(partitionKeys));
+      sb.append(", ");
+    }
+    if (hasAggFunctions()) {
+      sb.append("funcs=").append(TUtil.arrayToString(windowFuncs));
+    }
+    if (hasSortSpecs()) {
+      sb.append("sort=").append(TUtil.arrayToString(sortSpecs));
+    }
+    sb.append(")");
+    return sb.toString();
+  }
+  
+  @Override
+  public boolean equals(Object obj) {
+    if (obj instanceof WindowAggNode) {
+      WindowAggNode other = (WindowAggNode) obj;
+      boolean eq = super.equals(other);
+      eq = eq && TUtil.checkEquals(partitionKeys, other.partitionKeys);
+      eq = eq && TUtil.checkEquals(sortSpecs, other.sortSpecs);
+      eq = eq && TUtil.checkEquals(windowFuncs, other.windowFuncs);
+      eq = eq && TUtil.checkEquals(targets, other.targets);
+      eq = eq && TUtil.checkEquals(hasDistinct, other.hasDistinct);
+      return eq;
+    } else {
+      return false;  
+    }
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hashCode(partitionKeys, sortSpecs, windowFuncs, targets, hasDistinct);
+  }
+  
+  @Override
+  public Object clone() throws CloneNotSupportedException {
+    WindowAggNode grp = (WindowAggNode) super.clone();
+    if (partitionKeys != null) {
+      grp.partitionKeys = new Column[partitionKeys.length];
+      for (int i = 0; i < partitionKeys.length; i++) {
+        grp.partitionKeys[i] = partitionKeys[i];
+      }
+    }
+
+    if (windowFuncs != null) {
+      grp.windowFuncs = new WindowFunctionEval[windowFuncs.length];
+      for (int i = 0; i < windowFuncs.length; i++) {
+        grp.windowFuncs[i] = (WindowFunctionEval) windowFuncs[i].clone();
+      }
+    }
+
+    if (targets != null) {
+      grp.targets = new Target[targets.length];
+      for (int i = 0; i < targets.length; i++) {
+        grp.targets[i] = (Target) targets[i].clone();
+      }
+    }
+
+    return grp;
+  }
+
+  @Override
+  public PlanString getPlanString() {
+    PlanString planStr = new PlanString(this);
+
+    StringBuilder sb = new StringBuilder();
+    sb.append("(");
+    if (hasPartitionKeys()) {
+      sb.append("PARTITION BY ");
+      for (int j = 0; j < partitionKeys.length; j++) {
+        sb.append(partitionKeys[j].getSimpleName());
+        if(j < partitionKeys.length - 1) {
+          sb.append(",");
+        }
+      }
+    }
+
+    if (hasSortSpecs()) {
+      sb.append("ORDER BY ");
+      for (int i = 0; i < sortSpecs.length; i++) {
+        sb.append(sortSpecs[i].getSortKey().getSimpleName()).append(" ")
+            .append(sortSpecs[i].isAscending() ? "asc" : "desc");
+        if( i < sortSpecs.length - 1) {
+          sb.append(",");
+        }
+      }
+    }
+
+    sb.append(")");
+
+    planStr.appendTitle(sb.toString());
+
+    // there can be no aggregation functions
+    if (hasAggFunctions()) {
+      sb = new StringBuilder();
+      sb.append("(");
+
+      for (int j = 0; j < windowFuncs.length; j++) {
+        sb.append(windowFuncs[j]);
+        if(j < windowFuncs.length - 1) {
+          sb.append(",");
+        }
+      }
+      sb.append(")");
+      planStr.appendExplain("exprs: ").appendExplain(sb.toString());
+    }
+
+    sb = new StringBuilder("target list: ");
+    for (int i = 0; i < targets.length; i++) {
+      sb.append(targets[i]);
+      if( i < targets.length - 1) {
+        sb.append(", ");
+      }
+    }
+    planStr.addExplan(sb.toString());
+
+    planStr.addDetail("out schema:").appendDetail(getOutSchema().toString());
+    planStr.addDetail("in schema:").appendDetail(getInSchema().toString());
+
+    return planStr;
+  }
+}
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/logical/WindowSpec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/logical/WindowSpec.java
new file mode 100644
index 0000000..afec0ac
--- /dev/null
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/logical/WindowSpec.java
@@ -0,0 +1,208 @@
+/**
+ * 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.logical;
+
+
+import com.google.common.base.Objects;
+import com.google.gson.annotations.Expose;
+import org.apache.tajo.catalog.Column;
+import org.apache.tajo.engine.eval.EvalNode;
+import org.apache.tajo.util.TUtil;
+
+import static org.apache.tajo.algebra.WindowSpec.WindowFrameEndBoundType;
+import static org.apache.tajo.algebra.WindowSpec.WindowFrameStartBoundType;
+
+public class WindowSpec {
+  @Expose private String windowName;
+
+  @Expose private Column[] partitionKeys;
+
+  @Expose private WindowFrame windowFrame;
+
+  public String getWindowName() {
+    return windowName;
+  }
+
+  public boolean hasPartitionKeys() {
+    return partitionKeys != null;
+  }
+
+  public Column [] getPartitionKeys() {
+    return partitionKeys;
+  }
+
+  public boolean hasWindowFrame() {
+    return windowFrame != null;
+  }
+
+  public WindowFrame getWindowFrame() {
+    return windowFrame;
+  }
+
+  @Override
+  public boolean equals(Object obj) {
+    if (obj instanceof WindowSpec) {
+      WindowSpec another = (WindowSpec) obj;
+      return
+          TUtil.checkEquals(windowName, another.windowName) &&
+          TUtil.checkEquals(partitionKeys, another.partitionKeys) &&
+          TUtil.checkEquals(windowFrame, another.windowFrame);
+    } else {
+      return false;
+    }
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hashCode(windowName, partitionKeys, windowFrame);
+  }
+
+  public static class WindowFrame {
+    @Expose private WindowStartBound startBound;
+    @Expose private WindowEndBound endBound;
+    @Expose org.apache.tajo.algebra.WindowSpec.WindowFrameUnit unit; // TODO - to be supported
+
+    public WindowFrame() {
+      this.startBound = new WindowStartBound(WindowFrameStartBoundType.UNBOUNDED_PRECEDING);
+      this.endBound = new WindowEndBound(WindowFrameEndBoundType.UNBOUNDED_FOLLOWING);
+    }
+
+    public WindowFrame(WindowStartBound startBound) {
+      this.startBound = startBound;
+    }
+
+    public WindowFrame(WindowStartBound startBound, WindowEndBound endBound) {
+      this(startBound);
+      this.endBound = endBound;
+    }
+
+    public WindowStartBound getStartBound() {
+      return startBound;
+    }
+
+    public boolean hasEndBound() {
+      return endBound != null;
+    }
+
+    public WindowEndBound getEndBound() {
+      return endBound;
+    }
+
+    public boolean hasFrameUnit() {
+      return this.unit != null;
+    }
+
+    public void setFrameUnit(org.apache.tajo.algebra.WindowSpec.WindowFrameUnit unit) {
+      this.unit = unit;
+    }
+
+    public org.apache.tajo.algebra.WindowSpec.WindowFrameUnit getFrameUnit() {
+      return this.unit;
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+      if (obj instanceof WindowFrame) {
+        WindowFrame another = (WindowFrame) obj;
+        return
+            TUtil.checkEquals(startBound, another.startBound) &&
+            TUtil.checkEquals(endBound, another.endBound) &&
+            TUtil.checkEquals(unit, another.unit);
+      } else {
+        return false;
+      }
+    }
+
+    public int hashCode() {
+      return Objects.hashCode(startBound, endBound, unit);
+    }
+  }
+
+  public static class WindowStartBound {
+    @Expose private WindowFrameStartBoundType boundType;
+    @Expose private EvalNode number;
+
+    public WindowStartBound(WindowFrameStartBoundType type) {
+      this.boundType = type;
+    }
+
+    public WindowFrameStartBoundType getBoundType() {
+      return boundType;
+    }
+
+    public void setNumber(EvalNode number) {
+      this.number = number;
+    }
+
+    public EvalNode getNumber() {
+      return number;
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+      if (obj instanceof WindowStartBound) {
+        WindowStartBound other = (WindowStartBound) obj;
+        return boundType == other.boundType && number.equals(other.number);
+      } else {
+        return false;
+      }
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hashCode(boundType, number);
+    }
+  }
+
+  public static class WindowEndBound {
+    @Expose private WindowFrameEndBoundType boundType;
+    @Expose private EvalNode number;
+
+    public WindowEndBound(WindowFrameEndBoundType type) {
+      this.boundType = type;
+    }
+
+    public WindowFrameEndBoundType getBoundType() {
+      return boundType;
+    }
+
+    public EvalNode setNumber(EvalNode number) {
+      return number;
+    }
+
+    public EvalNode getNumber() {
+      return number;
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+      if (obj instanceof WindowStartBound) {
+        WindowEndBound other = (WindowEndBound) obj;
+        return boundType == other.boundType && number.equals(other.number);
+      } else {
+        return false;
+      }
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hashCode(boundType, number);
+    }
+  }
+}
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/WindowAggExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/WindowAggExec.java
new file mode 100644
index 0000000..7aeed13
--- /dev/null
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/WindowAggExec.java
@@ -0,0 +1,340 @@
+/**
+ * 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.physical;
+
+import com.google.common.collect.Lists;
+import org.apache.tajo.catalog.Column;
+import org.apache.tajo.catalog.Schema;
+import org.apache.tajo.catalog.SortSpec;
+import org.apache.tajo.datum.Datum;
+import org.apache.tajo.engine.eval.WindowFunctionEval;
+import org.apache.tajo.engine.function.FunctionContext;
+import org.apache.tajo.engine.planner.logical.WindowAggNode;
+import org.apache.tajo.engine.planner.logical.WindowSpec;
+import org.apache.tajo.storage.Tuple;
+import org.apache.tajo.storage.TupleComparator;
+import org.apache.tajo.storage.VTuple;
+import org.apache.tajo.worker.TaskAttemptContext;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+
+/**
+ * The sort-based window aggregation operator
+ */
+public class WindowAggExec extends UnaryPhysicalExec {
+  // plan information
+  protected final int outputColumnNum;
+  protected final int nonFunctionColumnNum;
+  protected final int nonFunctionColumns[];
+
+  protected final int functionNum;
+  protected final WindowFunctionEval functions[];
+
+  protected Schema schemaForOrderBy;
+  protected int sortKeyColumns[];
+  protected final boolean hasPartitionKeys;
+  protected final int partitionKeyNum;
+  protected final int partitionKeyIds[];
+
+  // for evaluation
+  protected FunctionContext contexts [];
+  protected Tuple lastKey = null;
+  protected boolean noMoreTuples = false;
+  private boolean [] orderedFuncFlags;
+  private boolean [] aggFuncFlags;
+  private boolean [] windowFuncFlags;
+  private boolean [] endUnboundedFollowingFlags;
+  private boolean [] endCurrentRowFlags;
+
+  private boolean endCurrentRow = false;
+
+  // operator state
+  enum WindowState {
+    NEW_WINDOW,
+    ACCUMULATING_WINDOW,
+    EVALUATION,
+    RETRIEVING_FROM_WINDOW,
+    END_OF_TUPLE
+  }
+
+  // Transient state
+  boolean firstTime = true;
+  List<Tuple> evaluatedTuples = null;
+  List<Tuple> accumulatedInTuples = null;
+  List<Tuple> nextAccumulatedProjected = null;
+  List<Tuple> nextAccumulatedInTuples = null;
+  WindowState state = WindowState.NEW_WINDOW;
+  Iterator<Tuple> tupleInFrameIterator = null;
+
+  public WindowAggExec(TaskAttemptContext context, WindowAggNode plan, PhysicalExec child) throws IOException {
+    super(context, plan.getInSchema(), plan.getOutSchema(), child);
+
+    if (plan.hasPartitionKeys()) {
+      final Column[] keyColumns = plan.getPartitionKeys();
+      partitionKeyNum = keyColumns.length;
+      partitionKeyIds = new int[partitionKeyNum];
+      Column col;
+      for (int idx = 0; idx < plan.getPartitionKeys().length; idx++) {
+        col = keyColumns[idx];
+        partitionKeyIds[idx] = inSchema.getColumnId(col.getQualifiedName());
+      }
+      hasPartitionKeys = true;
+    } else {
+      partitionKeyNum = 0;
+      partitionKeyIds = null;
+      hasPartitionKeys = false;
+    }
+
+    if (plan.hasAggFunctions()) {
+      functions = plan.getWindowFunctions();
+      functionNum = functions.length;
+
+      orderedFuncFlags = new boolean[functions.length];
+      windowFuncFlags = new boolean[functions.length];
+      aggFuncFlags = new boolean[functions.length];
+
+      endUnboundedFollowingFlags = new boolean[functions.length];
+      endCurrentRowFlags = new boolean[functions.length];
+
+      List<Column> additionalSortKeyColumns = Lists.newArrayList();
+      Schema rewrittenSchema = new Schema(outSchema);
+      for (int i = 0; i < functions.length; i++) {
+        WindowSpec.WindowEndBound endBound = functions[i].getWindowFrame().getEndBound();
+        switch (endBound.getBoundType()) {
+        case CURRENT_ROW:
+          endCurrentRowFlags[i] = true; break;
+        case UNBOUNDED_FOLLOWING:
+          endUnboundedFollowingFlags[i] = true; break;
+        default:
+        }
+
+        switch (functions[i].getFuncDesc().getFuncType()) {
+        case AGGREGATION:
+        case DISTINCT_AGGREGATION:
+          aggFuncFlags[i] = true; break;
+        case WINDOW:
+          windowFuncFlags[i] = true; break;
+        default:
+        }
+
+        if (functions[i].hasSortSpecs()) {
+          orderedFuncFlags[i] = true;
+
+          for (SortSpec sortSpec : functions[i].getSortSpecs()) {
+            if (!rewrittenSchema.contains(sortSpec.getSortKey())) {
+              additionalSortKeyColumns.add(sortSpec.getSortKey());
+            }
+          }
+        }
+      }
+
+      sortKeyColumns = new int[additionalSortKeyColumns.size()];
+      schemaForOrderBy = new Schema(outSchema);
+      for (int i = 0; i < additionalSortKeyColumns.size(); i++) {
+        sortKeyColumns[i] = i;
+        schemaForOrderBy.addColumn(additionalSortKeyColumns.get(i));
+      }
+    } else {
+      functions = new WindowFunctionEval[0];
+      functionNum = 0;
+      schemaForOrderBy = outSchema;
+    }
+
+
+    nonFunctionColumnNum = plan.getTargets().length - functionNum;
+    nonFunctionColumns = new int[nonFunctionColumnNum];
+    for (int idx = 0; idx < plan.getTargets().length - functionNum; idx++) {
+      nonFunctionColumns[idx] = inSchema.getColumnId(plan.getTargets()[idx].getCanonicalName());
+    }
+
+    outputColumnNum = nonFunctionColumnNum + functionNum;
+  }
+
+  private void transition(WindowState state) {
+    this.state = state;
+  }
+
+  @Override
+  public Tuple next() throws IOException {
+    Tuple currentKey = null;
+    Tuple readTuple = null;
+
+    while(!context.isStopped() && state != WindowState.END_OF_TUPLE) {
+
+      if (state == WindowState.NEW_WINDOW) {
+        initWindow();
+        transition(WindowState.ACCUMULATING_WINDOW);
+      }
+
+      if (state != WindowState.RETRIEVING_FROM_WINDOW) { // read an input tuple and build a partition key
+        readTuple = child.next();
+
+        if (readTuple == null) { // the end of tuple
+          noMoreTuples = true;
+          transition(WindowState.EVALUATION);
+        }
+
+        if (readTuple != null && hasPartitionKeys) { // get a key tuple
+          currentKey = new VTuple(partitionKeyIds.length);
+          for (int i = 0; i < partitionKeyIds.length; i++) {
+            currentKey.put(i, readTuple.get(partitionKeyIds[i]));
+          }
+        }
+      }
+
+      if (state == WindowState.ACCUMULATING_WINDOW) {
+        accumulatingWindow(currentKey, readTuple);
+      }
+
+      if (state == WindowState.EVALUATION) {
+        evaluationWindowFrame();
+
+        tupleInFrameIterator = evaluatedTuples.iterator();
+        transition(WindowState.RETRIEVING_FROM_WINDOW);
+      }
+
+      if (state == WindowState.RETRIEVING_FROM_WINDOW) {
+        if (tupleInFrameIterator.hasNext()) {
+          return tupleInFrameIterator.next();
+        } else {
+          finalizeWindow();
+        }
+      }
+    }
+
+    return null;
+  }
+
+  private void initWindow() {
+    if (firstTime) {
+      accumulatedInTuples = Lists.newArrayList();
+
+      contexts = new FunctionContext[functionNum];
+      for(int evalIdx = 0; evalIdx < functionNum; evalIdx++) {
+        contexts[evalIdx] = functions[evalIdx].newContext();
+      }
+      firstTime = false;
+    }
+  }
+
+  private void accumulatingWindow(Tuple currentKey, Tuple inTuple) {
+
+    if (lastKey == null || lastKey.equals(currentKey)) { // if the current key is same to the previous key
+      accumulatedInTuples.add(new VTuple(inTuple));
+
+    } else {
+      // if the current key is different from the previous key,
+      // the current key belongs to the next window frame. preaccumulatingNextWindow() will
+      // aggregate the current key for next window frame.
+      preAccumulatingNextWindow(inTuple);
+      transition(WindowState.EVALUATION);
+    }
+
+    lastKey = currentKey;
+  }
+
+  private void preAccumulatingNextWindow(Tuple inTuple) {
+    Tuple projectedTuple = new VTuple(outSchema.size());
+    for(int idx = 0; idx < nonFunctionColumnNum; idx++) {
+      projectedTuple.put(idx, inTuple.get(nonFunctionColumns[idx]));
+    }
+    nextAccumulatedProjected = Lists.newArrayList();
+    nextAccumulatedProjected.add(projectedTuple);
+    nextAccumulatedInTuples = Lists.newArrayList();
+    nextAccumulatedInTuples.add(new VTuple(inTuple));
+  }
+
+  private void evaluationWindowFrame() {
+    TupleComparator comp;
+
+    evaluatedTuples = new ArrayList<Tuple>();
+
+    for (int i = 0; i <accumulatedInTuples.size(); i++) {
+      Tuple inTuple = accumulatedInTuples.get(i);
+
+      Tuple projectedTuple = new VTuple(schemaForOrderBy.size());
+      for (int c = 0; c < nonFunctionColumnNum; c++) {
+        projectedTuple.put(c, inTuple.get(nonFunctionColumns[c]));
+      }
+      for (int c = 0; c < sortKeyColumns.length; c++) {
+        projectedTuple.put(outputColumnNum + c, inTuple.get(sortKeyColumns[c]));
+      }
+
+      evaluatedTuples.add(projectedTuple);
+    }
+
+    for (int idx = 0; idx < functions.length; idx++) {
+      if (orderedFuncFlags[idx]) {
+        comp = new TupleComparator(inSchema, functions[idx].getSortSpecs());
+        Collections.sort(accumulatedInTuples, comp);
+        comp = new TupleComparator(schemaForOrderBy, functions[idx].getSortSpecs());
+        Collections.sort(evaluatedTuples, comp);
+      }
+
+      for (int i = 0; i < accumulatedInTuples.size(); i++) {
+        Tuple inTuple = accumulatedInTuples.get(i);
+        Tuple outTuple = evaluatedTuples.get(i);
+
+        functions[idx].merge(contexts[idx], inSchema, inTuple);
+
+        if (windowFuncFlags[idx]) {
+          Datum result = functions[idx].terminate(contexts[idx]);
+          outTuple.put(nonFunctionColumnNum + idx, result);
+        }
+      }
+
+      if (aggFuncFlags[idx]) {
+        Datum result = functions[idx].terminate(contexts[idx]);
+        for (int i = 0; i < evaluatedTuples.size(); i++) {
+          Tuple outTuple = evaluatedTuples.get(i);
+          outTuple.put(nonFunctionColumnNum + idx, result);
+        }
+      }
+    }
+  }
+
+  private void finalizeWindow() {
+    evaluatedTuples.clear();
+    accumulatedInTuples.clear();
+
+    if (noMoreTuples) {
+      transition(WindowState.END_OF_TUPLE);
+    } else {
+      accumulatedInTuples = nextAccumulatedInTuples;
+
+      contexts = new FunctionContext[functionNum];
+      for(int evalIdx = 0; evalIdx < functionNum; evalIdx++) {
+        contexts[evalIdx] = functions[evalIdx].newContext();
+      }
+      transition(WindowState.NEW_WINDOW);
+    }
+  }
+
+  @Override
+  public void rescan() throws IOException {
+    super.rescan();
+
+    lastKey = null;
+    noMoreTuples = false;
+  }
+}
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/rewrite/FilterPushDownRule.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/rewrite/FilterPushDownRule.java
index 4215423..34c88a6 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/rewrite/FilterPushDownRule.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/rewrite/FilterPushDownRule.java
@@ -18,8 +18,7 @@
 
 package org.apache.tajo.engine.planner.rewrite;
 
-import com.google.common.collect.Lists;
-import com.google.common.collect.Sets;
+import com.google.common.collect.*;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.tajo.algebra.JoinType;
@@ -464,7 +463,6 @@
 
     context.setFiltersTobePushed(new HashSet<EvalNode>(transformedMap.keySet()));
     visit(context, plan, plan.getBlock(node.getSubQuery()));
-
     context.setToOrigin(transformedMap);
 
     return node;
@@ -512,18 +510,18 @@
     List<EvalNode> notMatched = new ArrayList<EvalNode>();
 
     //copy -> origin
-    Map<EvalNode, EvalNode> matched = findCanPushdownAndTransform(
+    BiMap<EvalNode, EvalNode> transformedMap = findCanPushdownAndTransform(
         context, projectionNode, childNode, notMatched, null, false, 0);
 
-    context.setFiltersTobePushed(matched.keySet());
+    context.setFiltersTobePushed(transformedMap.keySet());
 
     stack.push(projectionNode);
-    LogicalNode current = visit(context, plan, plan.getBlock(childNode), childNode, stack);
+    childNode = visit(context, plan, plan.getBlock(childNode), childNode, stack);
     stack.pop();
 
     // find not matched after visiting child
     for (EvalNode eval: context.pushingDownFilters) {
-      notMatched.add(matched.get(eval));
+      notMatched.add(transformedMap.get(eval));
     }
 
     EvalNode qual = null;
@@ -536,24 +534,39 @@
     }
 
     // If there is not matched node add SelectionNode and clear context.pushingDownFilters
-    if (qual != null) {
+    if (qual != null && LogicalPlanner.checkIfBeEvaluatedAtThis(qual, projectionNode)) {
       SelectionNode selectionNode = plan.createNode(SelectionNode.class);
-      selectionNode.setInSchema(current.getOutSchema());
-      selectionNode.setOutSchema(current.getOutSchema());
+      selectionNode.setInSchema(childNode.getOutSchema());
+      selectionNode.setOutSchema(childNode.getOutSchema());
       selectionNode.setQual(qual);
       block.registerNode(selectionNode);
 
       projectionNode.setChild(selectionNode);
-      selectionNode.setChild(current);
+      selectionNode.setChild(childNode);
+
+      // clean all remain filters because all conditions are merged into a qual
+      context.pushingDownFilters.clear();
     }
 
-    //notify all eval matched to upper
-    context.pushingDownFilters.clear();
+    // if there are remain filters, recover the original names and give back to the upper query block.
+    if (context.pushingDownFilters.size() > 0) {
+      ImmutableSet<EvalNode> copy = ImmutableSet.copyOf(context.pushingDownFilters);
+      context.pushingDownFilters.clear();
+      context.pushingDownFilters.addAll(reverseTransform(transformedMap, copy));
+    }
 
-    return current;
+    return projectionNode;
   }
 
-  private Map<EvalNode, EvalNode> findCanPushdownAndTransform(
+  private Collection<EvalNode> reverseTransform(BiMap<EvalNode, EvalNode> map, Set<EvalNode> remainFilters) {
+    Set<EvalNode> reversed = Sets.newHashSet();
+    for (EvalNode evalNode : remainFilters) {
+      reversed.add(map.get(evalNode));
+    }
+    return reversed;
+  }
+
+  private BiMap<EvalNode, EvalNode> findCanPushdownAndTransform(
       FilterPushDownContext context, Projectable node,
       LogicalNode childNode, List<EvalNode> notMatched,
       Set<String> partitionColumns,
@@ -565,7 +578,7 @@
     }
 
     // copy -> origin
-    Map<EvalNode, EvalNode> matched = new HashMap<EvalNode, EvalNode>();
+    BiMap<EvalNode, EvalNode> matched = HashBiMap.create();
 
     for (EvalNode eval : context.pushingDownFilters) {
       if (ignoreJoin && EvalTreeUtil.isJoinQual(eval, true)) {
@@ -754,6 +767,17 @@
   }
 
   @Override
+  public LogicalNode visitWindowAgg(FilterPushDownContext context, LogicalPlan plan,
+                                  LogicalPlan.QueryBlock block, WindowAggNode winAggNode,
+                                  Stack<LogicalNode> stack) throws PlanningException {
+    stack.push(winAggNode);
+    super.visitWindowAgg(context, plan, block, winAggNode, stack);
+    stack.pop();
+    return winAggNode;
+  }
+
+
+  @Override
   public LogicalNode visitGroupBy(FilterPushDownContext context, LogicalPlan plan,
                                   LogicalPlan.QueryBlock block, GroupbyNode groupbyNode,
                                   Stack<LogicalNode> stack) throws PlanningException {
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/rewrite/ProjectionPushDownRule.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/rewrite/ProjectionPushDownRule.java
index 4e4b5c3..2bc210c 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/rewrite/ProjectionPushDownRule.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/rewrite/ProjectionPushDownRule.java
@@ -592,6 +592,99 @@
     return node;
   }
 
+  public LogicalNode visitWindowAgg(Context context, LogicalPlan plan, LogicalPlan.QueryBlock block, WindowAggNode node,
+                        Stack<LogicalNode> stack) throws PlanningException {
+    Context newContext = new Context(context);
+
+    if (node.hasPartitionKeys()) {
+      for (Column c : node.getPartitionKeys()) {
+        newContext.addNecessaryReferences(new FieldEval(c));
+      }
+    }
+
+    if (node.hasSortSpecs()) {
+      for (SortSpec sortSpec : node.getSortSpecs()) {
+        newContext.addNecessaryReferences(new FieldEval(sortSpec.getSortKey()));
+      }
+    }
+
+    for (WindowFunctionEval winFunc : node.getWindowFunctions()) {
+      if (winFunc.hasSortSpecs()) {
+        for (SortSpec sortSpec : winFunc.getSortSpecs()) {
+          newContext.addNecessaryReferences(new FieldEval(sortSpec.getSortKey()));
+        }
+      }
+    }
+
+
+    int nonFunctionColumnNum = node.getTargets().length - node.getWindowFunctions().length;
+    LinkedHashSet<String> nonFunctionColumns = Sets.newLinkedHashSet();
+    for (int i = 0; i < nonFunctionColumnNum; i++) {
+      FieldEval fieldEval = (new FieldEval(node.getTargets()[i].getNamedColumn()));
+      nonFunctionColumns.add(newContext.addExpr(fieldEval));
+    }
+
+    final String [] aggEvalNames;
+    if (node.hasAggFunctions()) {
+      final int evalNum = node.getWindowFunctions().length;
+      aggEvalNames = new String[evalNum];
+      for (int evalIdx = 0, targetIdx = nonFunctionColumnNum; targetIdx < node.getTargets().length; evalIdx++,
+          targetIdx++) {
+        Target target = node.getTargets()[targetIdx];
+        WindowFunctionEval winFunc = node.getWindowFunctions()[evalIdx];
+        aggEvalNames[evalIdx] = newContext.addExpr(new Target(winFunc, target.getCanonicalName()));
+      }
+    } else {
+      aggEvalNames = null;
+    }
+
+    // visit a child node
+    LogicalNode child = super.visitWindowAgg(newContext, plan, block, node, stack);
+
+    node.setInSchema(child.getOutSchema());
+
+    List<Target> targets = Lists.newArrayList();
+    if (nonFunctionColumnNum > 0) {
+      for (String column : nonFunctionColumns) {
+        Target target = context.targetListMgr.getTarget(column);
+
+        // it rewrite grouping keys.
+        // This rewrite sets right column names and eliminates duplicated grouping keys.
+        if (context.targetListMgr.isEvaluated(column)) {
+          targets.add(new Target(new FieldEval(target.getNamedColumn())));
+        } else {
+          if (target.getEvalTree().getType() == EvalType.FIELD) {
+           targets.add(target);
+          }
+        }
+      }
+    }
+
+    // Getting projected targets
+    if (node.hasAggFunctions() && aggEvalNames != null) {
+      WindowFunctionEval [] aggEvals = new WindowFunctionEval[aggEvalNames.length];
+      int i = 0;
+      for (Iterator<String> it = getFilteredReferences(aggEvalNames, TUtil.newList(aggEvalNames)); it.hasNext();) {
+
+        String referenceName = it.next();
+        Target target = context.targetListMgr.getTarget(referenceName);
+
+        if (LogicalPlanner.checkIfBeEvaluatedAtWindowAgg(target.getEvalTree(), node)) {
+          aggEvals[i++] = target.getEvalTree();
+          context.targetListMgr.markAsEvaluated(target);
+
+          targets.add(new Target(new FieldEval(target.getNamedColumn())));
+        }
+      }
+      if (aggEvals.length > 0) {
+        node.setWindowFunctions(aggEvals);
+      }
+    }
+
+    node.setTargets(targets.toArray(new Target[targets.size()]));
+    return node;
+  }
+
   public LogicalNode visitGroupBy(Context context, LogicalPlan plan, LogicalPlan.QueryBlock block, GroupbyNode node,
                              Stack<LogicalNode> stack) throws PlanningException {
     Context newContext = new Context(context);
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/utils/DataTypeUtil.java b/tajo-core/src/main/java/org/apache/tajo/engine/utils/DataTypeUtil.java
new file mode 100644
index 0000000..084db21
--- /dev/null
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/utils/DataTypeUtil.java
@@ -0,0 +1,121 @@
+/**
+ * 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.utils;
+
+import org.apache.tajo.catalog.CatalogUtil;
+import org.apache.tajo.common.TajoDataTypes;
+import org.apache.tajo.engine.eval.InvalidEvalException;
+
+public class DataTypeUtil {
+  /**
+   * This is verified by ExprsVerifier.checkArithmeticOperand().
+   */
+  public static TajoDataTypes.DataType determineType(TajoDataTypes.DataType left, TajoDataTypes.DataType right) throws InvalidEvalException {
+    switch (left.getType()) {
+
+    case INT1:
+    case INT2:
+    case INT4: {
+      switch(right.getType()) {
+      case INT1:
+      case INT2:
+      case INT4: return CatalogUtil.newSimpleDataType(TajoDataTypes.Type.INT4);
+      case INT8: return CatalogUtil.newSimpleDataType(TajoDataTypes.Type.INT8);
+      case FLOAT4: return CatalogUtil.newSimpleDataType(TajoDataTypes.Type.FLOAT4);
+      case FLOAT8: return CatalogUtil.newSimpleDataType(TajoDataTypes.Type.FLOAT8);
+      case DATE: return CatalogUtil.newSimpleDataType(TajoDataTypes.Type.DATE);
+      case INTERVAL: return CatalogUtil.newSimpleDataType(TajoDataTypes.Type.INTERVAL);
+      }
+    }
+
+    case INT8: {
+      switch(right.getType()) {
+      case INT1:
+      case INT2:
+      case INT4:
+      case INT8: return CatalogUtil.newSimpleDataType(TajoDataTypes.Type.INT8);
+      case FLOAT4: return CatalogUtil.newSimpleDataType(TajoDataTypes.Type.FLOAT4);
+      case FLOAT8: return CatalogUtil.newSimpleDataType(TajoDataTypes.Type.FLOAT8);
+      case DATE: return CatalogUtil.newSimpleDataType(TajoDataTypes.Type.DATE);
+      case INTERVAL: return CatalogUtil.newSimpleDataType(TajoDataTypes.Type.INTERVAL);
+      }
+    }
+
+    case FLOAT4: {
+      switch(right.getType()) {
+      case INT1:
+      case INT2:
+      case INT4: return CatalogUtil.newSimpleDataType(TajoDataTypes.Type.FLOAT4);
+      case INT8: return CatalogUtil.newSimpleDataType(TajoDataTypes.Type.FLOAT4);
+      case FLOAT4: return CatalogUtil.newSimpleDataType(TajoDataTypes.Type.FLOAT4);
+      case FLOAT8: return CatalogUtil.newSimpleDataType(TajoDataTypes.Type.FLOAT8);
+      case INTERVAL: return CatalogUtil.newSimpleDataType(TajoDataTypes.Type.INTERVAL);
+      }
+    }
+
+    case FLOAT8: {
+      switch(right.getType()) {
+      case INT1:
+      case INT2:
+      case INT4:
+      case INT8:
+      case FLOAT4:
+      case FLOAT8: return CatalogUtil.newSimpleDataType(TajoDataTypes.Type.FLOAT8);
+      case INTERVAL: return CatalogUtil.newSimpleDataType(TajoDataTypes.Type.INTERVAL);
+      }
+    }
+
+    case DATE: {
+      switch(right.getType()) {
+      case INT2:
+      case INT4:
+      case INT8: return CatalogUtil.newSimpleDataType(TajoDataTypes.Type.DATE);
+      case INTERVAL:
+      case TIME: return CatalogUtil.newSimpleDataType(TajoDataTypes.Type.TIMESTAMP);
+      case DATE: return CatalogUtil.newSimpleDataType(TajoDataTypes.Type.INT4);
+      }
+    }
+
+    case TIME: {
+      switch(right.getType()) {
+      case INTERVAL: return CatalogUtil.newSimpleDataType(TajoDataTypes.Type.TIME);
+      case TIME: return CatalogUtil.newSimpleDataType(TajoDataTypes.Type.INTERVAL);
+      case DATE: return CatalogUtil.newSimpleDataType(TajoDataTypes.Type.INT4);
+      }
+    }
+
+    case TIMESTAMP: {
+      switch (right.getType()) {
+      case INTERVAL: return CatalogUtil.newSimpleDataType(TajoDataTypes.Type.TIMESTAMP);
+      case TIMESTAMP: return CatalogUtil.newSimpleDataType(TajoDataTypes.Type.INTERVAL);
+      }
+    }
+
+    case INTERVAL: {
+      switch (right.getType()) {
+      case INTERVAL:
+      case FLOAT4:
+      case FLOAT8: return CatalogUtil.newSimpleDataType(TajoDataTypes.Type.INTERVAL);
+      }
+    }
+
+    default: return left;
+    }
+  }
+}
diff --git a/tajo-core/src/main/java/org/apache/tajo/master/GlobalEngine.java b/tajo-core/src/main/java/org/apache/tajo/master/GlobalEngine.java
index 8954df1..94c4f25 100644
--- a/tajo-core/src/main/java/org/apache/tajo/master/GlobalEngine.java
+++ b/tajo-core/src/main/java/org/apache/tajo/master/GlobalEngine.java
@@ -471,6 +471,7 @@
       LOG.debug("Non Optimized Query: \n" + plan.toString());
       LOG.debug("=============================================");
     }
+    LOG.info("Non Optimized Query: \n" + plan.toString());
     optimizer.optimize(session, plan);
     LOG.info("=============================================");
     LOG.info("Optimized Query: \n" + plan.toString());
diff --git a/tajo-core/src/test/java/org/apache/tajo/engine/parser/TestSQLAnalyzer.java b/tajo-core/src/test/java/org/apache/tajo/engine/parser/TestSQLAnalyzer.java
index 1dc54f6..23314dd 100644
--- a/tajo-core/src/test/java/org/apache/tajo/engine/parser/TestSQLAnalyzer.java
+++ b/tajo-core/src/test/java/org/apache/tajo/engine/parser/TestSQLAnalyzer.java
@@ -57,7 +57,7 @@
     String result = FileUtil.readTextFileFromResource("results/TestSQLAnalyzer/" + resultFileName);
 
     Expr expr = parseQuery(sql);
-    assertEquals(expr.toJson().trim(), result.trim());
+    assertEquals(result.trim(), expr.toJson().trim());
   }
 
 
diff --git a/tajo-core/src/test/java/org/apache/tajo/engine/query/TestGroupByQuery.java b/tajo-core/src/test/java/org/apache/tajo/engine/query/TestGroupByQuery.java
index b5fd9f1..9f06317 100644
--- a/tajo-core/src/test/java/org/apache/tajo/engine/query/TestGroupByQuery.java
+++ b/tajo-core/src/test/java/org/apache/tajo/engine/query/TestGroupByQuery.java
@@ -280,7 +280,7 @@
         "from table10 group by id");
     String result = resultSetToString(res);
 
-    String expected = "id,?count,?avg_1,?min_2,?max_3,?sum_4,?cast_5,?cast_6,?cast_7,?cast_8\n" +
+    String expected = "id,?count_4,?avg_5,?min_6,?max_7,?sum_8,?cast_9,?cast_10,?cast_11,?cast_12\n" +
         "-------------------------------\n" +
         "1,2,4.0,0,5,12,4,0,5,12\n" +
         "2,3,2.0,0,3,6,7,0,8,21\n";
diff --git a/tajo-core/src/test/java/org/apache/tajo/engine/query/TestWindowQuery.java b/tajo-core/src/test/java/org/apache/tajo/engine/query/TestWindowQuery.java
new file mode 100644
index 0000000..d202153
--- /dev/null
+++ b/tajo-core/src/test/java/org/apache/tajo/engine/query/TestWindowQuery.java
@@ -0,0 +1,240 @@
+/**
+ * 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.query;
+
+import org.apache.tajo.IntegrationTest;
+import org.apache.tajo.QueryTestCaseBase;
+import org.apache.tajo.TajoConstants;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import java.sql.ResultSet;
+
+@Category(IntegrationTest.class)
+public class TestWindowQuery extends QueryTestCaseBase {
+
+  public TestWindowQuery() {
+    super(TajoConstants.DEFAULT_DATABASE_NAME);
+  }
+
+  @Test
+  public final void testWindow1() throws Exception {
+    ResultSet res = executeQuery();
+    assertResultSet(res);
+    cleanupQuery(res);
+  }
+
+  @Test
+  public final void testWindow2() throws Exception {
+    ResultSet res = executeQuery();
+    assertResultSet(res);
+    cleanupQuery(res);
+  }
+
+  @Test
+  public final void testWindow3() throws Exception {
+    ResultSet res = executeQuery();
+    assertResultSet(res);
+    cleanupQuery(res);
+  }
+
+  @Test
+  public final void testWindow4() throws Exception {
+    ResultSet res = executeQuery();
+    assertResultSet(res);
+    cleanupQuery(res);
+  }
+
+  @Test
+  public final void testWindow5() throws Exception {
+    ResultSet res = executeQuery();
+    assertResultSet(res);
+    cleanupQuery(res);
+  }
+
+  @Test
+  public final void testWindow6() throws Exception {
+    ResultSet res = executeQuery();
+    assertResultSet(res);
+    cleanupQuery(res);
+  }
+
+  @Test
+  public final void testWindow7() throws Exception {
+    ResultSet res = executeQuery();
+    assertResultSet(res);
+    cleanupQuery(res);
+  }
+
+  @Test
+  public final void testWindowWithOrderBy1() throws Exception {
+    ResultSet res = executeQuery();
+    assertResultSet(res);
+    cleanupQuery(res);
+  }
+
+  @Test
+  public final void testWindowWithOrderBy2() throws Exception {
+    ResultSet res = executeQuery();
+    assertResultSet(res);
+    cleanupQuery(res);
+  }
+
+  @Test
+  public final void testWindowWithOrderBy3() throws Exception {
+    ResultSet res = executeQuery();
+    assertResultSet(res);
+    cleanupQuery(res);
+  }
+
+  @Test
+  public final void testWindowWithOrderBy4() throws Exception {
+    ResultSet res = executeQuery();
+    assertResultSet(res);
+    cleanupQuery(res);
+  }
+
+  @Test
+  public final void testWindowWithOrderBy5() throws Exception {
+    ResultSet res = executeQuery();
+    assertResultSet(res);
+    cleanupQuery(res);
+  }
+
+  @Test
+  public final void testWindowBeforeLimit() throws Exception {
+    ResultSet res = executeQuery();
+    assertResultSet(res);
+    cleanupQuery(res);
+  }
+
+  @Test
+  public final void testWindowWithSubQuery() throws Exception {
+    ResultSet res = executeQuery();
+    assertResultSet(res);
+    cleanupQuery(res);
+  }
+
+  @Test
+  public final void testWindowWithSubQuery2() throws Exception {
+    ResultSet res = executeQuery();
+    assertResultSet(res);
+    cleanupQuery(res);
+  }
+
+  @Test
+  public final void testWindowWithSubQuery3() throws Exception {
+    ResultSet res = executeQuery();
+    assertResultSet(res);
+    cleanupQuery(res);
+  }
+
+  @Test
+  public final void testWindowWithSubQuery4() throws Exception {
+    ResultSet res = executeQuery();
+    assertResultSet(res);
+    cleanupQuery(res);
+  }
+
+  @Test
+  public final void testWindowWithSubQuery5() throws Exception {
+    // filter push down test
+    ResultSet res = executeQuery();
+    assertResultSet(res);
+    cleanupQuery(res);
+  }
+
+  @Test
+  public final void testWindowWithSubQuery6() throws Exception {
+    // filter push down test
+    ResultSet res = executeQuery();
+    assertResultSet(res);
+    cleanupQuery(res);
+  }
+
+  @Test
+  public final void testWindowWithAggregation1() throws Exception {
+    ResultSet res = executeQuery();
+    assertResultSet(res);
+    cleanupQuery(res);
+  }
+
+  @Test
+  public final void testWindowWithAggregation2() throws Exception {
+    ResultSet res = executeQuery();
+    assertResultSet(res);
+    cleanupQuery(res);
+  }
+
+  @Test
+  public final void testWindowWithAggregation3() throws Exception {
+    ResultSet res = executeQuery();
+    assertResultSet(res);
+    cleanupQuery(res);
+  }
+
+  @Test
+  public final void testWindowWithAggregation4() throws Exception {
+    ResultSet res = executeQuery();
+    assertResultSet(res);
+    cleanupQuery(res);
+  }
+
+  @Test
+  public final void testWindowWithAggregation5() throws Exception {
+    ResultSet res = executeQuery();
+    assertResultSet(res);
+    cleanupQuery(res);
+  }
+
+  @Test
+  public final void testWindowWithAggregation6() throws Exception {
+    ResultSet res = executeQuery();
+    assertResultSet(res);
+    cleanupQuery(res);
+  }
+
+  @Test
+  public final void testComplexOrderBy1() throws Exception {
+    ResultSet res = executeQuery();
+    assertResultSet(res);
+    cleanupQuery(res);
+  }
+
+  @Test
+  public final void rowNumber1() throws Exception {
+    ResultSet res = executeQuery();
+    assertResultSet(res);
+    cleanupQuery(res);
+  }
+
+  @Test
+  public final void rowNumber2() throws Exception {
+    ResultSet res = executeQuery();
+    assertResultSet(res);
+    cleanupQuery(res);
+  }
+
+  @Test
+  public final void rowNumber3() throws Exception {
+    ResultSet res = executeQuery();
+    assertResultSet(res);
+    cleanupQuery(res);
+  }
+}
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/queries/TestLogicalPlanner/window1.sql b/tajo-core/src/test/resources/queries/TestLogicalPlanner/window1.sql
new file mode 100644
index 0000000..9d30145
--- /dev/null
+++ b/tajo-core/src/test/resources/queries/TestLogicalPlanner/window1.sql
@@ -0,0 +1,5 @@
+ SELECT
+  l_orderkey,
+  sum(l_partkey) over()
+FROM
+  lineitem;
\ No newline at end of file
diff --git a/tajo-core/tajo-core-backend/src/test/resources/queries/TestSQLAnalyzer/window1.sql b/tajo-core/src/test/resources/queries/TestSQLAnalyzer/window1.sql
similarity index 100%
rename from tajo-core/tajo-core-backend/src/test/resources/queries/TestSQLAnalyzer/window1.sql
rename to tajo-core/src/test/resources/queries/TestSQLAnalyzer/window1.sql
diff --git a/tajo-core/tajo-core-backend/src/test/resources/queries/TestSQLAnalyzer/window2.sql b/tajo-core/src/test/resources/queries/TestSQLAnalyzer/window2.sql
similarity index 100%
rename from tajo-core/tajo-core-backend/src/test/resources/queries/TestSQLAnalyzer/window2.sql
rename to tajo-core/src/test/resources/queries/TestSQLAnalyzer/window2.sql
diff --git a/tajo-core/tajo-core-backend/src/test/resources/queries/TestSQLAnalyzer/window3.sql b/tajo-core/src/test/resources/queries/TestSQLAnalyzer/window3.sql
similarity index 100%
rename from tajo-core/tajo-core-backend/src/test/resources/queries/TestSQLAnalyzer/window3.sql
rename to tajo-core/src/test/resources/queries/TestSQLAnalyzer/window3.sql
diff --git a/tajo-core/tajo-core-backend/src/test/resources/queries/TestSQLAnalyzer/window4.sql b/tajo-core/src/test/resources/queries/TestSQLAnalyzer/window4.sql
similarity index 100%
rename from tajo-core/tajo-core-backend/src/test/resources/queries/TestSQLAnalyzer/window4.sql
rename to tajo-core/src/test/resources/queries/TestSQLAnalyzer/window4.sql
diff --git a/tajo-core/tajo-core-backend/src/test/resources/queries/TestSQLAnalyzer/window5.sql b/tajo-core/src/test/resources/queries/TestSQLAnalyzer/window5.sql
similarity index 100%
rename from tajo-core/tajo-core-backend/src/test/resources/queries/TestSQLAnalyzer/window5.sql
rename to tajo-core/src/test/resources/queries/TestSQLAnalyzer/window5.sql
diff --git a/tajo-core/tajo-core-backend/src/test/resources/queries/TestSQLAnalyzer/window6.sql b/tajo-core/src/test/resources/queries/TestSQLAnalyzer/window6.sql
similarity index 100%
rename from tajo-core/tajo-core-backend/src/test/resources/queries/TestSQLAnalyzer/window6.sql
rename to tajo-core/src/test/resources/queries/TestSQLAnalyzer/window6.sql
diff --git a/tajo-core/tajo-core-backend/src/test/resources/queries/TestSQLAnalyzer/window7.sql b/tajo-core/src/test/resources/queries/TestSQLAnalyzer/window7.sql
similarity index 100%
rename from tajo-core/tajo-core-backend/src/test/resources/queries/TestSQLAnalyzer/window7.sql
rename to tajo-core/src/test/resources/queries/TestSQLAnalyzer/window7.sql
diff --git a/tajo-core/tajo-core-backend/src/test/resources/queries/TestSQLAnalyzer/window8.sql b/tajo-core/src/test/resources/queries/TestSQLAnalyzer/window8.sql
similarity index 100%
rename from tajo-core/tajo-core-backend/src/test/resources/queries/TestSQLAnalyzer/window8.sql
rename to tajo-core/src/test/resources/queries/TestSQLAnalyzer/window8.sql
diff --git a/tajo-core/tajo-core-backend/src/test/resources/queries/TestSQLAnalyzer/window9.sql b/tajo-core/src/test/resources/queries/TestSQLAnalyzer/window9.sql
similarity index 100%
rename from tajo-core/tajo-core-backend/src/test/resources/queries/TestSQLAnalyzer/window9.sql
rename to tajo-core/src/test/resources/queries/TestSQLAnalyzer/window9.sql
diff --git a/tajo-core/src/test/resources/queries/TestWindowQuery/rowNumber1.sql b/tajo-core/src/test/resources/queries/TestWindowQuery/rowNumber1.sql
new file mode 100644
index 0000000..14a5128
--- /dev/null
+++ b/tajo-core/src/test/resources/queries/TestWindowQuery/rowNumber1.sql
@@ -0,0 +1,5 @@
+SELECT
+  l_orderkey,
+  row_number() OVER () as row_num
+FROM
+  LINEITEM;
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/queries/TestWindowQuery/rowNumber2.sql b/tajo-core/src/test/resources/queries/TestWindowQuery/rowNumber2.sql
new file mode 100644
index 0000000..2e45120
--- /dev/null
+++ b/tajo-core/src/test/resources/queries/TestWindowQuery/rowNumber2.sql
@@ -0,0 +1,5 @@
+SELECT
+  l_orderkey,
+  row_number() OVER (PARTITION BY L_ORDERKEY) as row_num
+FROM
+  LINEITEM
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/queries/TestWindowQuery/rowNumber3.sql b/tajo-core/src/test/resources/queries/TestWindowQuery/rowNumber3.sql
new file mode 100644
index 0000000..44594c7
--- /dev/null
+++ b/tajo-core/src/test/resources/queries/TestWindowQuery/rowNumber3.sql
@@ -0,0 +1,7 @@
+SELECT
+  l_orderkey,
+  row_number() OVER (PARTITION BY L_ORDERKEY) as row_num,
+  l_discount,
+  avg(l_discount) OVER (PARTITION BY L_ORDERKEY) as average
+FROM
+  LINEITEM
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/queries/TestWindowQuery/testComplexOrderBy1.sql b/tajo-core/src/test/resources/queries/TestWindowQuery/testComplexOrderBy1.sql
new file mode 100644
index 0000000..171815c
--- /dev/null
+++ b/tajo-core/src/test/resources/queries/TestWindowQuery/testComplexOrderBy1.sql
@@ -0,0 +1,5 @@
+select
+  l_orderkey,
+  row_number() over (order by l_quantity * (1 - l_discount)) row_num
+from
+  lineitem
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/queries/TestWindowQuery/testWindow1.sql b/tajo-core/src/test/resources/queries/TestWindowQuery/testWindow1.sql
new file mode 100644
index 0000000..8e6d083
--- /dev/null
+++ b/tajo-core/src/test/resources/queries/TestWindowQuery/testWindow1.sql
@@ -0,0 +1 @@
+SELECT sum(l_quantity) OVER () FROM LINEITEM;
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/queries/TestWindowQuery/testWindow2.sql b/tajo-core/src/test/resources/queries/TestWindowQuery/testWindow2.sql
new file mode 100644
index 0000000..baa92f6
--- /dev/null
+++ b/tajo-core/src/test/resources/queries/TestWindowQuery/testWindow2.sql
@@ -0,0 +1 @@
+SELECT l_orderkey, l_quantity, sum(l_quantity) OVER () FROM LINEITEM;
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/queries/TestWindowQuery/testWindow3.sql b/tajo-core/src/test/resources/queries/TestWindowQuery/testWindow3.sql
new file mode 100644
index 0000000..e4b92f0
--- /dev/null
+++ b/tajo-core/src/test/resources/queries/TestWindowQuery/testWindow3.sql
@@ -0,0 +1 @@
+SELECT l_orderkey, l_quantity, sum(l_quantity) OVER (PARTITION BY l_orderkey) FROM LINEITEM;
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/queries/TestWindowQuery/testWindow4.sql b/tajo-core/src/test/resources/queries/TestWindowQuery/testWindow4.sql
new file mode 100644
index 0000000..d32c63a
--- /dev/null
+++ b/tajo-core/src/test/resources/queries/TestWindowQuery/testWindow4.sql
@@ -0,0 +1 @@
+SELECT l_orderkey, l_discount, sum(l_discount) OVER (PARTITION BY l_orderkey), sum(l_quantity) OVER (PARTITION BY l_orderkey) FROM LINEITEM;
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/queries/TestWindowQuery/testWindow5.sql b/tajo-core/src/test/resources/queries/TestWindowQuery/testWindow5.sql
new file mode 100644
index 0000000..7f31991
--- /dev/null
+++ b/tajo-core/src/test/resources/queries/TestWindowQuery/testWindow5.sql
@@ -0,0 +1 @@
+SELECT l_orderkey, sum(l_discount) OVER (PARTITION BY l_orderkey), l_discount, sum(l_quantity) OVER (PARTITION BY l_orderkey) FROM LINEITEM;
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/queries/TestWindowQuery/testWindow6.sql b/tajo-core/src/test/resources/queries/TestWindowQuery/testWindow6.sql
new file mode 100644
index 0000000..c1e4a84
--- /dev/null
+++ b/tajo-core/src/test/resources/queries/TestWindowQuery/testWindow6.sql
@@ -0,0 +1 @@
+SELECT l_orderkey, l_discount, row_number() OVER (PARTITION BY l_orderkey) r1 , sum(l_discount) OVER (PARTITION BY l_orderkey) r2 FROM LINEITEM;
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/queries/TestWindowQuery/testWindow7.sql b/tajo-core/src/test/resources/queries/TestWindowQuery/testWindow7.sql
new file mode 100644
index 0000000..2caf0db
--- /dev/null
+++ b/tajo-core/src/test/resources/queries/TestWindowQuery/testWindow7.sql
@@ -0,0 +1 @@
+select l_orderkey, l_quantity, rank() over (partition by l_orderkey) as r from lineitem
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowBeforeLimit.sql b/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowBeforeLimit.sql
new file mode 100644
index 0000000..c4d7fa6
--- /dev/null
+++ b/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowBeforeLimit.sql
@@ -0,0 +1,6 @@
+select
+  r_name,
+  rank() over (order by r_regionkey) as ran
+from
+  region
+limit 3;
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowWithAggregation1.sql b/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowWithAggregation1.sql
new file mode 100644
index 0000000..7afdae2
--- /dev/null
+++ b/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowWithAggregation1.sql
@@ -0,0 +1,4 @@
+select
+  row_number() over (order by count(*) desc) row_num
+from
+  lineitem
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowWithAggregation2.sql b/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowWithAggregation2.sql
new file mode 100644
index 0000000..1d77ca1
--- /dev/null
+++ b/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowWithAggregation2.sql
@@ -0,0 +1,7 @@
+select
+  l_orderkey,
+  row_number() over (partition by l_orderkey order by count(*) desc) row_num
+from
+  lineitem
+group by
+  l_orderkey;
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowWithAggregation3.sql b/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowWithAggregation3.sql
new file mode 100644
index 0000000..f3d3703
--- /dev/null
+++ b/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowWithAggregation3.sql
@@ -0,0 +1,5 @@
+select
+  count(*) as cnt,
+  row_number() over (order by count(*) desc) row_num
+from
+  lineitem
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowWithAggregation4.sql b/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowWithAggregation4.sql
new file mode 100644
index 0000000..af695ba
--- /dev/null
+++ b/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowWithAggregation4.sql
@@ -0,0 +1,8 @@
+select
+  l_orderkey,
+  count(*) as cnt,
+  row_number() over (order by count(*) desc) row_num
+from
+  lineitem
+group by
+  l_orderkey
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowWithAggregation5.sql b/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowWithAggregation5.sql
new file mode 100644
index 0000000..2764ae9
--- /dev/null
+++ b/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowWithAggregation5.sql
@@ -0,0 +1,9 @@
+select
+  l_orderkey,
+  count(*) as cnt,
+  row_number() over (partition by l_orderkey order by count(*) desc)
+  row_num
+from
+  lineitem
+group by
+  l_orderkey
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowWithAggregation6.sql b/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowWithAggregation6.sql
new file mode 100644
index 0000000..1a9d46b
--- /dev/null
+++ b/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowWithAggregation6.sql
@@ -0,0 +1,10 @@
+select
+  l_orderkey,
+  count(*) as cnt,
+  row_number() over (order by count(*) desc) row_num
+from
+  lineitem
+group by
+  l_orderkey
+order by
+  l_orderkey;
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowWithOrderBy1.sql b/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowWithOrderBy1.sql
new file mode 100644
index 0000000..d047bbd
--- /dev/null
+++ b/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowWithOrderBy1.sql
@@ -0,0 +1,6 @@
+SELECT
+  l_orderkey,
+  l_discount,
+  rank() OVER (ORDER BY l_discount) r1
+FROM
+  LINEITEM;
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowWithOrderBy2.sql b/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowWithOrderBy2.sql
new file mode 100644
index 0000000..7e6b339
--- /dev/null
+++ b/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowWithOrderBy2.sql
@@ -0,0 +1,6 @@
+SELECT
+  l_orderkey,
+  l_partkey,
+  rank() OVER (PARTITION BY L_ORDERKEY ORDER BY l_partkey) r1
+FROM
+  LINEITEM;
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowWithOrderBy3.sql b/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowWithOrderBy3.sql
new file mode 100644
index 0000000..5fc7317
--- /dev/null
+++ b/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowWithOrderBy3.sql
@@ -0,0 +1,6 @@
+SELECT
+  l_orderkey,
+  l_partkey,
+  rank() OVER (PARTITION BY L_ORDERKEY ORDER BY l_partkey desc) r1
+FROM
+  LINEITEM
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowWithOrderBy4.sql b/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowWithOrderBy4.sql
new file mode 100644
index 0000000..26cfaa7
--- /dev/null
+++ b/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowWithOrderBy4.sql
@@ -0,0 +1,7 @@
+SELECT
+  l_orderkey,
+  l_partkey,
+  rank() OVER (ORDER BY l_orderkey) r1,
+  rank() OVER(ORDER BY l_partkey desc) r2
+FROM
+  LINEITEM;
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowWithOrderBy5.sql b/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowWithOrderBy5.sql
new file mode 100644
index 0000000..2dc87af
--- /dev/null
+++ b/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowWithOrderBy5.sql
@@ -0,0 +1,9 @@
+SELECT
+  l_orderkey,
+  l_partkey,
+  rank() OVER (ORDER BY l_orderkey) r1,
+  rank() OVER(ORDER BY l_partkey desc) r2
+FROM
+  LINEITEM
+where
+  l_partkey > 0 and l_partkey < 100;
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowWithSubQuery.sql b/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowWithSubQuery.sql
new file mode 100644
index 0000000..5fa6d00
--- /dev/null
+++ b/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowWithSubQuery.sql
@@ -0,0 +1,14 @@
+select
+  r_name,
+  c,
+  rank() over (order by r_regionkey) as ran
+from (
+  select
+    r_name,
+    r_regionkey,
+    count(*) as c
+  from
+    region
+  group by
+    r_name, r_regionkey
+) a;
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowWithSubQuery2.sql b/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowWithSubQuery2.sql
new file mode 100644
index 0000000..87520fb
--- /dev/null
+++ b/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowWithSubQuery2.sql
@@ -0,0 +1,15 @@
+select
+  r_name,
+  c,
+  rank() over (partition by r_regionkey order by r_regionkey) as ran
+from (
+  select
+    r_name,
+    r_regionkey,
+    count(*) as c
+  from
+    region
+  group by
+    r_name, r_regionkey
+) a
+limit 3;
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowWithSubQuery3.sql b/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowWithSubQuery3.sql
new file mode 100644
index 0000000..10d8863
--- /dev/null
+++ b/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowWithSubQuery3.sql
@@ -0,0 +1,14 @@
+select
+  a.r_name,
+  a.r_regionkey,
+  row_number() over (partition by a.r_name order by a.cnt desc) mk
+from (
+  select
+    r_name,
+    r_regionkey,
+    count(*) cnt
+  from
+    default.region
+  group by
+    r_name, r_regionkey
+) a;
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowWithSubQuery4.sql b/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowWithSubQuery4.sql
new file mode 100644
index 0000000..fe32eb1
--- /dev/null
+++ b/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowWithSubQuery4.sql
@@ -0,0 +1,14 @@
+select
+  a.r_name,
+  a.r_regionkey,
+  row_number() over (partition by a.r_regionkey order by a.cnt desc) mk
+from (
+  select
+    r_name,
+    r_regionkey,
+    count(*) cnt
+  from
+    default.region
+  group by
+    r_name, r_regionkey
+) a;
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowWithSubQuery5.sql b/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowWithSubQuery5.sql
new file mode 100644
index 0000000..2db397b
--- /dev/null
+++ b/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowWithSubQuery5.sql
@@ -0,0 +1,11 @@
+select
+  *
+from (
+  select
+    r_name,
+    rank() over (order by r_regionkey) as ran
+  from
+    region
+) a
+where
+  ran >= 3;
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowWithSubQuery6.sql b/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowWithSubQuery6.sql
new file mode 100644
index 0000000..8678fe6
--- /dev/null
+++ b/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowWithSubQuery6.sql
@@ -0,0 +1,9 @@
+select
+  *
+from (
+  select
+    r_name,
+    rank() over (order by r_regionkey) as ran from region
+) a
+where
+  r_name LIKE 'ASIA'
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/results/TestSQLAnalyzer/window1.result b/tajo-core/src/test/resources/results/TestSQLAnalyzer/window1.result
new file mode 100644
index 0000000..a21e755
--- /dev/null
+++ b/tajo-core/src/test/resources/results/TestSQLAnalyzer/window1.result
@@ -0,0 +1,37 @@
+{
+  "IsDistinct": false,
+  "Projections": [
+    {
+      "Expr": {
+        "ColumnName": "dt",
+        "OpType": "Column"
+      },
+      "OpType": "Target"
+    },
+    {
+      "Expr": {
+        "WindowSpec": {},
+        "IsDistinct": false,
+        "Signature": "sum",
+        "FuncParams": [
+          {
+            "ColumnName": "xy",
+            "OpType": "Column"
+          }
+        ],
+        "OpType": "WindowFunction"
+      },
+      "OpType": "Target"
+    }
+  ],
+  "Expr": {
+    "Relations": [
+      {
+        "TableName": "sum_example",
+        "OpType": "Relation"
+      }
+    ],
+    "OpType": "RelationList"
+  },
+  "OpType": "Projection"
+}
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/results/TestSQLAnalyzer/window2.result b/tajo-core/src/test/resources/results/TestSQLAnalyzer/window2.result
new file mode 100644
index 0000000..3708802
--- /dev/null
+++ b/tajo-core/src/test/resources/results/TestSQLAnalyzer/window2.result
@@ -0,0 +1,44 @@
+{
+  "IsDistinct": false,
+  "Projections": [
+    {
+      "Expr": {
+        "ColumnName": "dt",
+        "OpType": "Column"
+      },
+      "OpType": "Target"
+    },
+    {
+      "Expr": {
+        "WindowSpec": {
+          "partitionKeys": [
+            {
+              "ColumnName": "dt",
+              "OpType": "Column"
+            }
+          ]
+        },
+        "IsDistinct": false,
+        "Signature": "sum",
+        "FuncParams": [
+          {
+            "ColumnName": "xy",
+            "OpType": "Column"
+          }
+        ],
+        "OpType": "WindowFunction"
+      },
+      "OpType": "Target"
+    }
+  ],
+  "Expr": {
+    "Relations": [
+      {
+        "TableName": "sum_example",
+        "OpType": "Relation"
+      }
+    ],
+    "OpType": "RelationList"
+  },
+  "OpType": "Projection"
+}
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/results/TestSQLAnalyzer/window3.result b/tajo-core/src/test/resources/results/TestSQLAnalyzer/window3.result
new file mode 100644
index 0000000..fca2651
--- /dev/null
+++ b/tajo-core/src/test/resources/results/TestSQLAnalyzer/window3.result
@@ -0,0 +1,50 @@
+{
+  "IsDistinct": false,
+  "Projections": [
+    {
+      "Expr": {
+        "ColumnName": "dt",
+        "OpType": "Column"
+      },
+      "OpType": "Target"
+    },
+    {
+      "Expr": {
+        "WindowSpec": {
+          "partitionKeys": [
+            {
+              "Signature": "round",
+              "FuncParams": [
+                {
+                  "ColumnName": "dt",
+                  "OpType": "Column"
+                }
+              ],
+              "OpType": "Function"
+            }
+          ]
+        },
+        "IsDistinct": false,
+        "Signature": "sum",
+        "FuncParams": [
+          {
+            "ColumnName": "xy",
+            "OpType": "Column"
+          }
+        ],
+        "OpType": "WindowFunction"
+      },
+      "OpType": "Target"
+    }
+  ],
+  "Expr": {
+    "Relations": [
+      {
+        "TableName": "sum_example",
+        "OpType": "Relation"
+      }
+    ],
+    "OpType": "RelationList"
+  },
+  "OpType": "Projection"
+}
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/results/TestSQLAnalyzer/window4.result b/tajo-core/src/test/resources/results/TestSQLAnalyzer/window4.result
new file mode 100644
index 0000000..45ccdcc
--- /dev/null
+++ b/tajo-core/src/test/resources/results/TestSQLAnalyzer/window4.result
@@ -0,0 +1,48 @@
+{
+  "IsDistinct": false,
+  "Projections": [
+    {
+      "Expr": {
+        "ColumnName": "dt",
+        "OpType": "Column"
+      },
+      "OpType": "Target"
+    },
+    {
+      "Expr": {
+        "WindowSpec": {
+          "sortSpecs": [
+            {
+              "SortKey": {
+                "ColumnName": "dt",
+                "OpType": "Column"
+              },
+              "IsAsc": true,
+              "IsNullFirst": false
+            }
+          ]
+        },
+        "IsDistinct": false,
+        "Signature": "sum",
+        "FuncParams": [
+          {
+            "ColumnName": "xy",
+            "OpType": "Column"
+          }
+        ],
+        "OpType": "WindowFunction"
+      },
+      "OpType": "Target"
+    }
+  ],
+  "Expr": {
+    "Relations": [
+      {
+        "TableName": "sum_example",
+        "OpType": "Relation"
+      }
+    ],
+    "OpType": "RelationList"
+  },
+  "OpType": "Projection"
+}
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/results/TestSQLAnalyzer/window5.result b/tajo-core/src/test/resources/results/TestSQLAnalyzer/window5.result
new file mode 100644
index 0000000..aa00a09
--- /dev/null
+++ b/tajo-core/src/test/resources/results/TestSQLAnalyzer/window5.result
@@ -0,0 +1,77 @@
+{
+  "IsDistinct": false,
+  "Projections": [
+    {
+      "Expr": {
+        "ColumnName": "dt",
+        "OpType": "Column"
+      },
+      "OpType": "Target"
+    },
+    {
+      "Expr": {
+        "ColumnName": "dt2",
+        "OpType": "Column"
+      },
+      "OpType": "Target"
+    },
+    {
+      "Expr": {
+        "WindowSpec": {
+          "partitionKeys": [
+            {
+              "Signature": "round",
+              "FuncParams": [
+                {
+                  "ColumnName": "dt",
+                  "OpType": "Column"
+                }
+              ],
+              "OpType": "Function"
+            },
+            {
+              "ColumnName": "dt2",
+              "OpType": "Column"
+            }
+          ],
+          "sortSpecs": [
+            {
+              "SortKey": {
+                "Signature": "ceil",
+                "FuncParams": [
+                  {
+                    "ColumnName": "dt",
+                    "OpType": "Column"
+                  }
+                ],
+                "OpType": "Function"
+              },
+              "IsAsc": true,
+              "IsNullFirst": false
+            }
+          ]
+        },
+        "IsDistinct": false,
+        "Signature": "sum",
+        "FuncParams": [
+          {
+            "ColumnName": "xy",
+            "OpType": "Column"
+          }
+        ],
+        "OpType": "WindowFunction"
+      },
+      "OpType": "Target"
+    }
+  ],
+  "Expr": {
+    "Relations": [
+      {
+        "TableName": "sum_example",
+        "OpType": "Relation"
+      }
+    ],
+    "OpType": "RelationList"
+  },
+  "OpType": "Projection"
+}
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/results/TestSQLAnalyzer/window6.result b/tajo-core/src/test/resources/results/TestSQLAnalyzer/window6.result
new file mode 100644
index 0000000..2c990c6
--- /dev/null
+++ b/tajo-core/src/test/resources/results/TestSQLAnalyzer/window6.result
@@ -0,0 +1,83 @@
+{
+  "IsDistinct": false,
+  "Projections": [
+    {
+      "Expr": {
+        "ColumnName": "dt",
+        "OpType": "Column"
+      },
+      "OpType": "Target"
+    },
+    {
+      "Expr": {
+        "ColumnName": "dt2",
+        "OpType": "Column"
+      },
+      "OpType": "Target"
+    },
+    {
+      "Expr": {
+        "WindowSpec": {
+          "partitionKeys": [
+            {
+              "Signature": "round",
+              "FuncParams": [
+                {
+                  "ColumnName": "dt",
+                  "OpType": "Column"
+                }
+              ],
+              "OpType": "Function"
+            },
+            {
+              "ColumnName": "dt2",
+              "OpType": "Column"
+            }
+          ],
+          "sortSpecs": [
+            {
+              "SortKey": {
+                "Signature": "ceil",
+                "FuncParams": [
+                  {
+                    "ColumnName": "dt",
+                    "OpType": "Column"
+                  }
+                ],
+                "OpType": "Function"
+              },
+              "IsAsc": true,
+              "IsNullFirst": false
+            }
+          ],
+          "windowFrame": {
+            "unit": "ROW",
+            "startBound": {
+              "boundType": "UNBOUNDED_PRECEDING"
+            }
+          }
+        },
+        "IsDistinct": false,
+        "Signature": "sum",
+        "FuncParams": [
+          {
+            "ColumnName": "xy",
+            "OpType": "Column"
+          }
+        ],
+        "OpType": "WindowFunction"
+      },
+      "OpType": "Target"
+    }
+  ],
+  "Expr": {
+    "Relations": [
+      {
+        "TableName": "sum_example",
+        "OpType": "Relation"
+      }
+    ],
+    "OpType": "RelationList"
+  },
+  "OpType": "Projection"
+}
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/results/TestSQLAnalyzer/window7.result b/tajo-core/src/test/resources/results/TestSQLAnalyzer/window7.result
new file mode 100644
index 0000000..efb3878
--- /dev/null
+++ b/tajo-core/src/test/resources/results/TestSQLAnalyzer/window7.result
@@ -0,0 +1,81 @@
+{
+  "IsDistinct": false,
+  "Projections": [
+    {
+      "Expr": {
+        "ColumnName": "dt",
+        "OpType": "Column"
+      },
+      "OpType": "Target"
+    },
+    {
+      "Expr": {
+        "ColumnName": "dt2",
+        "OpType": "Column"
+      },
+      "OpType": "Target"
+    },
+    {
+      "Expr": {
+        "WindowSpec": {
+          "partitionKeys": [
+            {
+              "Signature": "round",
+              "FuncParams": [
+                {
+                  "ColumnName": "dt",
+                  "OpType": "Column"
+                }
+              ],
+              "OpType": "Function"
+            },
+            {
+              "ColumnName": "dt2",
+              "OpType": "Column"
+            }
+          ],
+          "sortSpecs": [
+            {
+              "SortKey": {
+                "Signature": "ceil",
+                "FuncParams": [
+                  {
+                    "ColumnName": "dt",
+                    "OpType": "Column"
+                  }
+                ],
+                "OpType": "Function"
+              },
+              "IsAsc": true,
+              "IsNullFirst": false
+            }
+          ],
+          "windowFrame": {
+            "unit": "ROW",
+            "startBound": {
+              "boundType": "UNBOUNDED_PRECEDING"
+            },
+            "endBound": {
+              "boundType": "UNBOUNDED_FOLLOWING"
+            }
+          }
+        },
+        "IsDistinct": false,
+        "Signature": "row_number",
+        "FuncParams": [],
+        "OpType": "WindowFunction"
+      },
+      "OpType": "Target"
+    }
+  ],
+  "Expr": {
+    "Relations": [
+      {
+        "TableName": "sum_example",
+        "OpType": "Relation"
+      }
+    ],
+    "OpType": "RelationList"
+  },
+  "OpType": "Projection"
+}
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/results/TestSQLAnalyzer/window8.result b/tajo-core/src/test/resources/results/TestSQLAnalyzer/window8.result
new file mode 100644
index 0000000..05b95a4
--- /dev/null
+++ b/tajo-core/src/test/resources/results/TestSQLAnalyzer/window8.result
@@ -0,0 +1,86 @@
+{
+  "IsDistinct": false,
+  "Projections": [
+    {
+      "Expr": {
+        "ColumnName": "dt",
+        "OpType": "Column"
+      },
+      "OpType": "Target"
+    },
+    {
+      "Expr": {
+        "ColumnName": "dt2",
+        "OpType": "Column"
+      },
+      "OpType": "Target"
+    },
+    {
+      "Expr": {
+        "WindowSpec": {
+          "partitionKeys": [
+            {
+              "Signature": "round",
+              "FuncParams": [
+                {
+                  "ColumnName": "dt",
+                  "OpType": "Column"
+                }
+              ],
+              "OpType": "Function"
+            },
+            {
+              "ColumnName": "dt2",
+              "OpType": "Column"
+            }
+          ],
+          "sortSpecs": [
+            {
+              "SortKey": {
+                "Signature": "ceil",
+                "FuncParams": [
+                  {
+                    "ColumnName": "dt",
+                    "OpType": "Column"
+                  }
+                ],
+                "OpType": "Function"
+              },
+              "IsAsc": true,
+              "IsNullFirst": false
+            }
+          ],
+          "windowFrame": {
+            "unit": "ROW",
+            "startBound": {
+              "boundType": "PRECEDING",
+              "number": {
+                "Value": "1",
+                "ValueType": "Unsigned_Integer",
+                "OpType": "Literal"
+              }
+            },
+            "endBound": {
+              "boundType": "CURRENT_ROW"
+            }
+          }
+        },
+        "IsDistinct": false,
+        "Signature": "row_number",
+        "FuncParams": [],
+        "OpType": "WindowFunction"
+      },
+      "OpType": "Target"
+    }
+  ],
+  "Expr": {
+    "Relations": [
+      {
+        "TableName": "sum_example",
+        "OpType": "Relation"
+      }
+    ],
+    "OpType": "RelationList"
+  },
+  "OpType": "Projection"
+}
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/results/TestSQLAnalyzer/window9.result b/tajo-core/src/test/resources/results/TestSQLAnalyzer/window9.result
new file mode 100644
index 0000000..d1f52dc
--- /dev/null
+++ b/tajo-core/src/test/resources/results/TestSQLAnalyzer/window9.result
@@ -0,0 +1,56 @@
+{
+  "IsDistinct": false,
+  "Projections": [
+    {
+      "Expr": {
+        "ColumnName": "dt",
+        "OpType": "Column"
+      },
+      "OpType": "Target"
+    },
+    {
+      "Expr": {
+        "ColumnName": "dt2",
+        "OpType": "Column"
+      },
+      "OpType": "Target"
+    },
+    {
+      "Expr": {
+        "WindowSpec": {
+          "windowName": "window1"
+        },
+        "IsDistinct": false,
+        "Signature": "row_number",
+        "FuncParams": [],
+        "OpType": "WindowFunction"
+      },
+      "OpType": "Target"
+    },
+    {
+      "Expr": {
+        "WindowSpec": {
+          "windowName": "window1"
+        },
+        "IsDistinct": false,
+        "Signature": "rank",
+        "FuncParams": [],
+        "OpType": "WindowFunction"
+      },
+      "OpType": "Target"
+    }
+  ],
+  "Expr": {
+    "Expr": {
+      "Relations": [
+        {
+          "TableName": "sum_example",
+          "OpType": "Relation"
+        }
+      ],
+      "OpType": "RelationList"
+    },
+    "OpType": "Window"
+  },
+  "OpType": "Projection"
+}
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/results/TestSelectQuery/testSumFloatOverflow.result b/tajo-core/src/test/resources/results/TestSelectQuery/testSumFloatOverflow.result
index 13b9ef4..a25bbbc 100644
--- a/tajo-core/src/test/resources/results/TestSelectQuery/testSumFloatOverflow.result
+++ b/tajo-core/src/test/resources/results/TestSelectQuery/testSumFloatOverflow.result
@@ -1,3 +1,3 @@
-?sum
+?sum_1
 -------------------------------
 6.838452478692677E38
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/results/TestSelectQuery/testSumIntOverflow.result b/tajo-core/src/test/resources/results/TestSelectQuery/testSumIntOverflow.result
index cf2e0a8..f288f86 100644
--- a/tajo-core/src/test/resources/results/TestSelectQuery/testSumIntOverflow.result
+++ b/tajo-core/src/test/resources/results/TestSelectQuery/testSumIntOverflow.result
@@ -1,3 +1,3 @@
-?sum
+?sum_1
 -------------------------------
 4673934905
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/results/TestWindowQuery/rowNumber1.result b/tajo-core/src/test/resources/results/TestWindowQuery/rowNumber1.result
new file mode 100644
index 0000000..5fc49ee
--- /dev/null
+++ b/tajo-core/src/test/resources/results/TestWindowQuery/rowNumber1.result
@@ -0,0 +1,7 @@
+l_orderkey,row_num
+-------------------------------
+1,1
+1,2
+2,3
+3,4
+3,5
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/results/TestWindowQuery/rowNumber2.result b/tajo-core/src/test/resources/results/TestWindowQuery/rowNumber2.result
new file mode 100644
index 0000000..db02a76
--- /dev/null
+++ b/tajo-core/src/test/resources/results/TestWindowQuery/rowNumber2.result
@@ -0,0 +1,7 @@
+l_orderkey,row_num
+-------------------------------
+1,1
+1,2
+2,1
+3,1
+3,2
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/results/TestWindowQuery/rowNumber3.result b/tajo-core/src/test/resources/results/TestWindowQuery/rowNumber3.result
new file mode 100644
index 0000000..1d780ff
--- /dev/null
+++ b/tajo-core/src/test/resources/results/TestWindowQuery/rowNumber3.result
@@ -0,0 +1,7 @@
+l_orderkey,row_num,l_discount,average
+-------------------------------
+1,1,0.04,0.065
+1,2,0.09,0.065
+2,1,0.0,0.0
+3,1,0.06,0.08
+3,2,0.1,0.08
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/results/TestWindowQuery/testComplexOrderBy1.result b/tajo-core/src/test/resources/results/TestWindowQuery/testComplexOrderBy1.result
new file mode 100644
index 0000000..5fc49ee
--- /dev/null
+++ b/tajo-core/src/test/resources/results/TestWindowQuery/testComplexOrderBy1.result
@@ -0,0 +1,7 @@
+l_orderkey,row_num
+-------------------------------
+1,1
+1,2
+2,3
+3,4
+3,5
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/results/TestWindowQuery/testWindow1.result b/tajo-core/src/test/resources/results/TestWindowQuery/testWindow1.result
new file mode 100644
index 0000000..f142256
--- /dev/null
+++ b/tajo-core/src/test/resources/results/TestWindowQuery/testWindow1.result
@@ -0,0 +1,7 @@
+?windowfunction
+-------------------------------
+185.0
+185.0
+185.0
+185.0
+185.0
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/results/TestWindowQuery/testWindow2.result b/tajo-core/src/test/resources/results/TestWindowQuery/testWindow2.result
new file mode 100644
index 0000000..136d66c
--- /dev/null
+++ b/tajo-core/src/test/resources/results/TestWindowQuery/testWindow2.result
@@ -0,0 +1,7 @@
+l_orderkey,l_quantity,?windowfunction
+-------------------------------
+1,17.0,185.0
+1,36.0,185.0
+2,38.0,185.0
+3,45.0,185.0
+3,49.0,185.0
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/results/TestWindowQuery/testWindow3.result b/tajo-core/src/test/resources/results/TestWindowQuery/testWindow3.result
new file mode 100644
index 0000000..8262c13
--- /dev/null
+++ b/tajo-core/src/test/resources/results/TestWindowQuery/testWindow3.result
@@ -0,0 +1,7 @@
+l_orderkey,l_quantity,?windowfunction
+-------------------------------
+1,17.0,53.0
+1,36.0,53.0
+2,38.0,38.0
+3,45.0,94.0
+3,49.0,94.0
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/results/TestWindowQuery/testWindow4.result b/tajo-core/src/test/resources/results/TestWindowQuery/testWindow4.result
new file mode 100644
index 0000000..3477c5b
--- /dev/null
+++ b/tajo-core/src/test/resources/results/TestWindowQuery/testWindow4.result
@@ -0,0 +1,7 @@
+l_orderkey,l_discount,?windowfunction,?windowfunction_1
+-------------------------------
+1,0.04,0.13,53.0
+1,0.09,0.13,53.0
+2,0.0,0.0,38.0
+3,0.06,0.16,94.0
+3,0.1,0.16,94.0
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/results/TestWindowQuery/testWindow5.result b/tajo-core/src/test/resources/results/TestWindowQuery/testWindow5.result
new file mode 100644
index 0000000..e42b1d3
--- /dev/null
+++ b/tajo-core/src/test/resources/results/TestWindowQuery/testWindow5.result
@@ -0,0 +1,7 @@
+l_orderkey,?windowfunction,l_discount,?windowfunction_1
+-------------------------------
+1,0.13,0.04,53.0
+1,0.13,0.09,53.0
+2,0.0,0.0,38.0
+3,0.16,0.06,94.0
+3,0.16,0.1,94.0
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/results/TestWindowQuery/testWindow6.result b/tajo-core/src/test/resources/results/TestWindowQuery/testWindow6.result
new file mode 100644
index 0000000..7cf7ae1
--- /dev/null
+++ b/tajo-core/src/test/resources/results/TestWindowQuery/testWindow6.result
@@ -0,0 +1,7 @@
+l_orderkey,l_discount,r1,r2
+-------------------------------
+1,0.04,1,0.13
+1,0.09,2,0.13
+2,0.0,1,0.0
+3,0.06,1,0.16
+3,0.1,2,0.16
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/results/TestWindowQuery/testWindow7.result b/tajo-core/src/test/resources/results/TestWindowQuery/testWindow7.result
new file mode 100644
index 0000000..0a7de1a
--- /dev/null
+++ b/tajo-core/src/test/resources/results/TestWindowQuery/testWindow7.result
@@ -0,0 +1,7 @@
+l_orderkey,l_quantity,r
+-------------------------------
+1,17.0,1
+1,36.0,1
+2,38.0,1
+3,45.0,1
+3,49.0,1
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/results/TestWindowQuery/testWindowBeforeLimit.result b/tajo-core/src/test/resources/results/TestWindowQuery/testWindowBeforeLimit.result
new file mode 100644
index 0000000..b01d89a
--- /dev/null
+++ b/tajo-core/src/test/resources/results/TestWindowQuery/testWindowBeforeLimit.result
@@ -0,0 +1,5 @@
+r_name,ran
+-------------------------------
+AFRICA,1
+AMERICA,2
+ASIA,3
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/results/TestWindowQuery/testWindowWithAggregation1.result b/tajo-core/src/test/resources/results/TestWindowQuery/testWindowWithAggregation1.result
new file mode 100644
index 0000000..7c5b94d
--- /dev/null
+++ b/tajo-core/src/test/resources/results/TestWindowQuery/testWindowWithAggregation1.result
@@ -0,0 +1,3 @@
+row_num
+-------------------------------
+1
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/results/TestWindowQuery/testWindowWithAggregation2.result b/tajo-core/src/test/resources/results/TestWindowQuery/testWindowWithAggregation2.result
new file mode 100644
index 0000000..ba11e16
--- /dev/null
+++ b/tajo-core/src/test/resources/results/TestWindowQuery/testWindowWithAggregation2.result
@@ -0,0 +1,5 @@
+l_orderkey,row_num
+-------------------------------
+1,1
+2,1
+3,1
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/results/TestWindowQuery/testWindowWithAggregation3.result b/tajo-core/src/test/resources/results/TestWindowQuery/testWindowWithAggregation3.result
new file mode 100644
index 0000000..a8b99f6
--- /dev/null
+++ b/tajo-core/src/test/resources/results/TestWindowQuery/testWindowWithAggregation3.result
@@ -0,0 +1,3 @@
+cnt,row_num
+-------------------------------
+5,1
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/results/TestWindowQuery/testWindowWithAggregation4.result b/tajo-core/src/test/resources/results/TestWindowQuery/testWindowWithAggregation4.result
new file mode 100644
index 0000000..c34577c
--- /dev/null
+++ b/tajo-core/src/test/resources/results/TestWindowQuery/testWindowWithAggregation4.result
@@ -0,0 +1,5 @@
+l_orderkey,cnt,row_num
+-------------------------------
+3,2,1
+1,2,2
+2,1,3
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/results/TestWindowQuery/testWindowWithAggregation5.result b/tajo-core/src/test/resources/results/TestWindowQuery/testWindowWithAggregation5.result
new file mode 100644
index 0000000..23ff045
--- /dev/null
+++ b/tajo-core/src/test/resources/results/TestWindowQuery/testWindowWithAggregation5.result
@@ -0,0 +1,5 @@
+l_orderkey,cnt,row_num
+-------------------------------
+1,2,1
+2,1,1
+3,2,1
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/results/TestWindowQuery/testWindowWithAggregation6.result b/tajo-core/src/test/resources/results/TestWindowQuery/testWindowWithAggregation6.result
new file mode 100644
index 0000000..7dd39ac
--- /dev/null
+++ b/tajo-core/src/test/resources/results/TestWindowQuery/testWindowWithAggregation6.result
@@ -0,0 +1,5 @@
+l_orderkey,cnt,row_num
+-------------------------------
+1,2,2
+2,1,3
+3,2,1
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/results/TestWindowQuery/testWindowWithOrderBy1.result b/tajo-core/src/test/resources/results/TestWindowQuery/testWindowWithOrderBy1.result
new file mode 100644
index 0000000..a44b4e0
--- /dev/null
+++ b/tajo-core/src/test/resources/results/TestWindowQuery/testWindowWithOrderBy1.result
@@ -0,0 +1,7 @@
+l_orderkey,l_discount,r1
+-------------------------------
+2,0.0,1
+1,0.04,2
+3,0.06,3
+1,0.09,4
+3,0.1,5
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/results/TestWindowQuery/testWindowWithOrderBy2.result b/tajo-core/src/test/resources/results/TestWindowQuery/testWindowWithOrderBy2.result
new file mode 100644
index 0000000..b87197a
--- /dev/null
+++ b/tajo-core/src/test/resources/results/TestWindowQuery/testWindowWithOrderBy2.result
@@ -0,0 +1,7 @@
+l_orderkey,l_partkey,r1
+-------------------------------
+1,1,1
+1,1,1
+2,2,1
+3,2,1
+3,3,2
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/results/TestWindowQuery/testWindowWithOrderBy3.result b/tajo-core/src/test/resources/results/TestWindowQuery/testWindowWithOrderBy3.result
new file mode 100644
index 0000000..68d0f85
--- /dev/null
+++ b/tajo-core/src/test/resources/results/TestWindowQuery/testWindowWithOrderBy3.result
@@ -0,0 +1,7 @@
+l_orderkey,l_partkey,r1
+-------------------------------
+1,1,1
+1,1,1
+2,2,1
+3,3,1
+3,2,2
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/results/TestWindowQuery/testWindowWithOrderBy4.result b/tajo-core/src/test/resources/results/TestWindowQuery/testWindowWithOrderBy4.result
new file mode 100644
index 0000000..7042fb5
--- /dev/null
+++ b/tajo-core/src/test/resources/results/TestWindowQuery/testWindowWithOrderBy4.result
@@ -0,0 +1,7 @@
+l_orderkey,l_partkey,r1,r2
+-------------------------------
+3,3,4,1
+2,2,3,2
+3,2,4,2
+1,1,1,4
+1,1,1,4
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/results/TestWindowQuery/testWindowWithOrderBy5.result b/tajo-core/src/test/resources/results/TestWindowQuery/testWindowWithOrderBy5.result
new file mode 100644
index 0000000..7042fb5
--- /dev/null
+++ b/tajo-core/src/test/resources/results/TestWindowQuery/testWindowWithOrderBy5.result
@@ -0,0 +1,7 @@
+l_orderkey,l_partkey,r1,r2
+-------------------------------
+3,3,4,1
+2,2,3,2
+3,2,4,2
+1,1,1,4
+1,1,1,4
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/results/TestWindowQuery/testWindowWithSubQuery.result b/tajo-core/src/test/resources/results/TestWindowQuery/testWindowWithSubQuery.result
new file mode 100644
index 0000000..bb3d8e6
--- /dev/null
+++ b/tajo-core/src/test/resources/results/TestWindowQuery/testWindowWithSubQuery.result
@@ -0,0 +1,7 @@
+r_name,c,ran
+-------------------------------
+AFRICA,1,1
+AMERICA,1,2
+ASIA,1,3
+EUROPE,1,4
+MIDDLE EAST,1,5
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/results/TestWindowQuery/testWindowWithSubQuery2.result b/tajo-core/src/test/resources/results/TestWindowQuery/testWindowWithSubQuery2.result
new file mode 100644
index 0000000..c3e96b5
--- /dev/null
+++ b/tajo-core/src/test/resources/results/TestWindowQuery/testWindowWithSubQuery2.result
@@ -0,0 +1,5 @@
+r_name,c,ran
+-------------------------------
+AFRICA,1,1
+AMERICA,1,1
+ASIA,1,1
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/results/TestWindowQuery/testWindowWithSubQuery3.result b/tajo-core/src/test/resources/results/TestWindowQuery/testWindowWithSubQuery3.result
new file mode 100644
index 0000000..1a91238
--- /dev/null
+++ b/tajo-core/src/test/resources/results/TestWindowQuery/testWindowWithSubQuery3.result
@@ -0,0 +1,7 @@
+r_name,r_regionkey,mk
+-------------------------------
+AFRICA,0,1
+AMERICA,1,1
+ASIA,2,1
+EUROPE,3,1
+MIDDLE EAST,4,1
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/results/TestWindowQuery/testWindowWithSubQuery4.result b/tajo-core/src/test/resources/results/TestWindowQuery/testWindowWithSubQuery4.result
new file mode 100644
index 0000000..1a91238
--- /dev/null
+++ b/tajo-core/src/test/resources/results/TestWindowQuery/testWindowWithSubQuery4.result
@@ -0,0 +1,7 @@
+r_name,r_regionkey,mk
+-------------------------------
+AFRICA,0,1
+AMERICA,1,1
+ASIA,2,1
+EUROPE,3,1
+MIDDLE EAST,4,1
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/results/TestWindowQuery/testWindowWithSubQuery5.result b/tajo-core/src/test/resources/results/TestWindowQuery/testWindowWithSubQuery5.result
new file mode 100644
index 0000000..7668823
--- /dev/null
+++ b/tajo-core/src/test/resources/results/TestWindowQuery/testWindowWithSubQuery5.result
@@ -0,0 +1,5 @@
+r_name,ran
+-------------------------------
+ASIA,3
+EUROPE,4
+MIDDLE EAST,5
\ No newline at end of file
diff --git a/tajo-core/src/test/resources/results/TestWindowQuery/testWindowWithSubQuery6.result b/tajo-core/src/test/resources/results/TestWindowQuery/testWindowWithSubQuery6.result
new file mode 100644
index 0000000..2a20b4e
--- /dev/null
+++ b/tajo-core/src/test/resources/results/TestWindowQuery/testWindowWithSubQuery6.result
@@ -0,0 +1,3 @@
+r_name,ran
+-------------------------------
+ASIA,1
\ No newline at end of file
diff --git a/tajo-core/tajo-core-backend/src/test/resources/results/TestSQLAnalyzer/window1.result b/tajo-core/tajo-core-backend/src/test/resources/results/TestSQLAnalyzer/window1.result
deleted file mode 100644
index 31ed149..0000000
--- a/tajo-core/tajo-core-backend/src/test/resources/results/TestSQLAnalyzer/window1.result
+++ /dev/null
@@ -1,42 +0,0 @@
-{
-  "distinct": false,
-  "targets": [
-    {
-      "child": {
-        "name": "dt",
-        "type": "Column"
-      },
-      "type": "Target"
-    },
-    {
-      "child": {
-        "function": {
-          "distinct": false,
-          "signature": "sum",
-          "params": [
-            {
-              "name": "xy",
-              "type": "Column"
-            }
-          ],
-          "type": "GeneralSetFunction"
-        },
-        "windowSpec": {
-          "type": "WindowSpec"
-        },
-        "type": "WindowFunction"
-      },
-      "type": "Target"
-    }
-  ],
-  "child": {
-    "relations": [
-      {
-        "tableName": "sum_example",
-        "type": "Relation"
-      }
-    ],
-    "type": "RelationList"
-  },
-  "type": "Projection"
-}
\ No newline at end of file
diff --git a/tajo-core/tajo-core-backend/src/test/resources/results/TestSQLAnalyzer/window2.result b/tajo-core/tajo-core-backend/src/test/resources/results/TestSQLAnalyzer/window2.result
deleted file mode 100644
index 378232c..0000000
--- a/tajo-core/tajo-core-backend/src/test/resources/results/TestSQLAnalyzer/window2.result
+++ /dev/null
@@ -1,48 +0,0 @@
-{
-  "distinct": false,
-  "targets": [
-    {
-      "child": {
-        "name": "dt",
-        "type": "Column"
-      },
-      "type": "Target"
-    },
-    {
-      "child": {
-        "function": {
-          "distinct": false,
-          "signature": "sum",
-          "params": [
-            {
-              "name": "xy",
-              "type": "Column"
-            }
-          ],
-          "type": "GeneralSetFunction"
-        },
-        "windowSpec": {
-          "partitionKeys": [
-            {
-              "name": "dt",
-              "type": "Column"
-            }
-          ],
-          "type": "WindowSpec"
-        },
-        "type": "WindowFunction"
-      },
-      "type": "Target"
-    }
-  ],
-  "child": {
-    "relations": [
-      {
-        "tableName": "sum_example",
-        "type": "Relation"
-      }
-    ],
-    "type": "RelationList"
-  },
-  "type": "Projection"
-}
\ No newline at end of file
diff --git a/tajo-core/tajo-core-backend/src/test/resources/results/TestSQLAnalyzer/window3.result b/tajo-core/tajo-core-backend/src/test/resources/results/TestSQLAnalyzer/window3.result
deleted file mode 100644
index c5bc47d..0000000
--- a/tajo-core/tajo-core-backend/src/test/resources/results/TestSQLAnalyzer/window3.result
+++ /dev/null
@@ -1,54 +0,0 @@
-{
-  "distinct": false,
-  "targets": [
-    {
-      "child": {
-        "name": "dt",
-        "type": "Column"
-      },
-      "type": "Target"
-    },
-    {
-      "child": {
-        "function": {
-          "distinct": false,
-          "signature": "sum",
-          "params": [
-            {
-              "name": "xy",
-              "type": "Column"
-            }
-          ],
-          "type": "GeneralSetFunction"
-        },
-        "windowSpec": {
-          "partitionKeys": [
-            {
-              "signature": "round",
-              "params": [
-                {
-                  "name": "dt",
-                  "type": "Column"
-                }
-              ],
-              "type": "Function"
-            }
-          ],
-          "type": "WindowSpec"
-        },
-        "type": "WindowFunction"
-      },
-      "type": "Target"
-    }
-  ],
-  "child": {
-    "relations": [
-      {
-        "tableName": "sum_example",
-        "type": "Relation"
-      }
-    ],
-    "type": "RelationList"
-  },
-  "type": "Projection"
-}
\ No newline at end of file
diff --git a/tajo-core/tajo-core-backend/src/test/resources/results/TestSQLAnalyzer/window4.result b/tajo-core/tajo-core-backend/src/test/resources/results/TestSQLAnalyzer/window4.result
deleted file mode 100644
index 0edd7d4..0000000
--- a/tajo-core/tajo-core-backend/src/test/resources/results/TestSQLAnalyzer/window4.result
+++ /dev/null
@@ -1,52 +0,0 @@
-{
-  "distinct": false,
-  "targets": [
-    {
-      "child": {
-        "name": "dt",
-        "type": "Column"
-      },
-      "type": "Target"
-    },
-    {
-      "child": {
-        "function": {
-          "distinct": false,
-          "signature": "sum",
-          "params": [
-            {
-              "name": "xy",
-              "type": "Column"
-            }
-          ],
-          "type": "GeneralSetFunction"
-        },
-        "windowSpec": {
-          "sortSpecs": [
-            {
-              "key": {
-                "name": "dt",
-                "type": "Column"
-              },
-              "asc": true,
-              "nullFirst": false
-            }
-          ],
-          "type": "WindowSpec"
-        },
-        "type": "WindowFunction"
-      },
-      "type": "Target"
-    }
-  ],
-  "child": {
-    "relations": [
-      {
-        "tableName": "sum_example",
-        "type": "Relation"
-      }
-    ],
-    "type": "RelationList"
-  },
-  "type": "Projection"
-}
\ No newline at end of file
diff --git a/tajo-core/tajo-core-backend/src/test/resources/results/TestSQLAnalyzer/window5.result b/tajo-core/tajo-core-backend/src/test/resources/results/TestSQLAnalyzer/window5.result
deleted file mode 100644
index 06192ff..0000000
--- a/tajo-core/tajo-core-backend/src/test/resources/results/TestSQLAnalyzer/window5.result
+++ /dev/null
@@ -1,81 +0,0 @@
-{
-  "distinct": false,
-  "targets": [
-    {
-      "child": {
-        "name": "dt",
-        "type": "Column"
-      },
-      "type": "Target"
-    },
-    {
-      "child": {
-        "name": "dt2",
-        "type": "Column"
-      },
-      "type": "Target"
-    },
-    {
-      "child": {
-        "function": {
-          "distinct": false,
-          "signature": "sum",
-          "params": [
-            {
-              "name": "xy",
-              "type": "Column"
-            }
-          ],
-          "type": "GeneralSetFunction"
-        },
-        "windowSpec": {
-          "partitionKeys": [
-            {
-              "signature": "round",
-              "params": [
-                {
-                  "name": "dt",
-                  "type": "Column"
-                }
-              ],
-              "type": "Function"
-            },
-            {
-              "name": "dt2",
-              "type": "Column"
-            }
-          ],
-          "sortSpecs": [
-            {
-              "key": {
-                "signature": "ceil",
-                "params": [
-                  {
-                    "name": "dt",
-                    "type": "Column"
-                  }
-                ],
-                "type": "Function"
-              },
-              "asc": true,
-              "nullFirst": false
-            }
-          ],
-          "type": "WindowSpec"
-        },
-        "type": "WindowFunction"
-      },
-      "type": "Target"
-    }
-  ],
-  "child": {
-    "relations": [
-      {
-        "tableName": "sum_example",
-        "type": "Relation"
-      }
-    ],
-    "type": "RelationList"
-  },
-  "type": "Projection"
-}
\ No newline at end of file
diff --git a/tajo-core/tajo-core-backend/src/test/resources/results/TestSQLAnalyzer/window6.result b/tajo-core/tajo-core-backend/src/test/resources/results/TestSQLAnalyzer/window6.result
deleted file mode 100644
index 97bd236..0000000
--- a/tajo-core/tajo-core-backend/src/test/resources/results/TestSQLAnalyzer/window6.result
+++ /dev/null
@@ -1,87 +0,0 @@
-{
-  "distinct": false,
-  "targets": [
-    {
-      "child": {
-        "name": "dt",
-        "type": "Column"
-      },
-      "type": "Target"
-    },
-    {
-      "child": {
-        "name": "dt2",
-        "type": "Column"
-      },
-      "type": "Target"
-    },
-    {
-      "child": {
-        "function": {
-          "distinct": false,
-          "signature": "sum",
-          "params": [
-            {
-              "name": "xy",
-              "type": "Column"
-            }
-          ],
-          "type": "GeneralSetFunction"
-        },
-        "windowSpec": {
-          "partitionKeys": [
-            {
-              "signature": "round",
-              "params": [
-                {
-                  "name": "dt",
-                  "type": "Column"
-                }
-              ],
-              "type": "Function"
-            },
-            {
-              "name": "dt2",
-              "type": "Column"
-            }
-          ],
-          "sortSpecs": [
-            {
-              "key": {
-                "signature": "ceil",
-                "params": [
-                  {
-                    "name": "dt",
-                    "type": "Column"
-                  }
-                ],
-                "type": "Function"
-              },
-              "asc": true,
-              "nullFirst": false
-            }
-          ],
-          "windowFrame": {
-            "unit": "ROW",
-            "startBound": {
-              "boundType": "UNBOUNDED_PRECEDING"
-            }
-          },
-          "type": "WindowSpec"
-        },
-        "type": "WindowFunction"
-      },
-      "type": "Target"
-    }
-  ],
-  "child": {
-    "relations": [
-      {
-        "tableName": "sum_example",
-        "type": "Relation"
-      }
-    ],
-    "type": "RelationList"
-  },
-  "type": "Projection"
-}
\ No newline at end of file
diff --git a/tajo-core/tajo-core-backend/src/test/resources/results/TestSQLAnalyzer/window7.result b/tajo-core/tajo-core-backend/src/test/resources/results/TestSQLAnalyzer/window7.result
deleted file mode 100644
index 9ae01bd..0000000
--- a/tajo-core/tajo-core-backend/src/test/resources/results/TestSQLAnalyzer/window7.result
+++ /dev/null
@@ -1,85 +0,0 @@
-{
-  "distinct": false,
-  "targets": [
-    {
-      "child": {
-        "name": "dt",
-        "type": "Column"
-      },
-      "type": "Target"
-    },
-    {
-      "child": {
-        "name": "dt2",
-        "type": "Column"
-      },
-      "type": "Target"
-    },
-    {
-      "child": {
-        "function": {
-          "distinct": false,
-          "signature": "row_number",
-          "params": [],
-          "type": "GeneralSetFunction"
-        },
-        "windowSpec": {
-          "partitionKeys": [
-            {
-              "signature": "round",
-              "params": [
-                {
-                  "name": "dt",
-                  "type": "Column"
-                }
-              ],
-              "type": "Function"
-            },
-            {
-              "name": "dt2",
-              "type": "Column"
-            }
-          ],
-          "sortSpecs": [
-            {
-              "key": {
-                "signature": "ceil",
-                "params": [
-                  {
-                    "name": "dt",
-                    "type": "Column"
-                  }
-                ],
-                "type": "Function"
-              },
-              "asc": true,
-              "nullFirst": false
-            }
-          ],
-          "windowFrame": {
-            "unit": "ROW",
-            "startBound": {
-              "boundType": "UNBOUNDED_PRECEDING"
-            },
-            "endBound": {
-              "boundType": "UNBOUNDED_FOLLOWING"
-            }
-          },
-          "type": "WindowSpec"
-        },
-        "type": "WindowFunction"
-      },
-      "type": "Target"
-    }
-  ],
-  "child": {
-    "relations": [
-      {
-        "tableName": "sum_example",
-        "type": "Relation"
-      }
-    ],
-    "type": "RelationList"
-  },
-  "type": "Projection"
-}
\ No newline at end of file
diff --git a/tajo-core/tajo-core-backend/src/test/resources/results/TestSQLAnalyzer/window8.result b/tajo-core/tajo-core-backend/src/test/resources/results/TestSQLAnalyzer/window8.result
deleted file mode 100644
index b9abace..0000000
--- a/tajo-core/tajo-core-backend/src/test/resources/results/TestSQLAnalyzer/window8.result
+++ /dev/null
@@ -1,90 +0,0 @@
-{
-  "distinct": false,
-  "targets": [
-    {
-      "child": {
-        "name": "dt",
-        "type": "Column"
-      },
-      "type": "Target"
-    },
-    {
-      "child": {
-        "name": "dt2",
-        "type": "Column"
-      },
-      "type": "Target"
-    },
-    {
-      "child": {
-        "function": {
-          "distinct": false,
-          "signature": "row_number",
-          "params": [],
-          "type": "GeneralSetFunction"
-        },
-        "windowSpec": {
-          "partitionKeys": [
-            {
-              "signature": "round",
-              "params": [
-                {
-                  "name": "dt",
-                  "type": "Column"
-                }
-              ],
-              "type": "Function"
-            },
-            {
-              "name": "dt2",
-              "type": "Column"
-            }
-          ],
-          "sortSpecs": [
-            {
-              "key": {
-                "signature": "ceil",
-                "params": [
-                  {
-                    "name": "dt",
-                    "type": "Column"
-                  }
-                ],
-                "type": "Function"
-              },
-              "asc": true,
-              "nullFirst": false
-            }
-          ],
-          "windowFrame": {
-            "unit": "ROW",
-            "startBound": {
-              "boundType": "PRECEDING",
-              "number": {
-                "valueType": "Unsigned_Integer",
-                "value": "1",
-                "type": "Literal"
-              }
-            },
-            "endBound": {
-              "boundType": "CURRENT_ROW"
-            }
-          },
-          "type": "WindowSpec"
-        },
-        "type": "WindowFunction"
-      },
-      "type": "Target"
-    }
-  ],
-  "child": {
-    "relations": [
-      {
-        "tableName": "sum_example",
-        "type": "Relation"
-      }
-    ],
-    "type": "RelationList"
-  },
-  "type": "Projection"
-}
\ No newline at end of file
diff --git a/tajo-core/tajo-core-backend/src/test/resources/results/TestSQLAnalyzer/window9.result b/tajo-core/tajo-core-backend/src/test/resources/results/TestSQLAnalyzer/window9.result
deleted file mode 100644
index 613fb9b..0000000
--- a/tajo-core/tajo-core-backend/src/test/resources/results/TestSQLAnalyzer/window9.result
+++ /dev/null
@@ -1,118 +0,0 @@
-{
-  "distinct": false,
-  "targets": [
-    {
-      "child": {
-        "name": "dt",
-        "type": "Column"
-      },
-      "type": "Target"
-    },
-    {
-      "child": {
-        "name": "dt2",
-        "type": "Column"
-      },
-      "type": "Target"
-    },
-    {
-      "child": {
-        "function": {
-          "distinct": false,
-          "signature": "row_number",
-          "params": [],
-          "type": "GeneralSetFunction"
-        },
-        "windowSpec": {
-          "windowName": "window1",
-          "type": "WindowSpec"
-        },
-        "type": "WindowFunction"
-      },
-      "type": "Target"
-    },
-    {
-      "child": {
-        "function": {
-          "distinct": false,
-          "signature": "rank",
-          "params": [],
-          "type": "GeneralSetFunction"
-        },
-        "windowSpec": {
-          "windowName": "window1",
-          "type": "WindowSpec"
-        },
-        "type": "WindowFunction"
-      },
-      "type": "Target"
-    }
-  ],
-  "child": {
-    "definitions": [
-      {
-        "windowName": "window1",
-        "windowSpec": {
-          "partitionKeys": [
-            {
-              "signature": "round",
-              "params": [
-                {
-                  "name": "dt",
-                  "type": "Column"
-                }
-              ],
-              "type": "Function"
-            },
-            {
-              "name": "dt2",
-              "type": "Column"
-            }
-          ],
-          "sortSpecs": [
-            {
-              "key": {
-                "signature": "ceil",
-                "params": [
-                  {
-                    "name": "dt",
-                    "type": "Column"
-                  }
-                ],
-                "type": "Function"
-              },
-              "asc": true,
-              "nullFirst": false
-            }
-          ],
-          "windowFrame": {
-            "unit": "ROW",
-            "startBound": {
-              "boundType": "PRECEDING",
-              "number": {
-                "valueType": "Unsigned_Integer",
-                "value": "1",
-                "type": "Literal"
-              }
-            },
-            "endBound": {
-              "boundType": "CURRENT_ROW"
-            }
-          },
-          "type": "WindowSpec"
-        }
-      }
-    ],
-    "child": {
-      "relations": [
-        {
-          "tableName": "sum_example",
-          "type": "Relation"
-        }
-      ],
-      "type": "RelationList"
-    },
-    "type": "Window"
-  },
-  "type": "Projection"
-}
\ No newline at end of file
diff --git a/tajo-storage/src/main/java/org/apache/tajo/storage/BinarySerializerDeserializer.java b/tajo-storage/src/main/java/org/apache/tajo/storage/BinarySerializerDeserializer.java
index ed034be..42b49a8 100644
--- a/tajo-storage/src/main/java/org/apache/tajo/storage/BinarySerializerDeserializer.java
+++ b/tajo-storage/src/main/java/org/apache/tajo/storage/BinarySerializerDeserializer.java
@@ -221,7 +221,7 @@
 
   private byte[] vLongBytes = new byte[9];
 
-  public int writeVLongToByteArray(byte[] bytes, int offset, long l) {
+  public static int writeVLongToByteArray(byte[] bytes, int offset, long l) {
     if (l >= -112 && l <= 127) {
       bytes[offset] = (byte) l;
       return 1;