Initial commit to add test.

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/WindowExpression.scala

@@ -0,0 +1,49 @@
+/*
+ * 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.spark.sql.catalyst.expressions
+
+import org.apache.spark.sql.types.DataType
+
+
+/**
+ * @param child the computation being performed
+ * @param windowSpec the window spec definition
+ */
+case class WindowExpression(child: Expression, windowSpec: WindowSpec) extends UnaryExpression {
+
+  override type EvaluatedType = Any
+
+  override def eval(input: Row): Any = child.eval(input)
+
+  override def dataType: DataType = child.dataType
+  override def foldable: Boolean = child.foldable
+  override def nullable: Boolean = child.nullable
+
+  override def toString: String = s"$child $windowSpec"
+}
+
+case class WindowSpec(windowPartition: WindowPartition, windowFrame: Option[WindowFrame])
+
+case class WindowPartition(partitionBy: Seq[Expression], sortBy: Seq[SortOrder])
+
+sealed trait FrameType
+
+case object RowFrame extends FrameType
+case object RangeFrame extends FrameType
+
+case class WindowFrame(frameType: FrameType, preceding: Int, following: Int)

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicOperators.scala

@@ -189,6 +189,16 @@ case class Aggregate(
   override def output: Seq[Attribute] = aggregateExpressions.map(_.toAttribute)
 }
 
+case class WindowAggregate(
+    partitionExpressions: Seq[Expression],
+    windowExpressions: Seq[Alias],
+    otherExpressions: Seq[NamedExpression],
+    child: LogicalPlan)
+  extends UnaryNode {
+
+  override def output: Seq[Attribute] = (windowExpressions ++ otherExpressions).map(_.toAttribute)
+}
+
 /**
  * Apply the all of the GroupExpressions to every input row, hence we will get
  * multiple output rows for a input row.

sql/hive/compatibility/src/test/scala/org/apache/spark/sql/hive/execution/HiveCompatibilitySuite.scala

@@ -185,7 +185,7 @@ class HiveCompatibilitySuite extends HiveQueryFileTest with BeforeAndAfter {
     // Hive does not support buckets.
     ".*bucket.*",
 
-    // No window support yet
+    // We have our own tests based on these query files.
     ".*window.*",
 
     // Fails in hive with authorization errors.

sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveContext.scala

@@ -250,6 +250,7 @@ class HiveContext(sc: SparkContext) extends SQLContext(sc) {
         catalog.PreInsertionCasts ::
         ExtractPythonUdfs ::
         ResolveUdtfsAlias ::
+        ResolveWindowUdaf ::
         sources.PreInsertCastAndRename ::
         Nil
     }
@@ -371,6 +372,7 @@ class HiveContext(sc: SparkContext) extends SQLContext(sc) {
       InMemoryScans,
       ParquetConversion, // Must be before HiveTableScans
       HiveTableScans,
+      WindowFunction,
       DataSinks,
       Scripts,
       HashAggregation,