You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tajo.apache.org by hy...@apache.org on 2014/07/09 08:21:43 UTC
[11/13] TAJO-774: Implement logical plan part and physical executor
for window function.
http://git-wip-us.apache.org/repos/asf/tajo/blob/d3ee50a9/tajo-core/src/main/java/org/apache/tajo/engine/planner/PhysicalPlannerImpl.java
----------------------------------------------------------------------
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 @@ public class PhysicalPlannerImpl implements PhysicalPlanner {
leftExec = createPlanRecursive(ctx, subQueryNode.getSubQuery(), stack);
stack.pop();
return new ProjectionExec(ctx, subQueryNode, leftExec);
-
}
case PARTITIONS_SCAN:
@@ -178,6 +177,13 @@ public class PhysicalPlannerImpl implements PhysicalPlanner {
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 class PhysicalPlannerImpl implements PhysicalPlanner {
}
}
+ 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 {
http://git-wip-us.apache.org/repos/asf/tajo/blob/d3ee50a9/tajo-core/src/main/java/org/apache/tajo/engine/planner/PlannerUtil.java
----------------------------------------------------------------------
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 class PlannerUtil {
}
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 @@ public class PlannerUtil {
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) {
http://git-wip-us.apache.org/repos/asf/tajo/blob/d3ee50a9/tajo-core/src/main/java/org/apache/tajo/engine/planner/SimpleAlgebraVisitor.java
----------------------------------------------------------------------
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 @@ public abstract class SimpleAlgebraVisitor<CONTEXT, RESULT> extends BaseAlgebraV
}
///////////////////////////////////////////////////////////////////////////////////////////////////////////
- // 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 @@ public abstract class SimpleAlgebraVisitor<CONTEXT, RESULT> extends BaseAlgebraV
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 @@ public abstract class SimpleAlgebraVisitor<CONTEXT, RESULT> extends BaseAlgebraV
}
@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);
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/d3ee50a9/tajo-core/src/main/java/org/apache/tajo/engine/planner/TypeDeterminant.java
----------------------------------------------------------------------
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);
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/d3ee50a9/tajo-core/src/main/java/org/apache/tajo/engine/planner/global/GlobalPlanner.java
----------------------------------------------------------------------
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 @@ public class GlobalPlanner {
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 @@ public class GlobalPlanner {
}
@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);
http://git-wip-us.apache.org/repos/asf/tajo/blob/d3ee50a9/tajo-core/src/main/java/org/apache/tajo/engine/planner/logical/NodeType.java
----------------------------------------------------------------------
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 @@ public enum NodeType {
SORT(SortNode.class),
HAVING(HavingNode.class),
GROUP_BY(GroupbyNode.class),
+ WINDOW_AGG(WindowAggNode.class),
SELECTION(SelectionNode.class),
JOIN(JoinNode.class),
UNION(UnionNode.class),
http://git-wip-us.apache.org/repos/asf/tajo/blob/d3ee50a9/tajo-core/src/main/java/org/apache/tajo/engine/planner/logical/WindowAggNode.java
----------------------------------------------------------------------
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;
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/d3ee50a9/tajo-core/src/main/java/org/apache/tajo/engine/planner/logical/WindowSpec.java
----------------------------------------------------------------------
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);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/d3ee50a9/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/WindowAggExec.java
----------------------------------------------------------------------
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;
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/d3ee50a9/tajo-core/src/main/java/org/apache/tajo/engine/planner/rewrite/FilterPushDownRule.java
----------------------------------------------------------------------
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 @@ public class FilterPushDownRule extends BasicLogicalPlanVisitor<FilterPushDownCo
context.setFiltersTobePushed(new HashSet<EvalNode>(transformedMap.keySet()));
visit(context, plan, plan.getBlock(node.getSubQuery()));
-
context.setToOrigin(transformedMap);
return node;
@@ -512,18 +510,18 @@ public class FilterPushDownRule extends BasicLogicalPlanVisitor<FilterPushDownCo
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 @@ public class FilterPushDownRule extends BasicLogicalPlanVisitor<FilterPushDownCo
}
// 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 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 Map<EvalNode, EvalNode> findCanPushdownAndTransform(
+ private BiMap<EvalNode, EvalNode> findCanPushdownAndTransform(
FilterPushDownContext context, Projectable node,
LogicalNode childNode, List<EvalNode> notMatched,
Set<String> partitionColumns,
@@ -565,7 +578,7 @@ public class FilterPushDownRule extends BasicLogicalPlanVisitor<FilterPushDownCo
}
// 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 @@ public class FilterPushDownRule extends BasicLogicalPlanVisitor<FilterPushDownCo
}
@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 {
http://git-wip-us.apache.org/repos/asf/tajo/blob/d3ee50a9/tajo-core/src/main/java/org/apache/tajo/engine/planner/rewrite/ProjectionPushDownRule.java
----------------------------------------------------------------------
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 @@ public class ProjectionPushDownRule extends
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);
http://git-wip-us.apache.org/repos/asf/tajo/blob/d3ee50a9/tajo-core/src/main/java/org/apache/tajo/engine/utils/DataTypeUtil.java
----------------------------------------------------------------------
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;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/d3ee50a9/tajo-core/src/main/java/org/apache/tajo/master/GlobalEngine.java
----------------------------------------------------------------------
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 @@ public class GlobalEngine extends AbstractService {
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());
http://git-wip-us.apache.org/repos/asf/tajo/blob/d3ee50a9/tajo-core/src/test/java/org/apache/tajo/engine/parser/TestSQLAnalyzer.java
----------------------------------------------------------------------
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 @@ public class TestSQLAnalyzer {
String result = FileUtil.readTextFileFromResource("results/TestSQLAnalyzer/" + resultFileName);
Expr expr = parseQuery(sql);
- assertEquals(expr.toJson().trim(), result.trim());
+ assertEquals(result.trim(), expr.toJson().trim());
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/d3ee50a9/tajo-core/src/test/java/org/apache/tajo/engine/query/TestGroupByQuery.java
----------------------------------------------------------------------
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 @@ public class TestGroupByQuery extends QueryTestCaseBase {
"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";
http://git-wip-us.apache.org/repos/asf/tajo/blob/d3ee50a9/tajo-core/src/test/java/org/apache/tajo/engine/query/TestWindowQuery.java
----------------------------------------------------------------------
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
http://git-wip-us.apache.org/repos/asf/tajo/blob/d3ee50a9/tajo-core/src/test/resources/queries/TestLogicalPlanner/window1.sql
----------------------------------------------------------------------
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
http://git-wip-us.apache.org/repos/asf/tajo/blob/d3ee50a9/tajo-core/src/test/resources/queries/TestSQLAnalyzer/window1.sql
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/resources/queries/TestSQLAnalyzer/window1.sql b/tajo-core/src/test/resources/queries/TestSQLAnalyzer/window1.sql
new file mode 100644
index 0000000..52156b8
--- /dev/null
+++ b/tajo-core/src/test/resources/queries/TestSQLAnalyzer/window1.sql
@@ -0,0 +1,5 @@
+ SELECT
+ dt,
+ sum(xy) over()
+FROM
+ sum_example;
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/d3ee50a9/tajo-core/src/test/resources/queries/TestSQLAnalyzer/window2.sql
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/resources/queries/TestSQLAnalyzer/window2.sql b/tajo-core/src/test/resources/queries/TestSQLAnalyzer/window2.sql
new file mode 100644
index 0000000..60f1fbf
--- /dev/null
+++ b/tajo-core/src/test/resources/queries/TestSQLAnalyzer/window2.sql
@@ -0,0 +1,5 @@
+ SELECT
+ dt,
+ sum(xy) over(partition by dt)
+FROM
+ sum_example;
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/d3ee50a9/tajo-core/src/test/resources/queries/TestSQLAnalyzer/window3.sql
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/resources/queries/TestSQLAnalyzer/window3.sql b/tajo-core/src/test/resources/queries/TestSQLAnalyzer/window3.sql
new file mode 100644
index 0000000..01dd542
--- /dev/null
+++ b/tajo-core/src/test/resources/queries/TestSQLAnalyzer/window3.sql
@@ -0,0 +1,5 @@
+ SELECT
+ dt,
+ sum(xy) over(partition by round(dt))
+FROM
+ sum_example;
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/d3ee50a9/tajo-core/src/test/resources/queries/TestSQLAnalyzer/window4.sql
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/resources/queries/TestSQLAnalyzer/window4.sql b/tajo-core/src/test/resources/queries/TestSQLAnalyzer/window4.sql
new file mode 100644
index 0000000..629d5e4
--- /dev/null
+++ b/tajo-core/src/test/resources/queries/TestSQLAnalyzer/window4.sql
@@ -0,0 +1,5 @@
+ SELECT
+ dt,
+ sum(xy) over(order by dt)
+FROM
+ sum_example;
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/d3ee50a9/tajo-core/src/test/resources/queries/TestSQLAnalyzer/window5.sql
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/resources/queries/TestSQLAnalyzer/window5.sql b/tajo-core/src/test/resources/queries/TestSQLAnalyzer/window5.sql
new file mode 100644
index 0000000..6338817
--- /dev/null
+++ b/tajo-core/src/test/resources/queries/TestSQLAnalyzer/window5.sql
@@ -0,0 +1,6 @@
+ SELECT
+ dt,
+ dt2,
+ sum(xy) over(partition by round(dt),dt2 order by ceil(dt) asc null last)
+FROM
+ sum_example;
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/d3ee50a9/tajo-core/src/test/resources/queries/TestSQLAnalyzer/window6.sql
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/resources/queries/TestSQLAnalyzer/window6.sql b/tajo-core/src/test/resources/queries/TestSQLAnalyzer/window6.sql
new file mode 100644
index 0000000..1f1870e
--- /dev/null
+++ b/tajo-core/src/test/resources/queries/TestSQLAnalyzer/window6.sql
@@ -0,0 +1,6 @@
+ SELECT
+ dt,
+ dt2,
+ sum(xy) over(partition by round(dt),dt2 order by ceil(dt) asc null last ROWS UNBOUNDED PRECEDING)
+FROM
+ sum_example;
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/d3ee50a9/tajo-core/src/test/resources/queries/TestSQLAnalyzer/window7.sql
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/resources/queries/TestSQLAnalyzer/window7.sql b/tajo-core/src/test/resources/queries/TestSQLAnalyzer/window7.sql
new file mode 100644
index 0000000..6daa806
--- /dev/null
+++ b/tajo-core/src/test/resources/queries/TestSQLAnalyzer/window7.sql
@@ -0,0 +1,6 @@
+ SELECT
+ dt,
+ dt2,
+ row_number() over(partition by round(dt),dt2 order by ceil(dt) asc null last ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING)
+FROM
+ sum_example;
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/d3ee50a9/tajo-core/src/test/resources/queries/TestSQLAnalyzer/window8.sql
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/resources/queries/TestSQLAnalyzer/window8.sql b/tajo-core/src/test/resources/queries/TestSQLAnalyzer/window8.sql
new file mode 100644
index 0000000..6f70690
--- /dev/null
+++ b/tajo-core/src/test/resources/queries/TestSQLAnalyzer/window8.sql
@@ -0,0 +1,6 @@
+ SELECT
+ dt,
+ dt2,
+ row_number() over(partition by round(dt),dt2 order by ceil(dt) asc null last ROWS BETWEEN 1 PRECEDING AND CURRENT ROW)
+FROM
+ sum_example;
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/d3ee50a9/tajo-core/src/test/resources/queries/TestSQLAnalyzer/window9.sql
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/resources/queries/TestSQLAnalyzer/window9.sql b/tajo-core/src/test/resources/queries/TestSQLAnalyzer/window9.sql
new file mode 100644
index 0000000..b289ebf
--- /dev/null
+++ b/tajo-core/src/test/resources/queries/TestSQLAnalyzer/window9.sql
@@ -0,0 +1,9 @@
+ SELECT
+ dt,
+ dt2,
+ row_number() over(window1),
+ rank() over(window1)
+FROM
+ sum_example
+WINDOW
+ window1 AS (partition by round(dt),dt2 order by ceil(dt) asc null last ROWS BETWEEN 1 PRECEDING AND CURRENT ROW);
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/d3ee50a9/tajo-core/src/test/resources/queries/TestWindowQuery/rowNumber1.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
http://git-wip-us.apache.org/repos/asf/tajo/blob/d3ee50a9/tajo-core/src/test/resources/queries/TestWindowQuery/rowNumber2.sql
----------------------------------------------------------------------
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
http://git-wip-us.apache.org/repos/asf/tajo/blob/d3ee50a9/tajo-core/src/test/resources/queries/TestWindowQuery/rowNumber3.sql
----------------------------------------------------------------------
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
http://git-wip-us.apache.org/repos/asf/tajo/blob/d3ee50a9/tajo-core/src/test/resources/queries/TestWindowQuery/testComplexOrderBy1.sql
----------------------------------------------------------------------
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
http://git-wip-us.apache.org/repos/asf/tajo/blob/d3ee50a9/tajo-core/src/test/resources/queries/TestWindowQuery/testWindow1.sql
----------------------------------------------------------------------
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
http://git-wip-us.apache.org/repos/asf/tajo/blob/d3ee50a9/tajo-core/src/test/resources/queries/TestWindowQuery/testWindow2.sql
----------------------------------------------------------------------
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
http://git-wip-us.apache.org/repos/asf/tajo/blob/d3ee50a9/tajo-core/src/test/resources/queries/TestWindowQuery/testWindow3.sql
----------------------------------------------------------------------
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
http://git-wip-us.apache.org/repos/asf/tajo/blob/d3ee50a9/tajo-core/src/test/resources/queries/TestWindowQuery/testWindow4.sql
----------------------------------------------------------------------
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
http://git-wip-us.apache.org/repos/asf/tajo/blob/d3ee50a9/tajo-core/src/test/resources/queries/TestWindowQuery/testWindow5.sql
----------------------------------------------------------------------
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
http://git-wip-us.apache.org/repos/asf/tajo/blob/d3ee50a9/tajo-core/src/test/resources/queries/TestWindowQuery/testWindow6.sql
----------------------------------------------------------------------
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
http://git-wip-us.apache.org/repos/asf/tajo/blob/d3ee50a9/tajo-core/src/test/resources/queries/TestWindowQuery/testWindow7.sql
----------------------------------------------------------------------
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
http://git-wip-us.apache.org/repos/asf/tajo/blob/d3ee50a9/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowBeforeLimit.sql
----------------------------------------------------------------------
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
http://git-wip-us.apache.org/repos/asf/tajo/blob/d3ee50a9/tajo-core/src/test/resources/queries/TestWindowQuery/testWindowWithAggregation1.sql
----------------------------------------------------------------------
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