quickstep-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From rathi...@apache.org
Subject [04/30] incubator-quickstep git commit: Added ExecutionGenerator support for Window Aggregation.
Date Fri, 05 Aug 2016 22:52:28 GMT
Added ExecutionGenerator support for Window Aggregation.


Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/7671a589
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/7671a589
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/7671a589

Branch: refs/heads/quickstep-28-29
Commit: 7671a5893349f060d95e20cd8b98ad076c614cec
Parents: 303e51e
Author: shixuan-fan <shixuan@apache.org>
Authored: Wed Jun 29 20:25:18 2016 +0000
Committer: Zuyu Zhang <zuyuz@apache.org>
Committed: Fri Jul 8 11:15:20 2016 -0500

----------------------------------------------------------------------
 query_execution/CMakeLists.txt                  |   2 +
 query_execution/QueryContext.cpp                |  14 ++
 query_execution/QueryContext.hpp                |  48 +++++
 query_execution/QueryContext.proto              |   5 +-
 query_optimizer/CMakeLists.txt                  |   2 +
 query_optimizer/ExecutionGenerator.cpp          | 101 ++++++++++-
 query_optimizer/ExecutionGenerator.hpp          |   8 +
 query_optimizer/cost_model/CMakeLists.txt       |   2 +
 query_optimizer/cost_model/SimpleCostModel.cpp  |   9 +
 query_optimizer/cost_model/SimpleCostModel.hpp  |   5 +
 .../cost_model/StarSchemaSimpleCostModel.cpp    |   8 +
 .../cost_model/StarSchemaSimpleCostModel.hpp    |   4 +
 query_optimizer/resolver/Resolver.cpp           |   7 +-
 .../tests/execution_generator/Select.test       |  39 ++++
 query_optimizer/tests/resolver/Select.test      |  33 ++++
 relational_operators/CMakeLists.txt             |  14 ++
 .../WindowAggregationOperator.cpp               |  82 +++++++++
 .../WindowAggregationOperator.hpp               | 166 +++++++++++++++++
 relational_operators/WorkOrder.proto            |   9 +
 storage/CMakeLists.txt                          |  45 ++++-
 storage/WindowAggregationOperationState.cpp     | 179 +++++++++++++++++++
 storage/WindowAggregationOperationState.hpp     | 177 ++++++++++++++++++
 storage/WindowAggregationOperationState.proto   |  33 ++++
 ...WindowAggregationOperationState_unittest.cpp |  92 ++++++++++
 24 files changed, 1079 insertions(+), 5 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/7671a589/query_execution/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/query_execution/CMakeLists.txt b/query_execution/CMakeLists.txt
index 2be451c..e1b1183 100644
--- a/query_execution/CMakeLists.txt
+++ b/query_execution/CMakeLists.txt
@@ -118,6 +118,7 @@ target_link_libraries(quickstep_queryexecution_QueryContext
                       quickstep_storage_HashTableFactory
                       quickstep_storage_InsertDestination
                       quickstep_storage_InsertDestination_proto
+                      quickstep_storage_WindowAggregationOperationState
                       quickstep_types_TypedValue
                       quickstep_types_containers_Tuple
                       quickstep_utility_BloomFilter
@@ -130,6 +131,7 @@ target_link_libraries(quickstep_queryexecution_QueryContext_proto
                       quickstep_storage_AggregationOperationState_proto
                       quickstep_storage_HashTable_proto
                       quickstep_storage_InsertDestination_proto
+                      quickstep_storage_WindowAggregationOperationState_proto
                       quickstep_types_containers_Tuple_proto
                       quickstep_utility_SortConfiguration_proto
                       ${PROTOBUF_LIBRARY})

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/7671a589/query_execution/QueryContext.cpp
----------------------------------------------------------------------
diff --git a/query_execution/QueryContext.cpp b/query_execution/QueryContext.cpp
index 54dd557..7019b6a 100644
--- a/query_execution/QueryContext.cpp
+++ b/query_execution/QueryContext.cpp
@@ -140,6 +140,13 @@ QueryContext::QueryContext(const serialization::QueryContext &proto,
 
     update_groups_.push_back(move(update_group));
   }
+
+  for (int i = 0; i < proto.window_aggregation_states_size(); ++i) {
+    window_aggregation_states_.emplace_back(
+        WindowAggregationOperationState::ReconstructFromProto(proto.window_aggregation_states(i),
+                                                              database,
+                                                              storage_manager));
+  }
 }
 
 bool QueryContext::ProtoIsValid(const serialization::QueryContext &proto,
@@ -231,6 +238,13 @@ bool QueryContext::ProtoIsValid(const serialization::QueryContext &proto,
     }
   }
 
+  for (int i = 0; i < proto.window_aggregation_states_size(); ++i) {
+    if (!WindowAggregationOperationState::ProtoIsValid(proto.window_aggregation_states(i),
+                                                       database)) {
+      return false;
+    }
+  }
+
   return proto.IsInitialized();
 }
 

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/7671a589/query_execution/QueryContext.hpp
----------------------------------------------------------------------
diff --git a/query_execution/QueryContext.hpp b/query_execution/QueryContext.hpp
index 7d5628d..9171250 100644
--- a/query_execution/QueryContext.hpp
+++ b/query_execution/QueryContext.hpp
@@ -33,6 +33,7 @@
 #include "storage/AggregationOperationState.hpp"
 #include "storage/HashTable.hpp"
 #include "storage/InsertDestination.hpp"
+#include "storage/WindowAggregationOperationState.hpp"
 #include "types/containers/Tuple.hpp"
 #include "utility/BloomFilter.hpp"
 #include "utility/Macros.hpp"
@@ -120,6 +121,11 @@ class QueryContext {
   typedef std::uint32_t update_group_id;
 
   /**
+   * @brief A unique identifier for a window aggregation state.
+   **/
+  typedef std::uint32_t window_aggregation_state_id;
+
+  /**
    * @brief Constructor.
    *
    * @param proto A serialized Protocol Buffer representation of a
@@ -460,6 +466,47 @@ class QueryContext {
     return update_groups_[id];
   }
 
+  /**
+   * @brief Whether the given WindowAggregationOperationState id is valid.
+   *
+   * @param id The WindowAggregationOperationState id.
+   *
+   * @return True if valid, otherwise false.
+   **/
+  bool isValidWindowAggregationStateId(const window_aggregation_state_id id) const {
+    return id < window_aggregation_states_.size();
+  }
+
+  /**
+   * @brief Get the WindowAggregationOperationState.
+   *
+   * @param id The WindowAggregationOperationState id in the query.
+   *
+   * @return The WindowAggregationOperationState, already created in the
+   *         constructor.
+   **/
+  inline WindowAggregationOperationState* getWindowAggregationState(
+      const window_aggregation_state_id id) {
+    DCHECK_LT(id, window_aggregation_states_.size());
+    DCHECK(window_aggregation_states_[id]);
+    return window_aggregation_states_[id].get();
+  }
+
+  /**
+   * @brief Release the given WindowAggregationOperationState.
+   *
+   * @param id The id of the WindowAggregationOperationState to destroy.
+   *
+   * @return The WindowAggregationOperationState, already created in the
+   *         constructor.
+   **/
+  inline WindowAggregationOperationState* releaseWindowAggregationState(
+      const window_aggregation_state_id id) {
+    DCHECK_LT(id, window_aggregation_states_.size());
+    DCHECK(window_aggregation_states_[id]);
+    return window_aggregation_states_[id].release();
+  }
+
  private:
   std::vector<std::unique_ptr<AggregationOperationState>> aggregation_states_;
   std::vector<std::unique_ptr<BloomFilter>> bloom_filters_;
@@ -471,6 +518,7 @@ class QueryContext {
   std::vector<std::unique_ptr<const SortConfiguration>> sort_configs_;
   std::vector<std::unique_ptr<Tuple>> tuples_;
   std::vector<std::unordered_map<attribute_id, std::unique_ptr<const Scalar>>> update_groups_;
+  std::vector<std::unique_ptr<WindowAggregationOperationState>> window_aggregation_states_;
 
   DISALLOW_COPY_AND_ASSIGN(QueryContext);
 };

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/7671a589/query_execution/QueryContext.proto
----------------------------------------------------------------------
diff --git a/query_execution/QueryContext.proto b/query_execution/QueryContext.proto
index 98cd0b6..ddf8326 100644
--- a/query_execution/QueryContext.proto
+++ b/query_execution/QueryContext.proto
@@ -22,6 +22,7 @@ import "expressions/table_generator/GeneratorFunction.proto";
 import "storage/AggregationOperationState.proto";
 import "storage/HashTable.proto";
 import "storage/InsertDestination.proto";
+import "storage/WindowAggregationOperationState.proto";
 import "types/containers/Tuple.proto";
 import "utility/BloomFilter.proto";
 import "utility/SortConfiguration.proto";
@@ -55,5 +56,7 @@ message QueryContext {
   // NOTE(zuyu): For UpdateWorkOrder only.
   repeated UpdateGroup update_groups = 10;
 
-  required uint64 query_id = 11;
+  repeated WindowAggregationOperationState window_aggregation_states = 11;
+
+  required uint64 query_id = 12;
 }

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/7671a589/query_optimizer/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/query_optimizer/CMakeLists.txt b/query_optimizer/CMakeLists.txt
index 8912414..7e53b9d 100644
--- a/query_optimizer/CMakeLists.txt
+++ b/query_optimizer/CMakeLists.txt
@@ -88,6 +88,7 @@ target_link_libraries(quickstep_queryoptimizer_ExecutionGenerator
                       quickstep_queryoptimizer_expressions_Predicate
                       quickstep_queryoptimizer_expressions_Scalar
                       quickstep_queryoptimizer_expressions_ScalarLiteral
+                      quickstep_queryoptimizer_expressions_WindowAggregateFunction
                       quickstep_queryoptimizer_physical_Aggregate
                       quickstep_queryoptimizer_physical_CopyFrom
                       quickstep_queryoptimizer_physical_CreateIndex
@@ -130,6 +131,7 @@ target_link_libraries(quickstep_queryoptimizer_ExecutionGenerator
                       quickstep_relationaloperators_TableGeneratorOperator
                       quickstep_relationaloperators_TextScanOperator
                       quickstep_relationaloperators_UpdateOperator
+                      quickstep_relationaloperators_WindowAggregationOperator
                       quickstep_storage_AggregationOperationState_proto
                       quickstep_storage_HashTableFactory
                       quickstep_storage_HashTable_proto

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/7671a589/query_optimizer/ExecutionGenerator.cpp
----------------------------------------------------------------------
diff --git a/query_optimizer/ExecutionGenerator.cpp b/query_optimizer/ExecutionGenerator.cpp
index 45f5f78..43d63f9 100644
--- a/query_optimizer/ExecutionGenerator.cpp
+++ b/query_optimizer/ExecutionGenerator.cpp
@@ -63,6 +63,7 @@
 #include "query_optimizer/expressions/PatternMatcher.hpp"
 #include "query_optimizer/expressions/Scalar.hpp"
 #include "query_optimizer/expressions/ScalarLiteral.hpp"
+#include "query_optimizer/expressions/WindowAggregateFunction.hpp"
 #include "query_optimizer/physical/CopyFrom.hpp"
 #include "query_optimizer/physical/CreateIndex.hpp"
 #include "query_optimizer/physical/CreateTable.hpp"
@@ -104,6 +105,7 @@
 #include "relational_operators/TableGeneratorOperator.hpp"
 #include "relational_operators/TextScanOperator.hpp"
 #include "relational_operators/UpdateOperator.hpp"
+#include "relational_operators/WindowAggregationOperator.hpp"
 #include "storage/AggregationOperationState.pb.h"
 #include "storage/HashTable.pb.h"
 #include "storage/HashTableFactory.hpp"
@@ -284,8 +286,8 @@ void ExecutionGenerator::generatePlanInternal(
       return convertUpdateTable(
           std::static_pointer_cast<const P::UpdateTable>(physical_plan));
     case P::PhysicalType::kWindowAggregate:
-      THROW_SQL_ERROR()
-          << "Window aggregate function is not supported yet :(";
+      return convertWindowAggregate(
+          std::static_pointer_cast<const P::WindowAggregate>(physical_plan));
     default:
       LOG(FATAL) << "Unknown physical plan node "
                  << physical_plan->getShortString();
@@ -1639,5 +1641,100 @@ void ExecutionGenerator::convertTableGenerator(
   temporary_relation_info_vec_.emplace_back(tablegen_index, output_relation);
 }
 
+void ExecutionGenerator::convertWindowAggregate(
+    const P::WindowAggregatePtr &physical_plan) {
+  // Create window_aggregation_operation_state proto.
+  const QueryContext::window_aggregation_state_id window_aggr_state_index =
+      query_context_proto_->window_aggregation_states_size();
+  S::WindowAggregationOperationState *window_aggr_state_proto =
+      query_context_proto_->add_window_aggregation_states();
+
+  // Get input.
+  const CatalogRelationInfo *input_relation_info =
+      findRelationInfoOutputByPhysical(physical_plan->input());
+  window_aggr_state_proto->set_relation_id(input_relation_info->relation->getID());
+
+  // Get window aggregate function expression.
+  const E::AliasPtr &named_window_aggregate_expression =
+      physical_plan->window_aggregate_expression();
+  const E::WindowAggregateFunctionPtr &window_aggregate_function =
+      std::static_pointer_cast<const E::WindowAggregateFunction>(
+          named_window_aggregate_expression->expression());
+
+  // Set the AggregateFunction.
+  window_aggr_state_proto->mutable_function()->MergeFrom(
+      window_aggregate_function->window_aggregate().getProto());
+
+  // Set the arguments.
+  for (const E::ScalarPtr &argument : window_aggregate_function->arguments()) {
+    unique_ptr<const Scalar> concretized_argument(argument->concretize(attribute_substitution_map_));
+    window_aggr_state_proto->add_arguments()->MergeFrom(concretized_argument->getProto());
+  }
+
+  // Set partition keys.
+  const E::WindowInfo &window_info = window_aggregate_function->window_info();
+  for (const E::ScalarPtr &partition_by_attribute
+      : window_info.partition_by_attributes) {
+    unique_ptr<const Scalar> concretized_partition_by_attribute(
+        partition_by_attribute->concretize(attribute_substitution_map_));
+    window_aggr_state_proto->add_partition_by_attributes()
+        ->MergeFrom(concretized_partition_by_attribute->getProto());
+  }
+
+  // Set window frame info.
+  if (window_info.frame_info == nullptr) {
+    // If the frame is not specified, use the default setting:
+    //   1. If ORDER BY key is specified, use cumulative aggregation:
+    //      ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW.
+    //   2. If ORDER BY key is not specified either, use the whole partition:
+    //      ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING.
+    window_aggr_state_proto->set_is_row(true);  // frame mode: ROWS.
+    window_aggr_state_proto->set_num_preceding(-1);  // UNBOUNDED PRECEDING.
+    window_aggr_state_proto->set_num_following(
+        window_info.order_by_attributes.empty()
+            ? -1  // UNBOUNDED FOLLOWING.
+            : 0);  // CURRENT ROW.
+  } else {
+    const E::WindowFrameInfo *window_frame_info = window_info.frame_info;
+    window_aggr_state_proto->set_is_row(window_frame_info->is_row);
+    window_aggr_state_proto->set_num_preceding(window_frame_info->num_preceding);
+    window_aggr_state_proto->set_num_following(window_frame_info->num_following);
+  }
+
+  // Create InsertDestination proto.
+  const CatalogRelation *output_relation = nullptr;
+  const QueryContext::insert_destination_id insert_destination_index =
+      query_context_proto_->insert_destinations_size();
+  S::InsertDestination *insert_destination_proto = query_context_proto_->add_insert_destinations();
+  createTemporaryCatalogRelation(physical_plan,
+                                 &output_relation,
+                                 insert_destination_proto);
+
+  const QueryPlan::DAGNodeIndex window_aggregation_operator_index =
+      execution_plan_->addRelationalOperator(
+          new WindowAggregationOperator(query_handle_->query_id(),
+                                        *output_relation,
+                                        window_aggr_state_index,
+                                        insert_destination_index));
+
+  // TODO(Shixuan): Once parallelism is introduced, the is_pipeline_breaker
+  //                could be set to false.
+  if (!input_relation_info->isStoredRelation()) {
+    execution_plan_->addDirectDependency(window_aggregation_operator_index,
+                                         input_relation_info->producer_operator_index,
+                                         true /* is_pipeline_breaker */);
+  }
+
+  insert_destination_proto->set_relational_op_index(window_aggregation_operator_index);
+
+  // Add to map and temp_relation_info_vec.
+  physical_to_output_relation_map_.emplace(
+      std::piecewise_construct,
+      std::forward_as_tuple(physical_plan),
+      std::forward_as_tuple(window_aggregation_operator_index, output_relation));
+  temporary_relation_info_vec_.emplace_back(window_aggregation_operator_index,
+                                            output_relation);
+}
+
 }  // namespace optimizer
 }  // namespace quickstep

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/7671a589/query_optimizer/ExecutionGenerator.hpp
----------------------------------------------------------------------
diff --git a/query_optimizer/ExecutionGenerator.hpp b/query_optimizer/ExecutionGenerator.hpp
index c7fd018..9186707 100644
--- a/query_optimizer/ExecutionGenerator.hpp
+++ b/query_optimizer/ExecutionGenerator.hpp
@@ -60,6 +60,7 @@
 #include "query_optimizer/physical/TableReference.hpp"
 #include "query_optimizer/physical/TopLevelPlan.hpp"
 #include "query_optimizer/physical/UpdateTable.hpp"
+#include "query_optimizer/physical/WindowAggregate.hpp"
 #include "utility/Macros.hpp"
 
 #include "glog/logging.h"
@@ -347,6 +348,13 @@ class ExecutionGenerator {
   void convertTableGenerator(const physical::TableGeneratorPtr &physical_plan);
 
   /**
+   * @brief Converts a physical WindowAggregate to a WindowAggregation operator.
+   *
+   * @param physical_plan The WindowAggregate to be converted.
+   */
+  void convertWindowAggregate(const physical::WindowAggregatePtr &physical_plan);
+
+  /**
    * @brief Converts a list of NamedExpressions in the optimizer expression
    *        system to scalars proto in QueryContext proto.
    *

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/7671a589/query_optimizer/cost_model/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/query_optimizer/cost_model/CMakeLists.txt b/query_optimizer/cost_model/CMakeLists.txt
index 6bf5240..5d5b596 100644
--- a/query_optimizer/cost_model/CMakeLists.txt
+++ b/query_optimizer/cost_model/CMakeLists.txt
@@ -40,6 +40,7 @@ target_link_libraries(quickstep_queryoptimizer_costmodel_SimpleCostModel
                       quickstep_queryoptimizer_physical_TableGenerator
                       quickstep_queryoptimizer_physical_TableReference
                       quickstep_queryoptimizer_physical_TopLevelPlan
+                      quickstep_queryoptimizer_physical_WindowAggregate
                       quickstep_utility_Macros)
 target_link_libraries(quickstep_queryoptimizer_costmodel_StarSchemaSimpleCostModel
                       glog
@@ -64,6 +65,7 @@ target_link_libraries(quickstep_queryoptimizer_costmodel_StarSchemaSimpleCostMod
                       quickstep_queryoptimizer_physical_TableGenerator
                       quickstep_queryoptimizer_physical_TableReference
                       quickstep_queryoptimizer_physical_TopLevelPlan
+                      quickstep_queryoptimizer_physical_WindowAggregate
                       quickstep_utility_Macros)
 
 # Module all-in-one library:

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/7671a589/query_optimizer/cost_model/SimpleCostModel.cpp
----------------------------------------------------------------------
diff --git a/query_optimizer/cost_model/SimpleCostModel.cpp b/query_optimizer/cost_model/SimpleCostModel.cpp
index 48f76fa..e5222ff 100644
--- a/query_optimizer/cost_model/SimpleCostModel.cpp
+++ b/query_optimizer/cost_model/SimpleCostModel.cpp
@@ -33,6 +33,7 @@
 #include "query_optimizer/physical/TableGenerator.hpp"
 #include "query_optimizer/physical/TableReference.hpp"
 #include "query_optimizer/physical/TopLevelPlan.hpp"
+#include "query_optimizer/physical/WindowAggregate.hpp"
 
 #include "glog/logging.h"
 
@@ -72,6 +73,9 @@ std::size_t SimpleCostModel::estimateCardinality(
       return estimateCardinality(
           shared_subplans_[shared_subplan_reference->subplan_id()]);
     }
+    case P::PhysicalType::kWindowAggregate:
+      return estimateCardinalityForWindowAggregate(
+          std::static_pointer_cast<const P::WindowAggregate>(physical_plan));
     default:
       LOG(FATAL) << "Unsupported physical plan:" << physical_plan->toString();
   }
@@ -118,6 +122,11 @@ std::size_t SimpleCostModel::estimateCardinalityForAggregate(
                   estimateCardinality(physical_plan->input()) / 10);
 }
 
+std::size_t SimpleCostModel::estimateCardinalityForWindowAggregate(
+    const physical::WindowAggregatePtr &physical_plan) {
+  return estimateCardinality(physical_plan->input());
+}
+
 }  // namespace cost
 }  // namespace optimizer
 }  // namespace quickstep

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/7671a589/query_optimizer/cost_model/SimpleCostModel.hpp
----------------------------------------------------------------------
diff --git a/query_optimizer/cost_model/SimpleCostModel.hpp b/query_optimizer/cost_model/SimpleCostModel.hpp
index 9862198..9837039 100644
--- a/query_optimizer/cost_model/SimpleCostModel.hpp
+++ b/query_optimizer/cost_model/SimpleCostModel.hpp
@@ -32,6 +32,7 @@
 #include "query_optimizer/physical/TableGenerator.hpp"
 #include "query_optimizer/physical/TableReference.hpp"
 #include "query_optimizer/physical/TopLevelPlan.hpp"
+#include "query_optimizer/physical/WindowAggregate.hpp"
 #include "utility/Macros.hpp"
 
 namespace quickstep {
@@ -88,6 +89,10 @@ class SimpleCostModel : public CostModel {
   std::size_t estimateCardinalityForAggregate(
       const physical::AggregatePtr &physical_plan);
 
+  // Return the estimated cardinality of the input plan.
+  std::size_t estimateCardinalityForWindowAggregate(
+      const physical::WindowAggregatePtr &physical_plan);
+
   const std::vector<physical::PhysicalPtr> &shared_subplans_;
 
   DISALLOW_COPY_AND_ASSIGN(SimpleCostModel);

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/7671a589/query_optimizer/cost_model/StarSchemaSimpleCostModel.cpp
----------------------------------------------------------------------
diff --git a/query_optimizer/cost_model/StarSchemaSimpleCostModel.cpp b/query_optimizer/cost_model/StarSchemaSimpleCostModel.cpp
index eb9fcc1..badfeb1 100644
--- a/query_optimizer/cost_model/StarSchemaSimpleCostModel.cpp
+++ b/query_optimizer/cost_model/StarSchemaSimpleCostModel.cpp
@@ -85,6 +85,9 @@ std::size_t StarSchemaSimpleCostModel::estimateCardinality(
     case P::PhysicalType::kSort:
       return estimateCardinality(
           std::static_pointer_cast<const P::Sort>(physical_plan)->input());
+    case P::PhysicalType::kWindowAggregate:
+      return estimateCardinalityForWindowAggregate(
+          std::static_pointer_cast<const P::WindowAggregate>(physical_plan));
     default:
       LOG(FATAL) << "Unsupported physical plan:" << physical_plan->toString();
   }
@@ -141,6 +144,11 @@ std::size_t StarSchemaSimpleCostModel::estimateCardinalityForAggregate(
                   estimateCardinality(physical_plan->input()) / 10);
 }
 
+std::size_t StarSchemaSimpleCostModel::estimateCardinalityForWindowAggregate(
+    const P::WindowAggregatePtr &physical_plan) {
+  return estimateCardinality(physical_plan->input());
+}
+
 double StarSchemaSimpleCostModel::estimateSelectivity(
     const physical::PhysicalPtr &physical_plan) {
   switch (physical_plan->getPhysicalType()) {

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/7671a589/query_optimizer/cost_model/StarSchemaSimpleCostModel.hpp
----------------------------------------------------------------------
diff --git a/query_optimizer/cost_model/StarSchemaSimpleCostModel.hpp b/query_optimizer/cost_model/StarSchemaSimpleCostModel.hpp
index c63e55a..83032cf 100644
--- a/query_optimizer/cost_model/StarSchemaSimpleCostModel.hpp
+++ b/query_optimizer/cost_model/StarSchemaSimpleCostModel.hpp
@@ -33,6 +33,7 @@
 #include "query_optimizer/physical/TableGenerator.hpp"
 #include "query_optimizer/physical/TableReference.hpp"
 #include "query_optimizer/physical/TopLevelPlan.hpp"
+#include "query_optimizer/physical/WindowAggregate.hpp"
 #include "utility/Macros.hpp"
 
 namespace quickstep {
@@ -94,6 +95,9 @@ class StarSchemaSimpleCostModel : public CostModel {
   std::size_t estimateCardinalityForAggregate(
       const physical::AggregatePtr &physical_plan);
 
+  std::size_t estimateCardinalityForWindowAggregate(
+      const physical::WindowAggregatePtr &physical_plan);
+
   double estimateSelectivityForSelection(
       const physical::SelectionPtr &physical_plan);
 

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/7671a589/query_optimizer/resolver/Resolver.cpp
----------------------------------------------------------------------
diff --git a/query_optimizer/resolver/Resolver.cpp b/query_optimizer/resolver/Resolver.cpp
index c07751a..f10378b 100644
--- a/query_optimizer/resolver/Resolver.cpp
+++ b/query_optimizer/resolver/Resolver.cpp
@@ -1860,12 +1860,17 @@ L::LogicalPtr Resolver::resolveJoinedTableReference(
 L::LogicalPtr Resolver::resolveSortInWindow(
     const L::LogicalPtr &logical_plan,
     const E::WindowInfo &window_info) {
-  // Sort the table by (p_key, o_key)
+  // Sort the table by (p_key, o_key).
   std::vector<E::AttributeReferencePtr> sort_attributes(window_info.partition_by_attributes);
   sort_attributes.insert(sort_attributes.end(),
                          window_info.order_by_attributes.begin(),
                          window_info.order_by_attributes.end());
 
+  // If (p_key, o_key) is empty, no sort is needed.
+  if (sort_attributes.empty()) {
+    return logical_plan;
+  }
+
   std::vector<bool> sort_directions(
       window_info.partition_by_attributes.size(), true);
   sort_directions.insert(sort_directions.end(),

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/7671a589/query_optimizer/tests/execution_generator/Select.test
----------------------------------------------------------------------
diff --git a/query_optimizer/tests/execution_generator/Select.test b/query_optimizer/tests/execution_generator/Select.test
index 05f7108..16127cc 100644
--- a/query_optimizer/tests/execution_generator/Select.test
+++ b/query_optimizer/tests/execution_generator/Select.test
@@ -950,3 +950,42 @@ WHERE double_col < 0
 +--------------------+
 |                   5|
 +--------------------+
+==
+
+# Window Aggregation Test.
+# Currently this is not supported, an empty table will be returned.
+SELECT avg(int_col) OVER w FROM test
+WINDOW w AS
+(PARTITION BY char_col
+ ORDER BY long_col DESC NULLS LAST
+ ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW);
+--
++------------------------+
+|avg(int_col)            |
++------------------------+
++------------------------+
+==
+
+SELECT int_col, sum(float_col) OVER
+(PARTITION BY char_col, long_col
+ ORDER BY double_col DESC NULLS LAST, int_col ASC NULLS FIRST
+ RANGE BETWEEN 3 PRECEDING AND 3 FOLLOWING)
+FROM test;
+--
++-----------+------------------------+
+|int_col    |sum(float_col)          |
++-----------+------------------------+
++-----------+------------------------+
+==
+
+SELECT sum(avg(int_col) OVER w) FROM test
+WINDOW w AS
+(PARTITION BY char_col
+ ORDER BY long_col
+ ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW);
+--
++------------------------+
+|sum(avg(int_col))       |
++------------------------+
+|                    NULL|
++------------------------+

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/7671a589/query_optimizer/tests/resolver/Select.test
----------------------------------------------------------------------
diff --git a/query_optimizer/tests/resolver/Select.test b/query_optimizer/tests/resolver/Select.test
index 89ab84d..5e11ac0 100644
--- a/query_optimizer/tests/resolver/Select.test
+++ b/query_optimizer/tests/resolver/Select.test
@@ -3269,6 +3269,39 @@ TopLevelPlan
     type=Double NULL]
 ==
 
+SELECT avg(int_col) OVER w FROM test
+WINDOW w AS
+(ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING);
+--
+TopLevelPlan
++-plan=Project
+| +-input=WindowAggregate
+| | +-input=TableReference[relation_name=Test,relation_alias=test]
+| | | +-AttributeReference[id=0,name=int_col,relation=test,type=Int NULL]
+| | | +-AttributeReference[id=1,name=long_col,relation=test,type=Long]
+| | | +-AttributeReference[id=2,name=float_col,relation=test,type=Float]
+| | | +-AttributeReference[id=3,name=double_col,relation=test,type=Double NULL]
+| | | +-AttributeReference[id=4,name=char_col,relation=test,type=Char(20)]
+| | | +-AttributeReference[id=5,name=vchar_col,relation=test,
+| | |   type=VarChar(20) NULL]
+| | +-window_aggregate_expression=Alias[id=6,name=,alias=$window_aggregate0,
+| |   relation=$window_aggregate,type=Double NULL]
+| |   +-WindowAggregateFunction[function=AVG,window_name=w,is_ascending=[],
+| |     nulls_first=[],frame_mode=row,num_preceding=-1,num_following=-1]
+| |     +-arguments=
+| |     | +-AttributeReference[id=0,name=int_col,relation=test,type=Int NULL]
+| |     +-partition_by=
+| |     | +-[]
+| |     +-order_by=
+| |       +-[]
+| +-project_list=
+|   +-Alias[id=6,name=,alias=avg(int_col),relation=,type=Double NULL]
+|     +-AttributeReference[id=6,name=,alias=$window_aggregate0,
+|       relation=$window_aggregate,type=Double NULL]
++-output_attributes=
+  +-AttributeReference[id=6,name=,alias=avg(int_col),relation=,type=Double NULL]
+==
+
 SELECT int_col, sum(float_col) OVER w1 FROM test
 WINDOW w2 AS
 (PARTITION BY vchar_col, long_col

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/7671a589/relational_operators/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/relational_operators/CMakeLists.txt b/relational_operators/CMakeLists.txt
index 91d1097..249441d 100644
--- a/relational_operators/CMakeLists.txt
+++ b/relational_operators/CMakeLists.txt
@@ -61,6 +61,7 @@ add_library(quickstep_relationaloperators_SortRunGenerationOperator SortRunGener
 add_library(quickstep_relationaloperators_TableGeneratorOperator TableGeneratorOperator.cpp TableGeneratorOperator.hpp)
 add_library(quickstep_relationaloperators_TextScanOperator TextScanOperator.cpp TextScanOperator.hpp)
 add_library(quickstep_relationaloperators_UpdateOperator UpdateOperator.cpp UpdateOperator.hpp)
+add_library(quickstep_relationaloperators_WindowAggregationOperator WindowAggregationOperator.cpp WindowAggregationOperator.hpp)
 add_library(quickstep_relationaloperators_WorkOrder ../empty_src.cpp WorkOrder.hpp)
 add_library(quickstep_relationaloperators_WorkOrderFactory WorkOrderFactory.cpp WorkOrderFactory.hpp)
 add_library(quickstep_relationaloperators_WorkOrder_proto
@@ -423,6 +424,18 @@ target_link_libraries(quickstep_relationaloperators_UpdateOperator
                       quickstep_threading_ThreadIDBasedMap
                       quickstep_utility_Macros
                       tmb)
+target_link_libraries(quickstep_relationaloperators_WindowAggregationOperator
+                      glog
+                      quickstep_catalog_CatalogRelation
+                      quickstep_queryexecution_QueryContext
+                      quickstep_queryexecution_WorkOrderProtosContainer
+                      quickstep_queryexecution_WorkOrdersContainer
+                      quickstep_relationaloperators_RelationalOperator
+                      quickstep_relationaloperators_WorkOrder
+                      quickstep_relationaloperators_WorkOrder_proto
+                      quickstep_storage_StorageBlockInfo
+                      quickstep_utility_Macros
+                      tmb)                      
 target_link_libraries(quickstep_relationaloperators_WorkOrder
                       glog
                       quickstep_queryexecution_QueryExecutionTypedefs
@@ -487,6 +500,7 @@ target_link_libraries(quickstep_relationaloperators
                       quickstep_relationaloperators_TableGeneratorOperator
                       quickstep_relationaloperators_TextScanOperator
                       quickstep_relationaloperators_UpdateOperator
+                      quickstep_relationaloperators_WindowAggregationOperator
                       quickstep_relationaloperators_WorkOrder
                       quickstep_relationaloperators_WorkOrderFactory
                       quickstep_relationaloperators_WorkOrder_proto)

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/7671a589/relational_operators/WindowAggregationOperator.cpp
----------------------------------------------------------------------
diff --git a/relational_operators/WindowAggregationOperator.cpp b/relational_operators/WindowAggregationOperator.cpp
new file mode 100644
index 0000000..93cb9d4
--- /dev/null
+++ b/relational_operators/WindowAggregationOperator.cpp
@@ -0,0 +1,82 @@
+/**
+ *   Copyright 2011-2015 Quickstep Technologies LLC.
+ *   Copyright 2015-2016 Pivotal Software, Inc.
+ *   Copyright 2016, Quickstep Research Group, Computer Sciences Department,
+ *     University of Wisconsin—Madison.
+ *
+ *   Licensed 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.
+ **/
+
+#include "relational_operators/WindowAggregationOperator.hpp"
+
+#include <vector>
+
+#include "query_execution/QueryContext.hpp"
+#include "query_execution/WorkOrderProtosContainer.hpp"
+#include "query_execution/WorkOrdersContainer.hpp"
+#include "relational_operators/WorkOrder.pb.h"
+#include "storage/StorageBlockInfo.hpp"
+
+#include "tmb/id_typedefs.h"
+
+namespace quickstep {
+
+bool WindowAggregationOperator::getAllWorkOrders(
+    WorkOrdersContainer *container,
+    QueryContext *query_context,
+    StorageManager *storage_manager,
+    const tmb::client_id scheduler_client_id,
+    tmb::MessageBus *bus) {
+  DCHECK(query_context != nullptr);
+
+  if (blocking_dependencies_met_ && !generated_) {
+    container->addNormalWorkOrder(
+        new WindowAggregationWorkOrder(
+            query_id_,
+            query_context->releaseWindowAggregationState(window_aggregation_state_index_),
+            query_context->getInsertDestination(output_destination_index_)),
+        op_index_);
+    generated_ = true;
+  }
+
+  return generated_;
+}
+
+bool WindowAggregationOperator::getAllWorkOrderProtos(WorkOrderProtosContainer *container) {
+  if (blocking_dependencies_met_ && !generated_) {
+    container->addWorkOrderProto(createWorkOrderProto(), op_index_);
+    generated_ = true;
+  }
+
+  return generated_;
+}
+
+serialization::WorkOrder* WindowAggregationOperator::createWorkOrderProto() {
+  serialization::WorkOrder *proto = new serialization::WorkOrder;
+  proto->set_work_order_type(serialization::WINDOW_AGGREGATION);
+  proto->set_query_id(query_id_);
+  proto->SetExtension(serialization::WindowAggregationWorkOrder::window_aggr_state_index,
+                      window_aggregation_state_index_);
+  proto->SetExtension(serialization::WindowAggregationWorkOrder::insert_destination_index,
+                      output_destination_index_);
+
+  return proto;
+}
+
+
+void WindowAggregationWorkOrder::execute() {
+  std::cout << "Window aggregation is not supported yet.\n"
+      << "An empty table is returned\n";
+}
+
+}  // namespace quickstep

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/7671a589/relational_operators/WindowAggregationOperator.hpp
----------------------------------------------------------------------
diff --git a/relational_operators/WindowAggregationOperator.hpp b/relational_operators/WindowAggregationOperator.hpp
new file mode 100644
index 0000000..f3dfd14
--- /dev/null
+++ b/relational_operators/WindowAggregationOperator.hpp
@@ -0,0 +1,166 @@
+/**
+ *   Copyright 2011-2015 Quickstep Technologies LLC.
+ *   Copyright 2015-2016 Pivotal Software, Inc.
+ *   Copyright 2016, Quickstep Research Group, Computer Sciences Department,
+ *     University of Wisconsin—Madison.
+ *
+ *   Licensed 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.
+ **/
+
+#ifndef QUICKSTEP_RELATIONAL_OPERATORS_WINDOW_AGGREGATION_OPERATOR_HPP_
+#define QUICKSTEP_RELATIONAL_OPERATORS_WINDOW_AGGREGATION_OPERATOR_HPP_
+
+#include <vector>
+
+#include "catalog/CatalogRelation.hpp"
+#include "query_execution/QueryContext.hpp"
+#include "relational_operators/RelationalOperator.hpp"
+#include "relational_operators/WorkOrder.hpp"
+#include "storage/StorageBlockInfo.hpp"
+#include "utility/Macros.hpp"
+
+#include "glog/logging.h"
+
+#include "tmb/id_typedefs.h"
+
+namespace tmb { class MessageBus; }
+
+namespace quickstep {
+
+class StorageManager;
+class WindowAggregationOperationState;
+class WorkOrderProtosContainer;
+class WorkOrdersContainer;
+
+namespace serialization { class WorkOrder; }
+
+/** \addtogroup RelationalOperators
+ *  @{
+ */
+
+/**
+ * @brief An operator which performs window aggregation over a relation.
+ **/
+class WindowAggregationOperator : public RelationalOperator {
+ public:
+  /**
+   * @brief Constructor.
+   *
+   * @param query_id The ID of this query.
+   * @param input_relation The relation to perform aggregation over.
+   * @param window_aggregation_state_index The index of WindowAggregationState
+   *                                       in QueryContext.
+   * @param output_destination_index The index of InsertDestination in
+   *                                 QueryContext for the output.
+   **/
+  WindowAggregationOperator(const std::size_t query_id,
+                            const CatalogRelation &output_relation,
+                            const QueryContext::window_aggregation_state_id window_aggregation_state_index,
+                            const QueryContext::insert_destination_id output_destination_index)
+      : RelationalOperator(query_id),
+        output_relation_(output_relation),
+        window_aggregation_state_index_(window_aggregation_state_index),
+        output_destination_index_(output_destination_index),
+        generated_(false) {}
+
+  ~WindowAggregationOperator() override {}
+
+  bool getAllWorkOrders(WorkOrdersContainer *container,
+                        QueryContext *query_context,
+                        StorageManager *storage_manager,
+                        const tmb::client_id scheduler_client_id,
+                        tmb::MessageBus *bus) override;
+
+  bool getAllWorkOrderProtos(WorkOrderProtosContainer *container) override;
+
+  const relation_id getOutputRelationID() const override {
+    return output_relation_.getID();
+  }
+
+  QueryContext::insert_destination_id getInsertDestinationID() const override {
+    return output_destination_index_;
+  }
+
+ private:
+  /**
+   * @brief Create Work Order proto.
+   *
+   * @return A window aggregation work order.
+   **/
+  serialization::WorkOrder* createWorkOrderProto();
+
+  const CatalogRelation &output_relation_;
+  const QueryContext::window_aggregation_state_id window_aggregation_state_index_;
+  const QueryContext::insert_destination_id output_destination_index_;
+  bool generated_;
+
+  DISALLOW_COPY_AND_ASSIGN(WindowAggregationOperator);
+};
+
+/**
+ * @brief A WorkOrder produced by WindowAggregationOperator.
+ **/
+class WindowAggregationWorkOrder : public WorkOrder {
+ public:
+  /**
+   * @brief Constructor
+   *
+   * @param query_id The ID of this query.
+   * @param state The WindowAggregationOperatorState to use.
+   * @param output_destination The InsertDestination for output.
+   **/
+  WindowAggregationWorkOrder(const std::size_t query_id,
+                             WindowAggregationOperationState *state,
+                             InsertDestination *output_destination)
+      : WorkOrder(query_id),
+        state_(state),
+        output_destination_(output_destination)  {}
+
+  ~WindowAggregationWorkOrder() override {}
+
+  /**
+   * @brief Get the pointer to WindowAggregationOperationState.
+   * @note This is a quickfix for "unused variable". After the window aggregate
+   *       functions are built, these methods might be dropped.
+   *
+   * @return A pointer to the window aggregation operation state.
+   **/
+  WindowAggregationOperationState* state() {
+    return state_;
+  }
+
+  /**
+   * @brief Get the pointer to output destination.
+   * @note This is a quickfix for "unused variable". After the window aggregate
+   *       functions are built, these methods might be dropped.
+   *
+   * @return A pointer to the output destination.
+   **/
+  InsertDestination* output_destination() {
+    return output_destination_;
+  }
+
+  void execute() override;
+
+ private:
+  WindowAggregationOperationState *state_;
+  InsertDestination *output_destination_;
+
+  DISALLOW_COPY_AND_ASSIGN(WindowAggregationWorkOrder);
+};
+
+/** @} */
+
+}  // namespace quickstep
+
+#endif  // QUICKSTEP_RELATIONAL_OPERATORS_WINDOW_AGGREGATION_OPERATOR_HPP_

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/7671a589/relational_operators/WorkOrder.proto
----------------------------------------------------------------------
diff --git a/relational_operators/WorkOrder.proto b/relational_operators/WorkOrder.proto
index 3ed065a..69dee1b 100644
--- a/relational_operators/WorkOrder.proto
+++ b/relational_operators/WorkOrder.proto
@@ -41,6 +41,7 @@ enum WorkOrderType {
   TABLE_GENERATOR = 17;
   TEXT_SCAN = 18;
   UPDATE = 19;
+  WINDOW_AGGREGATION = 20;
 }
 
 message WorkOrder {
@@ -243,3 +244,11 @@ message UpdateWorkOrder {
     optional fixed64 block_id = 325;
   }
 }
+
+message WindowAggregationWorkOrder {
+  extend WorkOrder {
+    // All required
+    optional uint32 window_aggr_state_index = 336;
+    optional int32 insert_destination_index = 337;
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/7671a589/storage/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/storage/CMakeLists.txt b/storage/CMakeLists.txt
index b536411..9df66e1 100644
--- a/storage/CMakeLists.txt
+++ b/storage/CMakeLists.txt
@@ -131,6 +131,9 @@ QS_PROTOBUF_GENERATE_CPP(storage_InsertDestination_proto_srcs
 QS_PROTOBUF_GENERATE_CPP(storage_StorageBlockLayout_proto_srcs
                          storage_StorageBlockLayout_proto_hdrs
                          StorageBlockLayout.proto)
+QS_PROTOBUF_GENERATE_CPP(storage_WindowAggregationOperationState_proto_srcs
+                         storage_WindowAggregationOperationState_proto_hdrs
+                         WindowAggregationOperationState.proto)
 
 if (ENABLE_DISTRIBUTED)
   GRPC_GENERATE_CPP(storage_DataExchange_proto_srcs
@@ -254,6 +257,11 @@ add_library(quickstep_storage_TupleReference ../empty_src.cpp TupleReference.hpp
 add_library(quickstep_storage_TupleStorageSubBlock TupleStorageSubBlock.cpp TupleStorageSubBlock.hpp)
 add_library(quickstep_storage_ValueAccessor ../empty_src.cpp ValueAccessor.hpp)
 add_library(quickstep_storage_ValueAccessorUtil ../empty_src.cpp ValueAccessorUtil.hpp)
+add_library(quickstep_storage_WindowAggregationOperationState
+            WindowAggregationOperationState.hpp
+            WindowAggregationOperationState.cpp)
+add_library(quickstep_storage_WindowAggregationOperationState_proto ${storage_WindowAggregationOperationState_proto_srcs})
+
 
 # Link dependencies:
 target_link_libraries(quickstep_storage_AggregationOperationState
@@ -1038,6 +1046,27 @@ target_link_libraries(quickstep_storage_ValueAccessorUtil
                       quickstep_storage_ValueAccessor
                       quickstep_types_containers_ColumnVectorsValueAccessor
                       quickstep_utility_Macros)
+target_link_libraries(quickstep_storage_WindowAggregationOperationState
+                      glog
+                      quickstep_catalog_CatalogDatabaseLite
+                      quickstep_catalog_CatalogRelationSchema
+                      quickstep_catalog_CatalogTypedefs
+                      quickstep_expressions_ExpressionFactories
+                      quickstep_expressions_Expressions_proto
+                      quickstep_expressions_aggregation_AggregateFunction
+                      quickstep_expressions_aggregation_AggregateFunctionFactory
+                      quickstep_expressions_aggregation_AggregationHandle
+                      quickstep_expressions_aggregation_AggregationID
+                      quickstep_expressions_scalar_Scalar
+                      quickstep_expressions_scalar_ScalarAttribute
+                      quickstep_storage_StorageBlockInfo
+                      quickstep_storage_StorageManager
+                      quickstep_storage_WindowAggregationOperationState_proto
+                      quickstep_utility_Macros)
+target_link_libraries(quickstep_storage_WindowAggregationOperationState_proto
+                      quickstep_expressions_aggregation_AggregateFunction_proto
+                      quickstep_expressions_Expressions_proto
+                      ${PROTOBUF_LIBRARY})
 
 # Module all-in-one library:
 add_library(quickstep_storage ../empty_src.cpp StorageModule.hpp)
@@ -1096,7 +1125,9 @@ target_link_libraries(quickstep_storage
                       quickstep_storage_TupleReference
                       quickstep_storage_TupleStorageSubBlock
                       quickstep_storage_ValueAccessor
-                      quickstep_storage_ValueAccessorUtil)
+                      quickstep_storage_ValueAccessorUtil
+                      quickstep_storage_WindowAggregationOperationState
+                      quickstep_storage_WindowAggregationOperationState_proto)
 if (QUICKSTEP_HAVE_FILE_MANAGER_HDFS)
   target_link_libraries(quickstep_storage
                         quickstep_storage_FileManagerHdfs)
@@ -1636,6 +1667,18 @@ target_link_libraries(StorageManager_unittest
                       quickstep_storage_StorageBlockInfo
                       quickstep_storage_StorageManager
                       quickstep_utility_ShardedLockManager)
+
+add_executable(WindowAggregationOperationState_unittest
+               "${CMAKE_CURRENT_SOURCE_DIR}/tests/WindowAggregationOperationState_unittest.cpp")
+target_link_libraries(WindowAggregationOperationState_unittest
+                      gtest
+                      gtest_main
+                      quickstep_catalog_CatalogDatabase
+                      quickstep_catalog_CatalogRelation
+                      quickstep_catalog_CatalogTypedefs
+                      quickstep_storage_WindowAggregationOperationState
+                      quickstep_storage_WindowAggregationOperationState_proto
+                      ${LIBS})
 if (QUICKSTEP_HAVE_LIBNUMA)
   target_link_libraries(StorageManager_unittest
                         ${LIBNUMA_LIBRARY})

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/7671a589/storage/WindowAggregationOperationState.cpp
----------------------------------------------------------------------
diff --git a/storage/WindowAggregationOperationState.cpp b/storage/WindowAggregationOperationState.cpp
new file mode 100644
index 0000000..a0bcc37
--- /dev/null
+++ b/storage/WindowAggregationOperationState.cpp
@@ -0,0 +1,179 @@
+/**
+ *   Copyright 2011-2015 Quickstep Technologies LLC.
+ *   Copyright 2015-2016 Pivotal Software, Inc.
+ *   Copyright 2016, Quickstep Research Group, Computer Sciences Department,
+ *     University of Wisconsin—Madison.
+ *
+ *   Licensed 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.
+ **/
+
+#include "storage/WindowAggregationOperationState.hpp"
+
+#include <cstddef>
+#include <cstdio>
+#include <memory>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "catalog/CatalogDatabaseLite.hpp"
+#include "catalog/CatalogRelationSchema.hpp"
+#include "catalog/CatalogTypedefs.hpp"
+#include "expressions/ExpressionFactories.hpp"
+#include "expressions/Expressions.pb.h"
+#include "expressions/aggregation/AggregateFunction.hpp"
+#include "expressions/aggregation/AggregateFunctionFactory.hpp"
+#include "expressions/aggregation/AggregationHandle.hpp"
+#include "expressions/aggregation/AggregationID.hpp"
+#include "expressions/scalar/Scalar.hpp"
+#include "expressions/scalar/ScalarAttribute.hpp"
+#include "storage/StorageManager.hpp"
+#include "storage/WindowAggregationOperationState.pb.h"
+
+#include "glog/logging.h"
+
+namespace quickstep {
+
+WindowAggregationOperationState::WindowAggregationOperationState(
+    const CatalogRelationSchema &input_relation,
+    const AggregateFunction *window_aggregate_function,
+    std::vector<std::unique_ptr<const Scalar>> &&arguments,
+    std::vector<std::unique_ptr<const Scalar>> &&partition_by_attributes,
+    const bool is_row,
+    const std::int64_t num_preceding,
+    const std::int64_t num_following,
+    StorageManager *storage_manager)
+    : input_relation_(input_relation),
+      arguments_(std::move(arguments)),
+      partition_by_attributes_(std::move(partition_by_attributes)),
+      is_row_(is_row),
+      num_preceding_(num_preceding),
+      num_following_(num_following),
+      storage_manager_(storage_manager) {
+  // Get the Types of this window aggregate's arguments so that we can create an
+  // AggregationHandle.
+  // TODO(Shixuan): Next step: New handles for window aggregation function.
+  std::vector<const Type*> argument_types;
+  for (const std::unique_ptr<const Scalar> &argument : arguments_) {
+    argument_types.emplace_back(&argument->getType());
+  }
+
+  // Check if window aggregate function could apply to the arguments.
+  DCHECK(window_aggregate_function->canApplyToTypes(argument_types));
+
+  // Create the handle and initial state.
+  window_aggregation_handle_.reset(
+      window_aggregate_function->createHandle(argument_types));
+  window_aggregation_state_.reset(
+      window_aggregation_handle_->createInitialState());
+
+#ifdef QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_SELECTION
+  // See if all of this window aggregate's arguments are attributes in the input
+  // relation. If so, remember the attribute IDs so that we can do copy elision
+  // when actually performing the window aggregation.
+  arguments_as_attributes_.reserve(arguments_.size());
+  for (const std::unique_ptr<const Scalar> &argument : arguments_) {
+    const attribute_id argument_id = argument->getAttributeIdForValueAccessor();
+    if (argument_id == -1) {
+      arguments_as_attributes_.clear();
+      break;
+    } else {
+      DCHECK_EQ(input_relation_.getID(), argument->getRelationIdForValueAccessor());
+      arguments_as_attributes_.push_back(argument_id);
+    }
+  }
+#endif
+}
+
+WindowAggregationOperationState* WindowAggregationOperationState::ReconstructFromProto(
+    const serialization::WindowAggregationOperationState &proto,
+    const CatalogDatabaseLite &database,
+    StorageManager *storage_manager) {
+  DCHECK(ProtoIsValid(proto, database));
+
+  // Rebuild contructor arguments from their representation in 'proto'.
+  const AggregateFunction *aggregate_function
+      = &AggregateFunctionFactory::ReconstructFromProto(proto.function());
+
+  std::vector<std::unique_ptr<const Scalar>> arguments;
+  arguments.reserve(proto.arguments_size());
+  for (int argument_idx = 0; argument_idx < proto.arguments_size(); ++argument_idx) {
+    arguments.emplace_back(ScalarFactory::ReconstructFromProto(
+        proto.arguments(argument_idx),
+        database));
+  }
+
+  std::vector<std::unique_ptr<const Scalar>> partition_by_attributes;
+  for (int attribute_idx = 0;
+       attribute_idx < proto.partition_by_attributes_size();
+       ++attribute_idx) {
+    partition_by_attributes.emplace_back(ScalarFactory::ReconstructFromProto(
+        proto.partition_by_attributes(attribute_idx),
+        database));
+  }
+
+  const bool is_row = proto.is_row();
+  const std::int64_t num_preceding = proto.num_preceding();
+  const std::int64_t num_following = proto.num_following();
+
+  return new WindowAggregationOperationState(database.getRelationSchemaById(proto.relation_id()),
+                                             aggregate_function,
+                                             std::move(arguments),
+                                             std::move(partition_by_attributes),
+                                             is_row,
+                                             num_preceding,
+                                             num_following,
+                                             storage_manager);
+}
+
+bool WindowAggregationOperationState::ProtoIsValid(const serialization::WindowAggregationOperationState &proto,
+                                                   const CatalogDatabaseLite &database) {
+  if (!proto.IsInitialized() ||
+      !database.hasRelationWithId(proto.relation_id())) {
+    return false;
+  }
+
+  if (!AggregateFunctionFactory::ProtoIsValid(proto.function())) {
+    return false;
+  }
+
+  // TODO(chasseur): We may also want to check that the specified
+  // AggregateFunction is applicable to the specified arguments, but that
+  // requires partial deserialization and may be too heavyweight for this
+  // method.
+  // TODO(Shixuan): The TODO for AggregateFunction could also be applied here.
+  for (int argument_idx = 0;
+       argument_idx < proto.arguments_size();
+       ++argument_idx) {
+    if (!ScalarFactory::ProtoIsValid(proto.arguments(argument_idx), database)) {
+      return false;
+    }
+  }
+
+  for (int attribute_idx = 0;
+       attribute_idx < proto.partition_by_attributes_size();
+       ++attribute_idx) {
+    if (!ScalarFactory::ProtoIsValid(proto.partition_by_attributes(attribute_idx),
+                                     database)) {
+      return false;
+    }
+  }
+
+  if (proto.num_preceding() < -1 || proto.num_following() < -1) {
+    return false;
+  }
+
+  return true;
+}
+
+}  // namespace quickstep

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/7671a589/storage/WindowAggregationOperationState.hpp
----------------------------------------------------------------------
diff --git a/storage/WindowAggregationOperationState.hpp b/storage/WindowAggregationOperationState.hpp
new file mode 100644
index 0000000..d7b3e6a
--- /dev/null
+++ b/storage/WindowAggregationOperationState.hpp
@@ -0,0 +1,177 @@
+/**
+ *   Copyright 2011-2015 Quickstep Technologies LLC.
+ *   Copyright 2015-2016 Pivotal Software, Inc.
+ *   Copyright 2016, Quickstep Research Group, Computer Sciences Department,
+ *     University of Wisconsin—Madison.
+ *
+ *   Licensed 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.
+ **/
+
+#ifndef QUICKSTEP_STORAGE_WINDOW_AGGREGATION_OPERATION_STATE_HPP_
+#define QUICKSTEP_STORAGE_WINDOW_AGGREGATION_OPERATION_STATE_HPP_
+
+#include <cstddef>
+#include <memory>
+#include <vector>
+
+#include "catalog/CatalogTypedefs.hpp"
+#include "expressions/aggregation/AggregationHandle.hpp"
+#include "expressions/scalar/Scalar.hpp"
+#include "expressions/scalar/ScalarAttribute.hpp"
+#include "storage/StorageBlockInfo.hpp"
+#include "storage/WindowAggregationOperationState.pb.h"
+#include "utility/Macros.hpp"
+
+namespace quickstep {
+
+class AggregateFunction;
+class CatalogDatabaseLite;
+class CatalogRelationSchema;
+class InsertDestination;
+class StorageManager;
+
+/** \addtogroup Storage
+ *  @{
+ */
+
+/**
+ * @brief Helper class for maintaining the state of window aggregation. 
+ **/
+class WindowAggregationOperationState {
+ public:
+  /**
+   * @brief Constructor for window aggregation operation state.
+   *
+   * @param input_relation Input relation on which window aggregation is computed.
+   * @param window_aggregate_functions The window aggregate function to be
+   *                                   computed.
+   * @param arguments A list of argument expressions to that aggregate.
+   * @param partition_by_attributes A list of window partition key.
+   * @param is_row True if the window frame is calculated by ROW, false if it is
+   *               calculated by RANGE.
+   * @param num_preceding The number of rows/range for the tuples preceding the
+   *                      current row. -1 means UNBOUNDED PRECEDING.
+   * @param num_following The number of rows/range for the tuples following the
+   *                      current row. -1 means UNBOUNDED FOLLOWING.
+   * @param storage_manager The StorageManager to use for allocating hash
+   *        tables.
+   */
+  WindowAggregationOperationState(const CatalogRelationSchema &input_relation,
+                                  const AggregateFunction *window_aggregate_function,
+                                  std::vector<std::unique_ptr<const Scalar>> &&arguments,
+                                  std::vector<std::unique_ptr<const Scalar>> &&partition_by_attributes,
+                                  const bool is_row,
+                                  const std::int64_t num_preceding,
+                                  const std::int64_t num_following,
+                                  StorageManager *storage_manager);
+
+  ~WindowAggregationOperationState() {}
+
+  /**
+   * @brief Generate the window aggregation operation state from the serialized
+   *        Protocol Buffer representation.
+   *
+   * @param proto A serialized protocol buffer representation of a
+   *        WindowAggregationOperationState, originally generated by the
+   *        optimizer.
+   * @param database The database for resolving relation and attribute
+   *        references.
+   * @param storage_manager The StorageManager to use.
+   **/
+  static WindowAggregationOperationState* ReconstructFromProto(
+      const serialization::WindowAggregationOperationState &proto,
+      const CatalogDatabaseLite &database,
+      StorageManager *storage_manager);
+
+  /**
+   * @brief Check whether a serialization::AggregationOperationState is
+   *        fully-formed and all parts are valid.
+   *
+   * @param proto A serialized Protocol Buffer representation of an
+   *        AggregationOperationState, originally generated by the optimizer.
+   * @param database The Database to resolve relation and attribute references
+   *        in.
+   * @return Whether proto is fully-formed and valid.
+   **/
+  static bool ProtoIsValid(const serialization::WindowAggregationOperationState &proto,
+                           const CatalogDatabaseLite &database);
+
+  /**
+   * @brief Get the is_row info.
+   * @note This is a quickfix for "unused variable". After the window aggregate
+   *       functions are built, these methods might be dropped.
+   * 
+   * @return True if the frame mode is ROW, false if it is RANGE.
+   **/
+  const bool is_row() const { return is_row_; }
+
+  /**
+   * @brief Get the num_preceding info.
+   * @note This is a quickfix for "unused variable". After the window aggregate
+   *       functions are built, these methods might be dropped.
+   *
+   * @return The number of rows/range that precedes the current row.
+   **/
+  const std::int64_t num_preceding() const { return num_preceding_; }
+
+  /**
+   * @brief Get the num_following info.
+   * @note This is a quickfix for "unused variable". After the window aggregate
+   *       functions are built, these methods might be dropped.
+   *
+   * @return The number of rows/range that follows the current row.
+   **/
+  const std::int64_t num_following() const { return num_following_; }
+
+  /**
+   * @brief Get the pointer to StorageManager.
+   * @note This is a quickfix for "unused variable". After the window aggregate
+   *       functions are built, these methods might be dropped.
+   *
+   * @return A pointer to the storage manager.
+   **/
+  StorageManager *storage_manager() { return storage_manager_; }
+
+ private:
+  const CatalogRelationSchema &input_relation_;
+
+  // TODO(Shixuan): Handle and State for window aggregation will be needed for
+  //                actual calculation.
+  std::unique_ptr<AggregationHandle> window_aggregation_handle_;
+  std::unique_ptr<AggregationState> window_aggregation_state_;
+  std::vector<std::unique_ptr<const Scalar>> arguments_;
+
+  // We don't add order_by_attributes here since it is not needed after sorting.
+  std::vector<std::unique_ptr<const Scalar>> partition_by_attributes_;
+
+  // Window framing information.
+  const bool is_row_;
+  const std::int64_t num_preceding_;
+  const std::int64_t num_following_;
+
+  StorageManager *storage_manager_;
+
+#ifdef QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_SELECTION
+  // If all an aggregate's argument expressions are simply attributes in
+  // 'input_relation_', then this caches the attribute IDs of those arguments.
+  std::vector<attribute_id> arguments_as_attributes_;
+#endif
+
+  DISALLOW_COPY_AND_ASSIGN(WindowAggregationOperationState);
+};
+
+/** @} */
+
+}  // namespace quickstep
+
+#endif  // QUICKSTEP_STORAGE_WINDOW_AGGREGATION_OPERATION_STATE_HPP_

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/7671a589/storage/WindowAggregationOperationState.proto
----------------------------------------------------------------------
diff --git a/storage/WindowAggregationOperationState.proto b/storage/WindowAggregationOperationState.proto
new file mode 100644
index 0000000..c7bd0ef
--- /dev/null
+++ b/storage/WindowAggregationOperationState.proto
@@ -0,0 +1,33 @@
+//   Copyright 2011-2015 Quickstep Technologies LLC.
+//   Copyright 2015-2016 Pivotal Software, Inc.
+//   Copyright 2016, Quickstep Research Group, Computer Sciences Department,
+//     University of Wisconsin—Madison.
+//
+//   Licensed 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.
+
+syntax = "proto2";
+
+package quickstep.serialization;
+
+import "expressions/aggregation/AggregateFunction.proto";
+import "expressions/Expressions.proto";
+
+message WindowAggregationOperationState {
+  required int32 relation_id = 1;
+  required AggregateFunction function = 2;
+  repeated Scalar arguments = 3;
+  repeated Scalar partition_by_attributes = 4;
+  required bool is_row = 5;
+  required int64 num_preceding = 6;  // -1 means UNBOUNDED PRECEDING.
+  required int64 num_following = 7;  // -1 means UNBOUNDED FOLLOWING.
+}

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/7671a589/storage/tests/WindowAggregationOperationState_unittest.cpp
----------------------------------------------------------------------
diff --git a/storage/tests/WindowAggregationOperationState_unittest.cpp b/storage/tests/WindowAggregationOperationState_unittest.cpp
new file mode 100644
index 0000000..c572034
--- /dev/null
+++ b/storage/tests/WindowAggregationOperationState_unittest.cpp
@@ -0,0 +1,92 @@
+/**
+ *   Copyright 2011-2015 Quickstep Technologies LLC.
+ *   Copyright 2015-2016 Pivotal Software, Inc.
+ *   Copyright 2016, Quickstep Research Group, Computer Sciences Department,
+ *     University of Wisconsin—Madison.
+ *
+ *   Licensed 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.
+ **/
+
+#include <cstddef>
+#include <memory>
+
+#include "catalog/CatalogDatabase.hpp"
+#include "catalog/CatalogRelation.hpp"
+#include "catalog/CatalogTypedefs.hpp"
+#include "expressions/aggregation/AggregateFunction.pb.h"
+#include "storage/WindowAggregationOperationState.hpp"
+#include "storage/WindowAggregationOperationState.pb.h"
+
+#include "gtest/gtest.h"
+
+using std::unique_ptr;
+
+namespace quickstep {
+
+namespace {
+  constexpr relation_id kInvalidTableId = 100;
+  constexpr std::int64_t kInvalidNum = -10;
+  constexpr std::int64_t kValidNum = 10;
+}  // namespace
+
+class WindowAggregationOperationStateProtoTest : public ::testing::Test {
+ protected:
+  virtual void SetUp() {
+    database_.reset(new CatalogDatabase(nullptr, "db"));
+    rel_id_ = database_->addRelation(new CatalogRelation(nullptr, "rel"));
+  }
+
+  unique_ptr<CatalogDatabase> database_;
+  relation_id rel_id_;
+};
+
+TEST_F(WindowAggregationOperationStateProtoTest, UninitializationTest) {
+  serialization::WindowAggregationOperationState proto;
+  EXPECT_FALSE(WindowAggregationOperationState::ProtoIsValid(proto, *database_.get()));
+}
+
+TEST_F(WindowAggregationOperationStateProtoTest, InvalidRelationIdTest) {
+  serialization::WindowAggregationOperationState proto;
+  proto.set_relation_id(kInvalidTableId);
+  proto.mutable_function()->set_aggregation_id(serialization::AggregateFunction::AVG);
+  proto.set_is_row(true);
+  proto.set_num_preceding(kValidNum);
+  proto.set_num_following(kValidNum);
+  EXPECT_FALSE(WindowAggregationOperationState::ProtoIsValid(proto, *database_.get()));
+}
+
+TEST_F(WindowAggregationOperationStateProtoTest, InvalidNumTest) {
+  serialization::WindowAggregationOperationState proto;
+  proto.set_relation_id(rel_id_);
+  proto.mutable_function()->set_aggregation_id(serialization::AggregateFunction::AVG);
+  proto.set_is_row(true);
+  proto.set_num_preceding(kInvalidNum);
+  proto.set_num_following(kValidNum);
+  EXPECT_FALSE(WindowAggregationOperationState::ProtoIsValid(proto, *database_.get()));
+
+  proto.set_num_preceding(kValidNum);
+  proto.set_num_following(kInvalidNum);
+  EXPECT_FALSE(WindowAggregationOperationState::ProtoIsValid(proto, *database_.get()));
+}
+
+TEST_F(WindowAggregationOperationStateProtoTest, ValidTest) {
+  serialization::WindowAggregationOperationState proto;
+  proto.set_relation_id(rel_id_);
+  proto.mutable_function()->set_aggregation_id(serialization::AggregateFunction::AVG);
+  proto.set_is_row(true);
+  proto.set_num_preceding(kValidNum);
+  proto.set_num_following(kValidNum);
+  EXPECT_TRUE(WindowAggregationOperationState::ProtoIsValid(proto, *database_.get()));
+}
+
+}  // namespace quickstep


Mime
View raw message