quickstep-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From zu...@apache.org
Subject [50/50] [abbrv] incubator-quickstep git commit: Serialized WorkOrders as proto.
Date Mon, 30 May 2016 23:19:25 GMT
Serialized WorkOrders as proto.


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

Branch: refs/heads/work-order-serialization
Commit: 3e35844f8f3c879b6fa89e331a73b5413d83f419
Parents: 50b4e55
Author: Zuyu Zhang <zzhang@pivotal.io>
Authored: Tue Apr 12 16:55:48 2016 -0700
Committer: Zuyu Zhang <zzhang@pivotal.io>
Committed: Mon May 30 16:17:09 2016 -0700

----------------------------------------------------------------------
 query_execution/CMakeLists.txt                  |   6 +
 query_execution/WorkOrderProtosContainer.hpp    | 147 +++++++++++++++++++
 query_execution/tests/Foreman_unittest.cpp      |   6 +
 query_execution/tests/QueryManager_unittest.cpp |   6 +
 relational_operators/AggregationOperator.cpp    |  33 +++++
 relational_operators/AggregationOperator.hpp    |  12 ++
 relational_operators/BuildHashOperator.cpp      |  39 +++++
 relational_operators/BuildHashOperator.hpp      |  12 ++
 relational_operators/CMakeLists.txt             |  36 ++++-
 relational_operators/CreateIndexOperator.hpp    |   9 ++
 relational_operators/CreateTableOperator.hpp    |   8 +
 relational_operators/DeleteOperator.cpp         |  37 ++++-
 relational_operators/DeleteOperator.hpp         |  26 +++-
 relational_operators/DestroyHashOperator.cpp    |  16 ++
 relational_operators/DestroyHashOperator.hpp    |   3 +
 relational_operators/DropTableOperator.cpp      |  22 +++
 relational_operators/DropTableOperator.hpp      |   3 +
 .../FinalizeAggregationOperator.cpp             |  19 +++
 .../FinalizeAggregationOperator.hpp             |   3 +
 relational_operators/HashJoinOperator.cpp       | 123 ++++++++++++++++
 relational_operators/HashJoinOperator.hpp       |  21 +++
 relational_operators/InsertOperator.cpp         |  18 +++
 relational_operators/InsertOperator.hpp         |   3 +
 .../NestedLoopsJoinOperator.cpp                 | 141 ++++++++++++++++++
 .../NestedLoopsJoinOperator.hpp                 |  51 +++++++
 relational_operators/RelationalOperator.hpp     |  22 +++
 relational_operators/SampleOperator.cpp         | 100 ++++++++++---
 relational_operators/SampleOperator.hpp         |  12 ++
 relational_operators/SaveBlocksOperator.cpp     |  17 +++
 relational_operators/SaveBlocksOperator.hpp     |   3 +
 relational_operators/SelectOperator.cpp         |  42 ++++++
 relational_operators/SelectOperator.hpp         |  12 ++
 relational_operators/SortMergeRunOperator.cpp   |  67 +++++++++
 relational_operators/SortMergeRunOperator.hpp   |  12 ++
 .../SortRunGenerationOperator.cpp               |  38 +++++
 .../SortRunGenerationOperator.hpp               |  12 ++
 relational_operators/TableGeneratorOperator.cpp |  19 ++-
 relational_operators/TableGeneratorOperator.hpp |   5 +-
 relational_operators/TextScanOperator.cpp       |  77 ++++++++++
 relational_operators/TextScanOperator.hpp       |   3 +
 relational_operators/UpdateOperator.cpp         |  22 +++
 relational_operators/UpdateOperator.hpp         |   3 +
 42 files changed, 1238 insertions(+), 28 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/3e35844f/query_execution/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/query_execution/CMakeLists.txt b/query_execution/CMakeLists.txt
index 7d9d601..9b5c69f 100644
--- a/query_execution/CMakeLists.txt
+++ b/query_execution/CMakeLists.txt
@@ -36,6 +36,7 @@ add_library(quickstep_queryexecution_QueryExecutionState ../empty_src.cpp QueryE
 add_library(quickstep_queryexecution_QueryExecutionTypedefs ../empty_src.cpp QueryExecutionTypedefs.hpp)
 add_library(quickstep_queryexecution_QueryExecutionUtil ../empty_src.cpp QueryExecutionUtil.hpp)
 add_library(quickstep_queryexecution_QueryManager QueryManager.cpp QueryManager.hpp)
+add_library(quickstep_queryexecution_WorkOrderProtosContainer ../empty_src.cpp WorkOrderProtosContainer.hpp)
 add_library(quickstep_queryexecution_WorkOrdersContainer WorkOrdersContainer.cpp WorkOrdersContainer.hpp)
 add_library(quickstep_queryexecution_Worker Worker.cpp Worker.hpp)
 add_library(quickstep_queryexecution_WorkerDirectory ../empty_src.cpp WorkerDirectory.hpp)
@@ -151,6 +152,10 @@ target_link_libraries(quickstep_queryexecution_QueryManager
                       quickstep_utility_DAG
                       quickstep_utility_Macros
                       tmb)
+target_link_libraries(quickstep_queryexecution_WorkOrderProtosContainer
+                      glog
+                      quickstep_relationaloperators_WorkOrder_proto
+                      quickstep_utility_Macros)
 target_link_libraries(quickstep_queryexecution_WorkOrdersContainer
                       glog
                       quickstep_relationaloperators_WorkOrder
@@ -187,6 +192,7 @@ target_link_libraries(quickstep_queryexecution
                       quickstep_queryexecution_QueryExecutionTypedefs
                       quickstep_queryexecution_QueryExecutionUtil
                       quickstep_queryexecution_QueryManager
+                      quickstep_queryexecution_WorkOrderProtosContainer
                       quickstep_queryexecution_WorkOrdersContainer
                       quickstep_queryexecution_Worker
                       quickstep_queryexecution_WorkerDirectory

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/3e35844f/query_execution/WorkOrderProtosContainer.hpp
----------------------------------------------------------------------
diff --git a/query_execution/WorkOrderProtosContainer.hpp b/query_execution/WorkOrderProtosContainer.hpp
new file mode 100644
index 0000000..e0d6d22
--- /dev/null
+++ b/query_execution/WorkOrderProtosContainer.hpp
@@ -0,0 +1,147 @@
+/**
+ *   Copyright 2015-2016 Pivotal Software, Inc.
+ *
+ *   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_QUERY_EXECUTION_WORKORDER_PROTOS_CONTAINER_HPP_
+#define QUICKSTEP_QUERY_EXECUTION_WORKORDER_PROTOS_CONTAINER_HPP_
+
+#include <cstddef>
+#include <memory>
+#include <queue>
+#include <vector>
+
+#include "relational_operators/WorkOrder.pb.h"  // IWYU pragma: keep
+#include "utility/Macros.hpp"
+
+#include "glog/logging.h"
+
+namespace quickstep {
+
+/** \addtogroup QueryExecution
+ *  @{
+ */
+
+ /**
+  * @brief A container used in the distributed version to hold the normal
+  *        (non-rebuild) WorkOrder protos for a given query.
+  *
+  * @note This container stays alive during the lifetime of the query.
+  **/
+class WorkOrderProtosContainer {
+ public:
+  /**
+   * @brief Constructor
+   *
+   * @param num_operators Number of operators in the query DAG.
+   **/
+  explicit WorkOrderProtosContainer(const std::size_t num_operators)
+      : num_operators_(num_operators),
+        operator_containers_(num_operators_) {
+    DCHECK_NE(num_operators_, 0u);
+  }
+
+  /**
+   * @brief Destructor.
+   *
+   * @note If the query is executed normally, we should never encounter a
+   *       situation where at the time of deletion the WorkOrderProtosContainer has
+   *       pending WorkOrders.
+   **/
+  ~WorkOrderProtosContainer() {
+    for (std::size_t op = 0; op < num_operators_; ++op) {
+      if (hasWorkOrderProto(op)) {
+        LOG(WARNING) << "Destroying a WorkOrderProtosContainer that still has pending WorkOrder protos.";
+        break;
+      }
+    }
+  }
+
+  /**
+   * @brief Check if there are some pending WorkOrders for the given operator.
+   *
+   * @param operator_index Index of the operator.
+   *
+   * @return If there are pending WorkOrders.
+   **/
+  bool hasWorkOrderProto(const std::size_t operator_index) const {
+    DCHECK_LT(operator_index, num_operators_);
+    return !operator_containers_[operator_index].empty();
+  }
+
+  /**
+   * @brief Get a WorkOrder for a given operator.
+   *
+   * @param operator_index The index of the operator.
+   *
+   * @return Release a WorkOrder proto. If no WorkOrder proto is available,
+   *         return nullptr.
+   **/
+  serialization::WorkOrder* getWorkOrderProto(const std::size_t operator_index) {
+    DCHECK_LT(operator_index, num_operators_);
+
+    if (operator_containers_[operator_index].empty()) {
+      return nullptr;
+    }
+
+    serialization::WorkOrder *proto =
+        operator_containers_[operator_index].front().release();
+    operator_containers_[operator_index].pop();
+
+    return proto;
+  }
+
+  /**
+   * @brief Add a WorkOrder generated from a given
+   *        operator.
+   *
+   * @param workorder A pointer to the WorkOrder to be added.
+   * @param operator_index The index of the operator in the query DAG.
+   **/
+  void addWorkOrderProto(serialization::WorkOrder *proto,
+                         const std::size_t operator_index) {
+    DCHECK(proto != nullptr);
+    DCHECK_LT(operator_index, num_operators_);
+
+    operator_containers_[operator_index].emplace(
+        std::unique_ptr<serialization::WorkOrder>(proto));
+  }
+
+  /**
+   * @brief Get the number of all pending WorkOrders
+   *        for a given operator.
+   *
+   * @param operator_index The index of the operator.
+   *
+   * @return The number of pending WorkOrders.
+   **/
+  std::size_t getNumWorkOrderProtos(const std::size_t operator_index) const {
+    DCHECK_LT(operator_index, num_operators_);
+    return operator_containers_[operator_index].size();
+  }
+
+ private:
+  const std::size_t num_operators_;
+
+  std::vector<std::queue<std::unique_ptr<serialization::WorkOrder>>> operator_containers_;
+
+  DISALLOW_COPY_AND_ASSIGN(WorkOrderProtosContainer);
+};
+
+/** @} */
+
+}  // namespace quickstep
+
+
+#endif  // QUICKSTEP_QUERY_EXECUTION_WORKORDER_PROTOS_CONTAINER_HPP_

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/3e35844f/query_execution/tests/Foreman_unittest.cpp
----------------------------------------------------------------------
diff --git a/query_execution/tests/Foreman_unittest.cpp b/query_execution/tests/Foreman_unittest.cpp
index 47cc641..500c57d 100644
--- a/query_execution/tests/Foreman_unittest.cpp
+++ b/query_execution/tests/Foreman_unittest.cpp
@@ -58,6 +58,8 @@ using tmb::client_id;
 
 namespace quickstep {
 
+class WorkOrderProtosContainer;
+
 class MockWorkOrder : public WorkOrder {
  public:
   explicit MockWorkOrder(const int op_index)
@@ -167,6 +169,10 @@ class MockOperator: public RelationalOperator {
     return num_calls_get_workorders_ == max_getworkorder_iters_;
   }
 
+  bool getAllWorkOrderProtos(WorkOrderProtosContainer *container) override {
+    return true;
+  }
+
   void feedInputBlock(const block_id input_block_id,
                       const relation_id input_relation_id) override {
     ++num_calls_feedblock_;

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/3e35844f/query_execution/tests/QueryManager_unittest.cpp
----------------------------------------------------------------------
diff --git a/query_execution/tests/QueryManager_unittest.cpp b/query_execution/tests/QueryManager_unittest.cpp
index 1b9be48..4293569 100644
--- a/query_execution/tests/QueryManager_unittest.cpp
+++ b/query_execution/tests/QueryManager_unittest.cpp
@@ -59,6 +59,8 @@ using tmb::client_id;
 
 namespace quickstep {
 
+class WorkOrderProtosContainer;
+
 class MockWorkOrder : public WorkOrder {
  public:
   explicit MockWorkOrder(const int op_index)
@@ -168,6 +170,10 @@ class MockOperator: public RelationalOperator {
     return num_calls_get_workorders_ == max_getworkorder_iters_;
   }
 
+  bool getAllWorkOrderProtos(WorkOrderProtosContainer *container) override {
+    return true;
+  }
+
   void feedInputBlock(const block_id input_block_id,
                       const relation_id input_relation_id) override {
     ++num_calls_feedblock_;

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/3e35844f/relational_operators/AggregationOperator.cpp
----------------------------------------------------------------------
diff --git a/relational_operators/AggregationOperator.cpp b/relational_operators/AggregationOperator.cpp
index 1b935ee..28844f6 100644
--- a/relational_operators/AggregationOperator.cpp
+++ b/relational_operators/AggregationOperator.cpp
@@ -20,7 +20,9 @@
 #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/AggregationOperationState.hpp"
 #include "storage/StorageBlockInfo.hpp"
 
@@ -57,6 +59,37 @@ bool AggregationOperator::getAllWorkOrders(
   }
 }
 
+bool AggregationOperator::getAllWorkOrderProtos(WorkOrderProtosContainer *container) {
+  if (input_relation_is_stored_) {
+    if (!started_) {
+      for (const block_id input_block_id : input_relation_block_ids_) {
+        container->addWorkOrderProto(createWorkOrderProto(input_block_id), op_index_);
+      }
+      started_ = true;
+    }
+    return true;
+  } else {
+    while (num_workorders_generated_ < input_relation_block_ids_.size()) {
+      container->addWorkOrderProto(
+          createWorkOrderProto(input_relation_block_ids_[num_workorders_generated_]),
+          op_index_);
+      ++num_workorders_generated_;
+    }
+    return done_feeding_input_relation_;
+  }
+}
+
+serialization::WorkOrder* AggregationOperator::createWorkOrderProto(const block_id block) {
+  serialization::WorkOrder *proto = new serialization::WorkOrder;
+  proto->set_work_order_type(serialization::AGGREGATION);
+
+  proto->SetExtension(serialization::AggregationWorkOrder::block_id, block);
+  proto->SetExtension(serialization::AggregationWorkOrder::aggr_state_index, aggr_state_index_);
+
+  return proto;
+}
+
+
 void AggregationWorkOrder::execute() {
   state_->aggregateBlock(input_block_id_);
 }

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/3e35844f/relational_operators/AggregationOperator.hpp
----------------------------------------------------------------------
diff --git a/relational_operators/AggregationOperator.hpp b/relational_operators/AggregationOperator.hpp
index 0e74dfc..15da776 100644
--- a/relational_operators/AggregationOperator.hpp
+++ b/relational_operators/AggregationOperator.hpp
@@ -38,8 +38,11 @@ namespace quickstep {
 
 class AggregationOperationState;
 class StorageManager;
+class WorkOrderProtosContainer;
 class WorkOrdersContainer;
 
+namespace serialization { class WorkOrder; }
+
 /** \addtogroup RelationalOperators
  *  @{
  */
@@ -77,6 +80,8 @@ class AggregationOperator : public RelationalOperator {
                         const tmb::client_id scheduler_client_id,
                         tmb::MessageBus *bus) override;
 
+  bool getAllWorkOrderProtos(WorkOrderProtosContainer *container) override;
+
   void feedInputBlock(const block_id input_block_id, const relation_id input_relation_id) override {
     input_relation_block_ids_.push_back(input_block_id);
   }
@@ -88,6 +93,13 @@ class AggregationOperator : public RelationalOperator {
   }
 
  private:
+  /**
+   * @brief Create Work Order proto.
+   *
+   * @param block The block id used in the Work Order.
+   **/
+  serialization::WorkOrder* createWorkOrderProto(const block_id block);
+
   const bool input_relation_is_stored_;
   std::vector<block_id> input_relation_block_ids_;
   const QueryContext::aggregation_state_id aggr_state_index_;

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/3e35844f/relational_operators/BuildHashOperator.cpp
----------------------------------------------------------------------
diff --git a/relational_operators/BuildHashOperator.cpp b/relational_operators/BuildHashOperator.cpp
index df92159..f320e1a 100644
--- a/relational_operators/BuildHashOperator.cpp
+++ b/relational_operators/BuildHashOperator.cpp
@@ -22,7 +22,9 @@
 
 #include "catalog/CatalogRelation.hpp"
 #include "query_execution/QueryContext.hpp"
+#include "query_execution/WorkOrderProtosContainer.hpp"
 #include "query_execution/WorkOrdersContainer.hpp"
+#include "relational_operators/WorkOrder.pb.h"
 #include "storage/HashTable.hpp"
 #include "storage/StorageBlock.hpp"
 #include "storage/StorageBlockInfo.hpp"
@@ -97,6 +99,43 @@ bool BuildHashOperator::getAllWorkOrders(
   }
 }
 
+bool BuildHashOperator::getAllWorkOrderProtos(WorkOrderProtosContainer *container) {
+  if (input_relation_is_stored_) {
+    if (!started_) {
+      for (const block_id input_block_id : input_relation_block_ids_) {
+        container->addWorkOrderProto(createWorkOrderProto(input_block_id), op_index_);
+      }
+      started_ = true;
+    }
+    return true;
+  } else {
+    while (num_workorders_generated_ < input_relation_block_ids_.size()) {
+      container->addWorkOrderProto(
+          createWorkOrderProto(input_relation_block_ids_[num_workorders_generated_]),
+          op_index_);
+      ++num_workorders_generated_;
+    }
+    return done_feeding_input_relation_;
+  }
+}
+
+serialization::WorkOrder* BuildHashOperator::createWorkOrderProto(const block_id block) {
+  serialization::WorkOrder *proto = new serialization::WorkOrder;
+  proto->set_work_order_type(serialization::BUILD_HASH);
+
+  proto->SetExtension(serialization::BuildHashWorkOrder::relation_id, input_relation_.getID());
+  for (const attribute_id attr_id : join_key_attributes_) {
+    proto->AddExtension(serialization::BuildHashWorkOrder::join_key_attributes, attr_id);
+  }
+  proto->SetExtension(serialization::BuildHashWorkOrder::any_join_key_attributes_nullable,
+                      any_join_key_attributes_nullable_);
+  proto->SetExtension(serialization::BuildHashWorkOrder::join_hash_table_index, hash_table_index_);
+  proto->SetExtension(serialization::BuildHashWorkOrder::block_id, block);
+
+  return proto;
+}
+
+
 void BuildHashWorkOrder::execute() {
   BlockReference block(
       storage_manager_->getBlock(build_block_id_, input_relation_));

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/3e35844f/relational_operators/BuildHashOperator.hpp
----------------------------------------------------------------------
diff --git a/relational_operators/BuildHashOperator.hpp b/relational_operators/BuildHashOperator.hpp
index f9d830f..c7fd9dd 100644
--- a/relational_operators/BuildHashOperator.hpp
+++ b/relational_operators/BuildHashOperator.hpp
@@ -39,6 +39,7 @@ namespace quickstep {
 
 class CatalogRelationSchema;
 class StorageManager;
+class WorkOrderProtosContainer;
 class WorkOrdersContainer;
 
 struct TupleReference;
@@ -46,6 +47,8 @@ struct TupleReference;
 template <typename, bool, bool, bool, bool> class HashTable;
 typedef HashTable<TupleReference, true, false, false, true> JoinHashTable;
 
+namespace serialization { class WorkOrder; }
+
 /** \addtogroup RelationalOperators
  *  @{
  */
@@ -93,6 +96,8 @@ class BuildHashOperator : public RelationalOperator {
                         const tmb::client_id scheduler_client_id,
                         tmb::MessageBus *bus) override;
 
+  bool getAllWorkOrderProtos(WorkOrderProtosContainer *container) override;
+
   void feedInputBlock(const block_id input_block_id,
                       const relation_id input_relation_id) override {
     input_relation_block_ids_.push_back(input_block_id);
@@ -106,6 +111,13 @@ class BuildHashOperator : public RelationalOperator {
   }
 
  private:
+  /**
+   * @brief Create Work Order proto.
+   *
+   * @param block The block id used in the Work Order.
+   **/
+  serialization::WorkOrder* createWorkOrderProto(const block_id block);
+
   const CatalogRelation &input_relation_;
   const bool input_relation_is_stored_;
   const std::vector<attribute_id> join_key_attributes_;

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/3e35844f/relational_operators/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/relational_operators/CMakeLists.txt b/relational_operators/CMakeLists.txt
index eec5300..7c623ac 100644
--- a/relational_operators/CMakeLists.txt
+++ b/relational_operators/CMakeLists.txt
@@ -71,9 +71,11 @@ target_link_libraries(quickstep_relationaloperators_AggregationOperator
                       quickstep_catalog_CatalogRelation
                       quickstep_catalog_CatalogTypedefs
                       quickstep_queryexecution_QueryContext
+                      quickstep_queryexecution_WorkOrderProtosContainer
                       quickstep_queryexecution_WorkOrdersContainer
                       quickstep_relationaloperators_RelationalOperator
                       quickstep_relationaloperators_WorkOrder
+                      quickstep_relationaloperators_WorkOrder_proto
                       quickstep_storage_AggregationOperationState
                       quickstep_storage_StorageBlockInfo
                       quickstep_utility_Macros
@@ -83,9 +85,11 @@ target_link_libraries(quickstep_relationaloperators_BuildHashOperator
                       quickstep_catalog_CatalogRelation
                       quickstep_catalog_CatalogTypedefs
                       quickstep_queryexecution_QueryContext
+                      quickstep_queryexecution_WorkOrderProtosContainer
                       quickstep_queryexecution_WorkOrdersContainer
                       quickstep_relationaloperators_RelationalOperator
                       quickstep_relationaloperators_WorkOrder
+                      quickstep_relationaloperators_WorkOrder_proto
                       quickstep_storage_HashTable
                       quickstep_storage_StorageBlock
                       quickstep_storage_StorageBlockInfo
@@ -118,9 +122,11 @@ target_link_libraries(quickstep_relationaloperators_DeleteOperator
                       quickstep_queryexecution_QueryExecutionMessages_proto
                       quickstep_queryexecution_QueryExecutionTypedefs
                       quickstep_queryexecution_QueryExecutionUtil
+                      quickstep_queryexecution_WorkOrderProtosContainer
                       quickstep_queryexecution_WorkOrdersContainer
                       quickstep_relationaloperators_RelationalOperator
                       quickstep_relationaloperators_WorkOrder
+                      quickstep_relationaloperators_WorkOrder_proto
                       quickstep_storage_StorageBlock
                       quickstep_storage_StorageBlockInfo
                       quickstep_storage_StorageManager
@@ -130,9 +136,11 @@ target_link_libraries(quickstep_relationaloperators_DeleteOperator
 target_link_libraries(quickstep_relationaloperators_DestroyHashOperator
                       glog
                       quickstep_queryexecution_QueryContext
+                      quickstep_queryexecution_WorkOrderProtosContainer
                       quickstep_queryexecution_WorkOrdersContainer
                       quickstep_relationaloperators_RelationalOperator
                       quickstep_relationaloperators_WorkOrder
+                      quickstep_relationaloperators_WorkOrder_proto
                       quickstep_utility_Macros
                       tmb)
 target_link_libraries(quickstep_relationaloperators_DropTableOperator
@@ -141,9 +149,11 @@ target_link_libraries(quickstep_relationaloperators_DropTableOperator
                       quickstep_catalog_CatalogDatabaseLite
                       quickstep_catalog_CatalogRelation
                       quickstep_catalog_CatalogTypedefs
+                      quickstep_queryexecution_WorkOrderProtosContainer
                       quickstep_queryexecution_WorkOrdersContainer
                       quickstep_relationaloperators_RelationalOperator
                       quickstep_relationaloperators_WorkOrder
+                      quickstep_relationaloperators_WorkOrder_proto
                       quickstep_storage_StorageBlockInfo
                       quickstep_storage_StorageManager
                       quickstep_utility_Macros
@@ -153,9 +163,11 @@ target_link_libraries(quickstep_relationaloperators_FinalizeAggregationOperator
                       quickstep_catalog_CatalogRelation
                       quickstep_catalog_CatalogTypedefs
                       quickstep_queryexecution_QueryContext
+                      quickstep_queryexecution_WorkOrderProtosContainer
                       quickstep_queryexecution_WorkOrdersContainer
                       quickstep_relationaloperators_RelationalOperator
                       quickstep_relationaloperators_WorkOrder
+                      quickstep_relationaloperators_WorkOrder_proto
                       quickstep_storage_AggregationOperationState
                       quickstep_utility_Macros
                       tmb)
@@ -168,9 +180,11 @@ target_link_libraries(quickstep_relationaloperators_HashJoinOperator
                       quickstep_expressions_predicate_Predicate
                       quickstep_expressions_scalar_Scalar
                       quickstep_queryexecution_QueryContext
+                      quickstep_queryexecution_WorkOrderProtosContainer
                       quickstep_queryexecution_WorkOrdersContainer
                       quickstep_relationaloperators_RelationalOperator
                       quickstep_relationaloperators_WorkOrder
+                      quickstep_relationaloperators_WorkOrder_proto
                       quickstep_storage_HashTable
                       quickstep_storage_InsertDestination
                       quickstep_storage_StorageBlock
@@ -192,9 +206,11 @@ target_link_libraries(quickstep_relationaloperators_InsertOperator
                       quickstep_catalog_CatalogRelation
                       quickstep_catalog_CatalogTypedefs
                       quickstep_queryexecution_QueryContext
+                      quickstep_queryexecution_WorkOrderProtosContainer
                       quickstep_queryexecution_WorkOrdersContainer
                       quickstep_relationaloperators_RelationalOperator
                       quickstep_relationaloperators_WorkOrder
+                      quickstep_relationaloperators_WorkOrder_proto
                       quickstep_storage_InsertDestination
                       quickstep_types_containers_Tuple
                       quickstep_utility_Macros
@@ -207,9 +223,11 @@ target_link_libraries(quickstep_relationaloperators_NestedLoopsJoinOperator
                       quickstep_expressions_predicate_Predicate
                       quickstep_expressions_scalar_Scalar
                       quickstep_queryexecution_QueryContext
+                      quickstep_queryexecution_WorkOrderProtosContainer
                       quickstep_queryexecution_WorkOrdersContainer
                       quickstep_relationaloperators_RelationalOperator
                       quickstep_relationaloperators_WorkOrder
+                      quickstep_relationaloperators_WorkOrder_proto
                       quickstep_storage_InsertDestination
                       quickstep_storage_StorageBlock
                       quickstep_storage_StorageBlockInfo
@@ -243,9 +261,11 @@ target_link_libraries(quickstep_relationaloperators_SampleOperator
                       quickstep_catalog_CatalogRelation
                       quickstep_catalog_CatalogTypedefs
                       quickstep_queryexecution_QueryContext
-                      quickstep_relationaloperators_RelationalOperator
+                      quickstep_queryexecution_WorkOrderProtosContainer
                       quickstep_queryexecution_WorkOrdersContainer
+                      quickstep_relationaloperators_RelationalOperator
                       quickstep_relationaloperators_WorkOrder
+                      quickstep_relationaloperators_WorkOrder_proto
                       quickstep_storage_InsertDestination
                       quickstep_storage_StorageBlock
                       quickstep_storage_StorageBlockInfo
@@ -255,9 +275,11 @@ target_link_libraries(quickstep_relationaloperators_SampleOperator
 target_link_libraries(quickstep_relationaloperators_SaveBlocksOperator
                       glog
                       quickstep_catalog_CatalogTypedefs
+                      quickstep_queryexecution_WorkOrderProtosContainer
                       quickstep_queryexecution_WorkOrdersContainer
                       quickstep_relationaloperators_RelationalOperator
                       quickstep_relationaloperators_WorkOrder
+                      quickstep_relationaloperators_WorkOrder_proto
                       quickstep_storage_StorageBlockInfo
                       quickstep_storage_StorageManager
                       quickstep_utility_Macros
@@ -268,9 +290,11 @@ target_link_libraries(quickstep_relationaloperators_SelectOperator
                       quickstep_catalog_CatalogTypedefs
                       quickstep_catalog_PartitionSchemeHeader
                       quickstep_queryexecution_QueryContext
+                      quickstep_queryexecution_WorkOrderProtosContainer
                       quickstep_queryexecution_WorkOrdersContainer
                       quickstep_relationaloperators_RelationalOperator
                       quickstep_relationaloperators_WorkOrder
+                      quickstep_relationaloperators_WorkOrder_proto
                       quickstep_storage_InsertDestination
                       quickstep_storage_StorageBlock
                       quickstep_storage_StorageBlockInfo
@@ -287,11 +311,13 @@ target_link_libraries(quickstep_relationaloperators_SortMergeRunOperator
                       quickstep_catalog_CatalogTypedefs
                       quickstep_queryexecution_QueryContext
                       quickstep_queryexecution_QueryExecutionTypedefs
+                      quickstep_queryexecution_WorkOrderProtosContainer
                       quickstep_queryexecution_WorkOrdersContainer
                       quickstep_relationaloperators_RelationalOperator
                       quickstep_relationaloperators_SortMergeRunOperatorHelpers
                       quickstep_relationaloperators_SortMergeRunOperator_proto
                       quickstep_relationaloperators_WorkOrder
+                      quickstep_relationaloperators_WorkOrder_proto
                       quickstep_storage_StorageBlockInfo
                       quickstep_threading_ThreadIDBasedMap
                       quickstep_utility_Macros
@@ -324,9 +350,11 @@ target_link_libraries(quickstep_relationaloperators_SortRunGenerationOperator
                       quickstep_catalog_CatalogRelation
                       quickstep_catalog_CatalogTypedefs
                       quickstep_queryexecution_QueryContext
+                      quickstep_queryexecution_WorkOrderProtosContainer
                       quickstep_queryexecution_WorkOrdersContainer
                       quickstep_relationaloperators_RelationalOperator
                       quickstep_relationaloperators_WorkOrder
+                      quickstep_relationaloperators_WorkOrder_proto
                       quickstep_storage_InsertDestination
                       quickstep_storage_StorageBlock
                       quickstep_storage_StorageBlockInfo
@@ -341,9 +369,11 @@ target_link_libraries(quickstep_relationaloperators_TableGeneratorOperator
                       quickstep_catalog_CatalogTypedefs
                       quickstep_expressions_tablegenerator_GeneratorFunctionHandle
                       quickstep_queryexecution_QueryContext
+                      quickstep_queryexecution_WorkOrderProtosContainer
                       quickstep_queryexecution_WorkOrdersContainer
                       quickstep_relationaloperators_RelationalOperator
                       quickstep_relationaloperators_WorkOrder
+                      quickstep_relationaloperators_WorkOrder_proto
                       quickstep_storage_InsertDestination
                       quickstep_storage_StorageBlockInfo
                       quickstep_types_containers_ColumnVectorsValueAccessor
@@ -360,10 +390,12 @@ target_link_libraries(quickstep_relationaloperators_TextScanOperator
                       quickstep_queryexecution_QueryExecutionMessages_proto
                       quickstep_queryexecution_QueryExecutionTypedefs
                       quickstep_queryexecution_QueryExecutionUtil
+                      quickstep_queryexecution_WorkOrderProtosContainer
                       quickstep_queryexecution_WorkOrdersContainer
                       quickstep_relationaloperators_RelationalOperator
                       quickstep_relationaloperators_TextScanOperator_proto
                       quickstep_relationaloperators_WorkOrder
+                      quickstep_relationaloperators_WorkOrder_proto
                       quickstep_storage_InsertDestination
                       quickstep_storage_StorageBlob
                       quickstep_storage_StorageBlockInfo
@@ -385,9 +417,11 @@ target_link_libraries(quickstep_relationaloperators_UpdateOperator
                       quickstep_queryexecution_QueryExecutionMessages_proto
                       quickstep_queryexecution_QueryExecutionTypedefs
                       quickstep_queryexecution_QueryExecutionUtil
+                      quickstep_queryexecution_WorkOrderProtosContainer
                       quickstep_queryexecution_WorkOrdersContainer
                       quickstep_relationaloperators_RelationalOperator
                       quickstep_relationaloperators_WorkOrder
+                      quickstep_relationaloperators_WorkOrder_proto
                       quickstep_storage_InsertDestination
                       quickstep_storage_StorageBlock
                       quickstep_storage_StorageBlockInfo

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/3e35844f/relational_operators/CreateIndexOperator.hpp
----------------------------------------------------------------------
diff --git a/relational_operators/CreateIndexOperator.hpp b/relational_operators/CreateIndexOperator.hpp
index 2bfacc4..03f7fed 100644
--- a/relational_operators/CreateIndexOperator.hpp
+++ b/relational_operators/CreateIndexOperator.hpp
@@ -1,6 +1,7 @@
 /**
  *   Copyright 2016, Quickstep Research Group, Computer Sciences Department,
  *     University of Wisconsin—Madison.
+ *   Copyright 2016 Pivotal Software, Inc.
  *
  *   Licensed under the Apache License, Version 2.0 (the "License");
  *   you may not use this file except in compliance with the License.
@@ -36,6 +37,7 @@ namespace quickstep {
 class CatalogRelation;
 class QueryContext;
 class StorageManager;
+class WorkOrderProtosContainer;
 class WorkOrdersContainer;
 
 /** \addtogroup RelationalOperators
@@ -72,6 +74,13 @@ class CreateIndexOperator : public RelationalOperator {
                         const tmb::client_id scheduler_client_id,
                         tmb::MessageBus *bus) override;
 
+  /**
+   * @note no WorkOrder proto generated for this operator.
+   **/
+  bool getAllWorkOrderProtos(WorkOrderProtosContainer *container) override {
+    return true;
+  }
+
   void updateCatalogOnCompletion() override;
 
  private:

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/3e35844f/relational_operators/CreateTableOperator.hpp
----------------------------------------------------------------------
diff --git a/relational_operators/CreateTableOperator.hpp b/relational_operators/CreateTableOperator.hpp
index 98f3253..eac6828 100644
--- a/relational_operators/CreateTableOperator.hpp
+++ b/relational_operators/CreateTableOperator.hpp
@@ -35,6 +35,7 @@ namespace quickstep {
 class CatalogDatabase;
 class QueryContext;
 class StorageManager;
+class WorkOrderProtosContainer;
 class WorkOrdersContainer;
 
 /** \addtogroup RelationalOperators
@@ -70,6 +71,13 @@ class CreateTableOperator : public RelationalOperator {
                         const tmb::client_id scheduler_client_id,
                         tmb::MessageBus *bus) override;
 
+  /**
+   * @note no WorkOrder proto generated for this operator.
+   **/
+  bool getAllWorkOrderProtos(WorkOrderProtosContainer *container) override {
+    return true;
+  }
+
   void updateCatalogOnCompletion() override;
 
  private:

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/3e35844f/relational_operators/DeleteOperator.cpp
----------------------------------------------------------------------
diff --git a/relational_operators/DeleteOperator.cpp b/relational_operators/DeleteOperator.cpp
index 2c2c6de..4e02caf 100644
--- a/relational_operators/DeleteOperator.cpp
+++ b/relational_operators/DeleteOperator.cpp
@@ -26,13 +26,14 @@
 #include "query_execution/QueryContext.hpp"
 #include "query_execution/QueryExecutionMessages.pb.h"
 #include "query_execution/QueryExecutionUtil.hpp"
+#include "query_execution/WorkOrderProtosContainer.hpp"
 #include "query_execution/WorkOrdersContainer.hpp"
+#include "relational_operators/WorkOrder.pb.h"
 #include "storage/StorageBlock.hpp"
 #include "storage/StorageBlockInfo.hpp"
 #include "storage/StorageManager.hpp"
 #include "threading/ThreadIDBasedMap.hpp"
 
-
 #include "glog/logging.h"
 
 #include "tmb/id_typedefs.h"
@@ -83,6 +84,40 @@ bool DeleteOperator::getAllWorkOrders(
   }
 }
 
+bool DeleteOperator::getAllWorkOrderProtos(WorkOrderProtosContainer *container) {
+  if (relation_is_stored_) {
+    // If relation_ is stored, iterate over the list of blocks in relation_.
+    if (!started_) {
+      for (const block_id input_block_id : relation_block_ids_) {
+        container->addWorkOrderProto(createWorkOrderProto(input_block_id), op_index_);
+      }
+      started_ = true;
+    }
+    return true;
+  } else {
+    while (num_workorders_generated_ < relation_block_ids_.size()) {
+      container->addWorkOrderProto(
+          createWorkOrderProto(relation_block_ids_[num_workorders_generated_]),
+          op_index_);
+      ++num_workorders_generated_;
+    }
+    return done_feeding_input_relation_;
+  }
+}
+
+serialization::WorkOrder* DeleteOperator::createWorkOrderProto(const block_id block) {
+  serialization::WorkOrder *proto = new serialization::WorkOrder;
+  proto->set_work_order_type(serialization::DELETE);
+
+  proto->SetExtension(serialization::DeleteWorkOrder::operator_index, op_index_);
+  proto->SetExtension(serialization::DeleteWorkOrder::relation_id, relation_.getID());
+  proto->SetExtension(serialization::DeleteWorkOrder::predicate_index, predicate_index_);
+  proto->SetExtension(serialization::DeleteWorkOrder::block_id, block);
+
+  return proto;
+}
+
+
 void DeleteWorkOrder::execute() {
   MutableBlockReference block(
       storage_manager_->getBlockMutable(input_block_id_, input_relation_));

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/3e35844f/relational_operators/DeleteOperator.hpp
----------------------------------------------------------------------
diff --git a/relational_operators/DeleteOperator.hpp b/relational_operators/DeleteOperator.hpp
index 1d44552..6dc206b 100644
--- a/relational_operators/DeleteOperator.hpp
+++ b/relational_operators/DeleteOperator.hpp
@@ -41,8 +41,11 @@ namespace quickstep {
 class CatalogRelationSchema;
 class Predicate;
 class StorageManager;
+class WorkOrderProtosContainer;
 class WorkOrdersContainer;
 
+namespace serialization { class WorkOrder; }
+
 /** \addtogroup RelationalOperators
  *  @{
  */
@@ -65,13 +68,13 @@ class DeleteOperator : public RelationalOperator {
   DeleteOperator(const CatalogRelation &relation,
                  const QueryContext::predicate_id predicate_index,
                  const bool relation_is_stored)
-     :  relation_(relation),
-        predicate_index_(predicate_index),
-        relation_is_stored_(relation_is_stored),
-        started_(false),
-        relation_block_ids_(relation_is_stored ? relation.getBlocksSnapshot()
-                                               : std::vector<block_id>()),
-        num_workorders_generated_(0) {}
+     : relation_(relation),
+       predicate_index_(predicate_index),
+       relation_is_stored_(relation_is_stored),
+       started_(false),
+       relation_block_ids_(relation_is_stored ? relation.getBlocksSnapshot()
+                                              : std::vector<block_id>()),
+       num_workorders_generated_(0) {}
 
   ~DeleteOperator() override {}
 
@@ -81,6 +84,8 @@ class DeleteOperator : public RelationalOperator {
                         const tmb::client_id scheduler_client_id,
                         tmb::MessageBus *bus) override;
 
+  bool getAllWorkOrderProtos(WorkOrderProtosContainer *container) override;
+
   const relation_id getOutputRelationID() const override {
     return relation_.getID();
   }
@@ -98,6 +103,13 @@ class DeleteOperator : public RelationalOperator {
   }
 
  private:
+  /**
+   * @brief Create Work Order proto.
+   *
+   * @param block The block id used in the Work Order.
+   **/
+  serialization::WorkOrder* createWorkOrderProto(const block_id block);
+
   const CatalogRelation &relation_;
   const QueryContext::predicate_id predicate_index_;
 

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/3e35844f/relational_operators/DestroyHashOperator.cpp
----------------------------------------------------------------------
diff --git a/relational_operators/DestroyHashOperator.cpp b/relational_operators/DestroyHashOperator.cpp
index c2220d5..4567cf5 100644
--- a/relational_operators/DestroyHashOperator.cpp
+++ b/relational_operators/DestroyHashOperator.cpp
@@ -18,7 +18,9 @@
 #include "relational_operators/DestroyHashOperator.hpp"
 
 #include "query_execution/QueryContext.hpp"
+#include "query_execution/WorkOrderProtosContainer.hpp"
 #include "query_execution/WorkOrdersContainer.hpp"
+#include "relational_operators/WorkOrder.pb.h"
 
 #include "tmb/id_typedefs.h"
 
@@ -38,6 +40,20 @@ bool DestroyHashOperator::getAllWorkOrders(
   return work_generated_;
 }
 
+bool DestroyHashOperator::getAllWorkOrderProtos(WorkOrderProtosContainer *container) {
+  if (blocking_dependencies_met_ && !work_generated_) {
+    work_generated_ = true;
+
+    serialization::WorkOrder *proto = new serialization::WorkOrder;
+    proto->set_work_order_type(serialization::DESTROY_HASH);
+    proto->SetExtension(serialization::DestroyHashWorkOrder::join_hash_table_index, hash_table_index_);
+
+    container->addWorkOrderProto(proto, op_index_);
+  }
+  return work_generated_;
+}
+
+
 void DestroyHashWorkOrder::execute() {
   query_context_->destroyJoinHashTable(hash_table_index_);
 }

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/3e35844f/relational_operators/DestroyHashOperator.hpp
----------------------------------------------------------------------
diff --git a/relational_operators/DestroyHashOperator.hpp b/relational_operators/DestroyHashOperator.hpp
index 46331ba..67cbd4d 100644
--- a/relational_operators/DestroyHashOperator.hpp
+++ b/relational_operators/DestroyHashOperator.hpp
@@ -32,6 +32,7 @@ namespace tmb { class MessageBus; }
 namespace quickstep {
 
 class StorageManager;
+class WorkOrderProtosContainer;
 class WorkOrdersContainer;
 
 /** \addtogroup RelationalOperators
@@ -60,6 +61,8 @@ class DestroyHashOperator : public RelationalOperator {
                         const tmb::client_id scheduler_client_id,
                         tmb::MessageBus *bus) override;
 
+  bool getAllWorkOrderProtos(WorkOrderProtosContainer *container) override;
+
  private:
   const QueryContext::join_hash_table_id hash_table_index_;
   bool work_generated_;

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/3e35844f/relational_operators/DropTableOperator.cpp
----------------------------------------------------------------------
diff --git a/relational_operators/DropTableOperator.cpp b/relational_operators/DropTableOperator.cpp
index f3a3a2c..75ab221 100644
--- a/relational_operators/DropTableOperator.cpp
+++ b/relational_operators/DropTableOperator.cpp
@@ -24,7 +24,9 @@
 #include "catalog/CatalogDatabaseLite.hpp"
 #include "catalog/CatalogRelation.hpp"
 #include "catalog/CatalogTypedefs.hpp"
+#include "query_execution/WorkOrderProtosContainer.hpp"
 #include "query_execution/WorkOrdersContainer.hpp"
+#include "relational_operators/WorkOrder.pb.h"
 #include "storage/StorageBlockInfo.hpp"
 #include "storage/StorageManager.hpp"
 
@@ -54,6 +56,26 @@ bool DropTableOperator::getAllWorkOrders(
   return work_generated_;
 }
 
+bool DropTableOperator::getAllWorkOrderProtos(WorkOrderProtosContainer *container) {
+  if (blocking_dependencies_met_ && !work_generated_) {
+    work_generated_ = true;
+
+    serialization::WorkOrder *proto = new serialization::WorkOrder;
+    proto->set_work_order_type(serialization::DROP_TABLE);
+
+    std::vector<block_id> relation_blocks(relation_.getBlocksSnapshot());
+    for (const block_id relation_block : relation_blocks) {
+      proto->AddExtension(serialization::DropTableWorkOrder::block_ids, relation_block);
+    }
+
+    container->addWorkOrderProto(proto, op_index_);
+
+    database_->setStatus(CatalogDatabase::Status::kPendingBlockDeletions);
+  }
+
+  return work_generated_;
+}
+
 void DropTableOperator::updateCatalogOnCompletion() {
   const relation_id rel_id = relation_.getID();
   if (only_drop_blocks_) {

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/3e35844f/relational_operators/DropTableOperator.hpp
----------------------------------------------------------------------
diff --git a/relational_operators/DropTableOperator.hpp b/relational_operators/DropTableOperator.hpp
index bf9b1b1..ac3d96a 100644
--- a/relational_operators/DropTableOperator.hpp
+++ b/relational_operators/DropTableOperator.hpp
@@ -40,6 +40,7 @@ class CatalogDatabaseLite;
 class CatalogRelation;
 class QueryContext;
 class StorageManager;
+class WorkOrderProtosContainer;
 class WorkOrdersContainer;
 
 /** \addtogroup RelationalOperators
@@ -75,6 +76,8 @@ class DropTableOperator : public RelationalOperator {
                         const tmb::client_id scheduler_client_id,
                         tmb::MessageBus *bus) override;
 
+  bool getAllWorkOrderProtos(WorkOrderProtosContainer *container) override;
+
   void updateCatalogOnCompletion() override;
 
  private:

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/3e35844f/relational_operators/FinalizeAggregationOperator.cpp
----------------------------------------------------------------------
diff --git a/relational_operators/FinalizeAggregationOperator.cpp b/relational_operators/FinalizeAggregationOperator.cpp
index 410ec69..7db36c0 100644
--- a/relational_operators/FinalizeAggregationOperator.cpp
+++ b/relational_operators/FinalizeAggregationOperator.cpp
@@ -18,7 +18,9 @@
 #include "relational_operators/FinalizeAggregationOperator.hpp"
 
 #include "query_execution/QueryContext.hpp"
+#include "query_execution/WorkOrderProtosContainer.hpp"
 #include "query_execution/WorkOrdersContainer.hpp"
+#include "relational_operators/WorkOrder.pb.h"
 #include "storage/AggregationOperationState.hpp"
 
 #include "glog/logging.h"
@@ -45,6 +47,23 @@ bool FinalizeAggregationOperator::getAllWorkOrders(
   return started_;
 }
 
+bool FinalizeAggregationOperator::getAllWorkOrderProtos(WorkOrderProtosContainer *container) {
+  if (blocking_dependencies_met_ && !started_) {
+    started_ = true;
+
+    serialization::WorkOrder *proto = new serialization::WorkOrder;
+    proto->set_work_order_type(serialization::FINALIZE_AGGREGATION);
+    proto->SetExtension(serialization::FinalizeAggregationWorkOrder::aggr_state_index,
+                        aggr_state_index_);
+    proto->SetExtension(serialization::FinalizeAggregationWorkOrder::insert_destination_index,
+                        output_destination_index_);
+
+    container->addWorkOrderProto(proto, op_index_);
+  }
+  return started_;
+}
+
+
 void FinalizeAggregationWorkOrder::execute() {
   state_->finalizeAggregate(output_destination_);
 }

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/3e35844f/relational_operators/FinalizeAggregationOperator.hpp
----------------------------------------------------------------------
diff --git a/relational_operators/FinalizeAggregationOperator.hpp b/relational_operators/FinalizeAggregationOperator.hpp
index fb9608a..2254f6f 100644
--- a/relational_operators/FinalizeAggregationOperator.hpp
+++ b/relational_operators/FinalizeAggregationOperator.hpp
@@ -38,6 +38,7 @@ namespace quickstep {
 
 class InsertDestination;
 class StorageManager;
+class WorkOrderProtosContainer;
 class WorkOrdersContainer;
 
 /** \addtogroup RelationalOperators
@@ -74,6 +75,8 @@ class FinalizeAggregationOperator : public RelationalOperator {
                         const tmb::client_id scheduler_client_id,
                         tmb::MessageBus *bus) override;
 
+  bool getAllWorkOrderProtos(WorkOrderProtosContainer *container) override;
+
   QueryContext::insert_destination_id getInsertDestinationID() const override {
     return output_destination_index_;
   }

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/3e35844f/relational_operators/HashJoinOperator.cpp
----------------------------------------------------------------------
diff --git a/relational_operators/HashJoinOperator.cpp b/relational_operators/HashJoinOperator.cpp
index aa03794..87ce7da 100644
--- a/relational_operators/HashJoinOperator.cpp
+++ b/relational_operators/HashJoinOperator.cpp
@@ -31,7 +31,9 @@
 #include "expressions/predicate/Predicate.hpp"
 #include "expressions/scalar/Scalar.hpp"
 #include "query_execution/QueryContext.hpp"
+#include "query_execution/WorkOrderProtosContainer.hpp"
 #include "query_execution/WorkOrdersContainer.hpp"
+#include "relational_operators/WorkOrder.pb.h"
 #include "storage/HashTable.hpp"
 #include "storage/InsertDestination.hpp"
 #include "storage/StorageBlock.hpp"
@@ -386,6 +388,127 @@ bool HashJoinOperator::getAllOuterJoinWorkOrders(
   return false;
 }
 
+bool HashJoinOperator::getAllWorkOrderProtos(WorkOrderProtosContainer *container) {
+  switch (join_type_) {
+    case JoinType::kInnerJoin:
+      return getAllNonOuterJoinWorkOrderProtos(container, serialization::HashJoinWorkOrder::HASH_INNER_JOIN);
+    case JoinType::kLeftSemiJoin:
+      return getAllNonOuterJoinWorkOrderProtos(container, serialization::HashJoinWorkOrder::HASH_SEMI_JOIN);
+    case JoinType::kLeftAntiJoin:
+      return getAllNonOuterJoinWorkOrderProtos(container, serialization::HashJoinWorkOrder::HASH_ANTI_JOIN);
+    case JoinType::kLeftOuterJoin:
+      return getAllOuterJoinWorkOrderProtos(container);
+    default:
+      LOG(FATAL) << "Unknown join type in HashJoinOperator::getAllWorkOrderProtos()";
+  }
+}
+
+bool HashJoinOperator::getAllNonOuterJoinWorkOrderProtos(
+    WorkOrderProtosContainer *container,
+    const serialization::HashJoinWorkOrder::HashJoinWorkOrderType hash_join_type) {
+  // We wait until the building of global hash table is complete.
+  if (!blocking_dependencies_met_) {
+    return false;
+  }
+
+  if (probe_relation_is_stored_) {
+    if (!started_) {
+      for (const block_id probe_block_id : probe_relation_block_ids_) {
+        container->addWorkOrderProto(
+            createNonOuterJoinWorkOrderProto(hash_join_type, probe_block_id),
+            op_index_);
+      }
+      started_ = true;
+    }
+    return true;
+  } else {
+    while (num_workorders_generated_ < probe_relation_block_ids_.size()) {
+      container->addWorkOrderProto(
+          createNonOuterJoinWorkOrderProto(hash_join_type,
+                                           probe_relation_block_ids_[num_workorders_generated_]),
+          op_index_);
+      ++num_workorders_generated_;
+    }
+
+    return done_feeding_input_relation_;
+  }
+}
+
+serialization::WorkOrder* HashJoinOperator::createNonOuterJoinWorkOrderProto(
+    const serialization::HashJoinWorkOrder::HashJoinWorkOrderType hash_join_type,
+    const block_id block) {
+  serialization::WorkOrder *proto = new serialization::WorkOrder;
+  proto->set_work_order_type(serialization::HASH_JOIN);
+
+  proto->SetExtension(serialization::HashJoinWorkOrder::hash_join_work_order_type, hash_join_type);
+  proto->SetExtension(serialization::HashJoinWorkOrder::build_relation_id, build_relation_.getID());
+  proto->SetExtension(serialization::HashJoinWorkOrder::probe_relation_id, probe_relation_.getID());
+  for (const attribute_id attr_id : join_key_attributes_) {
+    proto->AddExtension(serialization::HashJoinWorkOrder::join_key_attributes, attr_id);
+  }
+  proto->SetExtension(serialization::HashJoinWorkOrder::any_join_key_attributes_nullable,
+                      any_join_key_attributes_nullable_);
+  proto->SetExtension(serialization::HashJoinWorkOrder::insert_destination_index, output_destination_index_);
+  proto->SetExtension(serialization::HashJoinWorkOrder::join_hash_table_index, hash_table_index_);
+  proto->SetExtension(serialization::HashJoinWorkOrder::selection_index, selection_index_);
+  proto->SetExtension(serialization::HashJoinWorkOrder::block_id, block);
+  proto->SetExtension(serialization::HashJoinWorkOrder::residual_predicate_index, residual_predicate_index_);
+
+  return proto;
+}
+
+bool HashJoinOperator::getAllOuterJoinWorkOrderProtos(WorkOrderProtosContainer *container) {
+  // We wait until the building of global hash table is complete.
+  if (!blocking_dependencies_met_) {
+    return false;
+  }
+
+  if (probe_relation_is_stored_) {
+    if (!started_) {
+      for (const block_id probe_block_id : probe_relation_block_ids_) {
+        container->addWorkOrderProto(createOuterJoinWorkOrderProto(probe_block_id), op_index_);
+      }
+      started_ = true;
+    }
+    return true;
+  } else {
+    while (num_workorders_generated_ < probe_relation_block_ids_.size()) {
+      container->addWorkOrderProto(
+          createOuterJoinWorkOrderProto(probe_relation_block_ids_[num_workorders_generated_]),
+          op_index_);
+      ++num_workorders_generated_;
+    }
+
+    return done_feeding_input_relation_;
+  }
+}
+
+serialization::WorkOrder* HashJoinOperator::createOuterJoinWorkOrderProto(const block_id block) {
+  serialization::WorkOrder *proto = new serialization::WorkOrder;
+  proto->set_work_order_type(serialization::HASH_JOIN);
+
+  proto->SetExtension(serialization::HashJoinWorkOrder::hash_join_work_order_type,
+                      serialization::HashJoinWorkOrder::HASH_OUTER_JOIN);
+  proto->SetExtension(serialization::HashJoinWorkOrder::build_relation_id, build_relation_.getID());
+  proto->SetExtension(serialization::HashJoinWorkOrder::probe_relation_id, probe_relation_.getID());
+  for (const attribute_id attr_id : join_key_attributes_) {
+    proto->AddExtension(serialization::HashJoinWorkOrder::join_key_attributes, attr_id);
+  }
+  proto->SetExtension(serialization::HashJoinWorkOrder::any_join_key_attributes_nullable,
+                      any_join_key_attributes_nullable_);
+  proto->SetExtension(serialization::HashJoinWorkOrder::insert_destination_index, output_destination_index_);
+  proto->SetExtension(serialization::HashJoinWorkOrder::join_hash_table_index, hash_table_index_);
+  proto->SetExtension(serialization::HashJoinWorkOrder::selection_index, selection_index_);
+  proto->SetExtension(serialization::HashJoinWorkOrder::block_id, block);
+
+  for (const bool is_attribute_on_build : is_selection_on_build_) {
+    proto->AddExtension(serialization::HashJoinWorkOrder::is_selection_on_build, is_attribute_on_build);
+  }
+
+  return proto;
+}
+
+
 void HashInnerJoinWorkOrder::execute() {
   if (FLAGS_vector_based_joined_tuple_collector) {
     executeWithCollectorType<VectorBasedJoinedTupleCollector>();

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/3e35844f/relational_operators/HashJoinOperator.hpp
----------------------------------------------------------------------
diff --git a/relational_operators/HashJoinOperator.hpp b/relational_operators/HashJoinOperator.hpp
index fcc087a..58610ed 100644
--- a/relational_operators/HashJoinOperator.hpp
+++ b/relational_operators/HashJoinOperator.hpp
@@ -30,6 +30,7 @@
 #include "query_execution/QueryContext.hpp"
 #include "relational_operators/RelationalOperator.hpp"
 #include "relational_operators/WorkOrder.hpp"
+#include "relational_operators/WorkOrder.pb.h"
 #include "storage/HashTable.hpp"
 #include "storage/StorageBlockInfo.hpp"
 #include "utility/Macros.hpp"
@@ -47,6 +48,7 @@ class InsertDestination;
 class Predicate;
 class Scalar;
 class StorageManager;
+class WorkOrderProtosContainer;
 class WorkOrdersContainer;
 
 /** \addtogroup RelationalOperators
@@ -157,6 +159,8 @@ class HashJoinOperator : public RelationalOperator {
                         const tmb::client_id scheduler_client_id,
                         tmb::MessageBus *bus) override;
 
+  bool getAllWorkOrderProtos(WorkOrderProtosContainer *container) override;
+
   void feedInputBlock(const block_id input_block_id,
                       const relation_id input_relation_id) override {
     DCHECK(input_relation_id == probe_relation_.getID());
@@ -198,6 +202,23 @@ class HashJoinOperator : public RelationalOperator {
                                  QueryContext *query_context,
                                  StorageManager *storage_manager);
 
+  bool getAllNonOuterJoinWorkOrderProtos(
+      WorkOrderProtosContainer *container,
+      const serialization::HashJoinWorkOrder::HashJoinWorkOrderType hash_join_type);
+
+  serialization::WorkOrder* createNonOuterJoinWorkOrderProto(
+      const serialization::HashJoinWorkOrder::HashJoinWorkOrderType hash_join_type,
+      const block_id block);
+
+  bool getAllOuterJoinWorkOrderProtos(WorkOrderProtosContainer *container);
+
+  /**
+   * @brief Create HashOuterJoinWorkOrder proto.
+   *
+   * @param block The block id used in the Work Order.
+   **/
+  serialization::WorkOrder* createOuterJoinWorkOrderProto(const block_id block);
+
   const CatalogRelation &build_relation_;
   const CatalogRelation &probe_relation_;
   const bool probe_relation_is_stored_;

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/3e35844f/relational_operators/InsertOperator.cpp
----------------------------------------------------------------------
diff --git a/relational_operators/InsertOperator.cpp b/relational_operators/InsertOperator.cpp
index 8d083e5..3dfde30 100644
--- a/relational_operators/InsertOperator.cpp
+++ b/relational_operators/InsertOperator.cpp
@@ -20,7 +20,9 @@
 #include <memory>
 
 #include "query_execution/QueryContext.hpp"
+#include "query_execution/WorkOrderProtosContainer.hpp"
 #include "query_execution/WorkOrdersContainer.hpp"
+#include "relational_operators/WorkOrder.pb.h"
 #include "storage/InsertDestination.hpp"
 
 #include "glog/logging.h"
@@ -47,6 +49,22 @@ bool InsertOperator::getAllWorkOrders(
   return work_generated_;
 }
 
+bool InsertOperator::getAllWorkOrderProtos(WorkOrderProtosContainer *container) {
+  if (blocking_dependencies_met_ && !work_generated_) {
+    work_generated_ = true;
+
+    serialization::WorkOrder *proto = new serialization::WorkOrder;
+    proto->set_work_order_type(serialization::INSERT);
+    proto->SetExtension(serialization::InsertWorkOrder::insert_destination_index, output_destination_index_);
+    proto->SetExtension(serialization::InsertWorkOrder::tuple_index, tuple_index_);
+
+    container->addWorkOrderProto(proto, op_index_);
+  }
+
+  return work_generated_;
+}
+
+
 void InsertWorkOrder::execute() {
   output_destination_->insertTuple(*tuple_);
 }

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/3e35844f/relational_operators/InsertOperator.hpp
----------------------------------------------------------------------
diff --git a/relational_operators/InsertOperator.hpp b/relational_operators/InsertOperator.hpp
index 8a06c94..133e67f 100644
--- a/relational_operators/InsertOperator.hpp
+++ b/relational_operators/InsertOperator.hpp
@@ -38,6 +38,7 @@ namespace quickstep {
 
 class InsertDestination;
 class StorageManager;
+class WorkOrderProtosContainer;
 class WorkOrdersContainer;
 
 /** \addtogroup RelationalOperators
@@ -73,6 +74,8 @@ class InsertOperator : public RelationalOperator {
                         const tmb::client_id scheduler_client_id,
                         tmb::MessageBus *bus) override;
 
+  bool getAllWorkOrderProtos(WorkOrderProtosContainer *container) override;
+
   QueryContext::insert_destination_id getInsertDestinationID() const override {
     return output_destination_index_;
   }

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/3e35844f/relational_operators/NestedLoopsJoinOperator.cpp
----------------------------------------------------------------------
diff --git a/relational_operators/NestedLoopsJoinOperator.cpp b/relational_operators/NestedLoopsJoinOperator.cpp
index 5cc498b..ca90df6 100644
--- a/relational_operators/NestedLoopsJoinOperator.cpp
+++ b/relational_operators/NestedLoopsJoinOperator.cpp
@@ -26,7 +26,9 @@
 #include "expressions/predicate/Predicate.hpp"
 #include "expressions/scalar/Scalar.hpp"
 #include "query_execution/QueryContext.hpp"
+#include "query_execution/WorkOrderProtosContainer.hpp"
 #include "query_execution/WorkOrdersContainer.hpp"
+#include "relational_operators/WorkOrder.pb.h"
 #include "storage/InsertDestination.hpp"
 #include "storage/StorageBlock.hpp"
 #include "storage/StorageBlockInfo.hpp"
@@ -147,6 +149,72 @@ bool NestedLoopsJoinOperator::getAllWorkOrders(
   }
 }
 
+bool NestedLoopsJoinOperator::getAllWorkOrderProtos(WorkOrderProtosContainer *container) {
+  if (left_relation_is_stored_ && right_relation_is_stored_) {
+    // Make sure we generate workorders only once.
+    if (!all_workorders_generated_) {
+      for (const block_id left_block_id : left_relation_block_ids_) {
+        for (const block_id right_block_id : right_relation_block_ids_) {
+          container->addWorkOrderProto(createWorkOrderProto(left_block_id, right_block_id),
+                                       op_index_);
+        }
+      }
+      all_workorders_generated_ = true;
+    }
+    return true;
+  } else if (!(left_relation_is_stored_ || right_relation_is_stored_)) {
+    // Both relations are not stored.
+    const std::vector<block_id>::size_type new_left_blocks
+        = left_relation_block_ids_.size() - num_left_workorders_generated_;
+    const std::vector<block_id>::size_type new_right_blocks
+        = right_relation_block_ids_.size() - num_right_workorders_generated_;
+
+    std::size_t new_workorders = 0;
+    if (new_left_blocks > 0 && new_right_blocks > 0) {
+      // Blocks added to both left and right relations.
+      // First generate (left + new_left_blocks) * (new_right_blocks).
+      new_workorders =
+          getAllWorkOrderProtosHelperBothNotStored(container,
+                                                   0,
+                                                   left_relation_block_ids_.size(),
+                                                   num_right_workorders_generated_,
+                                                   right_relation_block_ids_.size());
+
+      // Now generate new_left_blocks * (right).
+      new_workorders +=
+          getAllWorkOrderProtosHelperBothNotStored(container,
+                                                   num_left_workorders_generated_,
+                                                   left_relation_block_ids_.size(),
+                                                   0,
+                                                   num_right_workorders_generated_);
+    } else if (new_left_blocks == 0 && new_right_blocks > 0) {
+      // Only new right blocks are added. Generate left * new_right_blocks.
+      new_workorders =
+          getAllWorkOrderProtosHelperBothNotStored(container,
+                                                   0,
+                                                   left_relation_block_ids_.size(),
+                                                   num_right_workorders_generated_,
+                                                   right_relation_block_ids_.size());
+    } else if (new_left_blocks > 0 && new_right_blocks == 0) {
+      // Generate new_left_blocks * right
+      new_workorders =
+          getAllWorkOrderProtosHelperBothNotStored(container,
+                                                   num_left_workorders_generated_,
+                                                   left_relation_block_ids_.size(),
+                                                   0,
+                                                   right_relation_block_ids_.size());
+    }
+    if (new_workorders > 0) {
+      num_left_workorders_generated_ = left_relation_block_ids_.size();
+      num_right_workorders_generated_ = right_relation_block_ids_.size();
+    }
+    return done_feeding_left_relation_ && done_feeding_right_relation_;
+  } else {
+    // Only one relation is a stored relation.
+    return getAllWorkOrderProtosHelperOneStored(container);
+  }
+}
+
 std::size_t NestedLoopsJoinOperator::getAllWorkOrdersHelperBothNotStored(WorkOrdersContainer *container,
                                                                          QueryContext *query_context,
                                                                          StorageManager *storage_manager,
@@ -234,6 +302,79 @@ bool NestedLoopsJoinOperator::getAllWorkOrdersHelperOneStored(WorkOrdersContaine
   }
 }
 
+std::size_t NestedLoopsJoinOperator::getAllWorkOrderProtosHelperBothNotStored(
+    WorkOrderProtosContainer *container,
+    const std::vector<block_id>::size_type left_min,
+    const std::vector<block_id>::size_type left_max,
+    const std::vector<block_id>::size_type right_min,
+    const std::vector<block_id>::size_type right_max) {
+  DCHECK(!(left_relation_is_stored_ || right_relation_is_stored_));
+  DCHECK_LE(left_min, left_max);
+  DCHECK_LE(right_min, right_max);
+
+  for (std::vector<block_id>::size_type left_index = left_min;
+       left_index < left_max;
+       ++left_index) {
+    for (std::vector<block_id>::size_type right_index = right_min;
+         right_index < right_max;
+         ++right_index) {
+      container->addWorkOrderProto(
+          createWorkOrderProto(left_relation_block_ids_[left_index], right_relation_block_ids_[right_index]),
+          op_index_);
+    }
+  }
+  // Return the number of workorders produced.
+  return (left_max - left_min) * (right_max - right_min);
+}
+
+bool NestedLoopsJoinOperator::getAllWorkOrderProtosHelperOneStored(WorkOrderProtosContainer *container) {
+  DCHECK(left_relation_is_stored_ ^ right_relation_is_stored_);
+
+  if (left_relation_is_stored_) {
+    for (std::vector<block_id>::size_type right_index = num_right_workorders_generated_;
+         right_index < right_relation_block_ids_.size();
+         ++right_index) {
+      for (const block_id left_block_id : left_relation_block_ids_) {
+        container->addWorkOrderProto(
+            createWorkOrderProto(left_block_id, right_relation_block_ids_[right_index]),
+            op_index_);
+      }
+    }
+    num_right_workorders_generated_ = right_relation_block_ids_.size();
+    return done_feeding_right_relation_;
+  } else {
+    for (std::vector<block_id>::size_type left_index = num_left_workorders_generated_;
+         left_index < left_relation_block_ids_.size();
+         ++left_index) {
+      for (const block_id right_block_id : right_relation_block_ids_) {
+        container->addWorkOrderProto(
+            createWorkOrderProto(left_relation_block_ids_[left_index], right_block_id),
+            op_index_);
+      }
+    }
+    num_left_workorders_generated_ = left_relation_block_ids_.size();
+    return done_feeding_left_relation_;
+  }
+}
+
+serialization::WorkOrder* NestedLoopsJoinOperator::createWorkOrderProto(const block_id left_block,
+                                                                        const block_id right_block) {
+  serialization::WorkOrder *proto = new serialization::WorkOrder;
+  proto->set_work_order_type(serialization::NESTED_LOOP_JOIN);
+
+  proto->SetExtension(serialization::NestedLoopsJoinWorkOrder::left_relation_id, left_input_relation_.getID());
+  proto->SetExtension(serialization::NestedLoopsJoinWorkOrder::right_relation_id, right_input_relation_.getID());
+  proto->SetExtension(serialization::NestedLoopsJoinWorkOrder::left_block_id, left_block);
+  proto->SetExtension(serialization::NestedLoopsJoinWorkOrder::right_block_id, right_block);
+  proto->SetExtension(serialization::NestedLoopsJoinWorkOrder::insert_destination_index,
+                      output_destination_index_);
+  proto->SetExtension(serialization::NestedLoopsJoinWorkOrder::join_predicate_index, join_predicate_index_);
+  proto->SetExtension(serialization::NestedLoopsJoinWorkOrder::selection_index, selection_index_);
+
+  return proto;
+}
+
+
 template <bool LEFT_PACKED, bool RIGHT_PACKED>
 void NestedLoopsJoinWorkOrder::executeHelper(const TupleStorageSubBlock &left_store,
                                              const TupleStorageSubBlock &right_store) {

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/3e35844f/relational_operators/NestedLoopsJoinOperator.hpp
----------------------------------------------------------------------
diff --git a/relational_operators/NestedLoopsJoinOperator.hpp b/relational_operators/NestedLoopsJoinOperator.hpp
index a52ca25..0d14af5 100644
--- a/relational_operators/NestedLoopsJoinOperator.hpp
+++ b/relational_operators/NestedLoopsJoinOperator.hpp
@@ -44,8 +44,11 @@ class Predicate;
 class Scalar;
 class StorageManager;
 class TupleStorageSubBlock;
+class WorkOrderProtosContainer;
 class WorkOrdersContainer;
 
+namespace serialization { class WorkOrder; }
+
 /** \addtogroup RelationalOperators
  *  @{
  */
@@ -113,6 +116,8 @@ class NestedLoopsJoinOperator : public RelationalOperator {
                         const tmb::client_id scheduler_client_id,
                         tmb::MessageBus *bus) override;
 
+  bool getAllWorkOrderProtos(WorkOrderProtosContainer *container) override;
+
   void doneFeedingInputBlocks(const relation_id rel_id) override {
     if (rel_id == left_input_relation_.getID()) {
       done_feeding_left_relation_ = true;
@@ -181,6 +186,52 @@ class NestedLoopsJoinOperator : public RelationalOperator {
                                        QueryContext *query_context,
                                        StorageManager *storage_manager);
 
+  /**
+   * @brief Pairs block IDs from left and right relation block IDs and generates
+   *        NestedLoopsJoinWorkOrder protos and pushes them to the
+   *        WorkOrderProtosContainer when both relations are not stored
+   *        relations.
+   *
+   * @param container A pointer to the WorkOrderProtosContainer to store the
+   *                  resulting WorkOrder protos.
+   * @param left_min The starting index in left_relation_block_ids_ from where
+   *                 we begin generating NestedLoopsJoinWorkOrders.
+   * @param left_max The index in left_relation_block_ids_ until which we
+   *                 generate NestedLoopsJoinWorkOrders (excluding left_max).
+   * @param right_min The starting index in right_relation_block_ids_ from where
+   *                  we begin generating NestedLoopsJoinWorkOrders.
+   * @param right_max The index in right_relation_block_ids_ until which we
+   *                  generate NestedLoopsJoinWorkOrders. (excluding right_max).
+   *
+   * @return The number of workorder protos generated during the execution of this
+   *         function.
+   **/
+  std::size_t getAllWorkOrderProtosHelperBothNotStored(WorkOrderProtosContainer *container,
+                                                       const std::vector<block_id>::size_type left_min,
+                                                       const std::vector<block_id>::size_type left_max,
+                                                       const std::vector<block_id>::size_type right_min,
+                                                       const std::vector<block_id>::size_type right_max);
+
+  /**
+   * @brief Pairs block IDs from left and right relation block IDs and generates
+   *        NestedLoopsJoinWorkOrder protos and pushes them to the
+   *        WorkOrderProtosContainer when only one relation is a stored relation.
+   *
+   * @param container A pointer to the WorkOrderProtosContainer to store the
+   *                  resulting WorkOrder protos.
+   *
+   * @return Whether all work orders have been generated.
+   **/
+  bool getAllWorkOrderProtosHelperOneStored(WorkOrderProtosContainer *container);
+
+  /**
+   * @brief Create Work Order proto.
+   *
+   * @param block The block id used in the Work Order.
+   **/
+  serialization::WorkOrder* createWorkOrderProto(const block_id left_block,
+                                                 const block_id right_block);
+
   const CatalogRelation &left_input_relation_;
   const CatalogRelation &right_input_relation_;
 

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/3e35844f/relational_operators/RelationalOperator.hpp
----------------------------------------------------------------------
diff --git a/relational_operators/RelationalOperator.hpp b/relational_operators/RelationalOperator.hpp
index 75fde17..2b0373c 100644
--- a/relational_operators/RelationalOperator.hpp
+++ b/relational_operators/RelationalOperator.hpp
@@ -36,6 +36,7 @@ namespace tmb { class MessageBus; }
 namespace quickstep {
 
 class StorageManager;
+class WorkOrderProtosContainer;
 class WorkOrdersContainer;
 
 /** \addtogroup RelationalOperators
@@ -82,6 +83,27 @@ class RelationalOperator {
                                 tmb::MessageBus *bus) = 0;
 
   /**
+    * @brief For the distributed version, generate all the next WorkOrder protos
+    *        for this RelationalOperator
+    *
+    * @note If a RelationalOperator has blocking dependencies, it should not
+    *       generate workorders unless all of the blocking dependencies have been
+    *       met.
+    *
+    * @note If a RelationalOperator is not parallelizeable on a block-level, then
+    *       only one WorkOrder consisting of all the work for this
+    *       RelationalOperator should be generated.
+    *
+    * @param container A pointer to a WorkOrderProtosContainer to be used to
+    *        store the generated WorkOrder protos.
+    *
+    * @return Whether the operator has finished generating work order protos. If
+    *         \c false, the execution engine will invoke this method after at
+    *         least one pending work order has finished executing.
+    **/
+  virtual bool getAllWorkOrderProtos(WorkOrderProtosContainer *container) = 0;
+
+  /**
    * @brief Update Catalog upon the completion of this RelationalOperator, if
    *        necessary.
    *

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/3e35844f/relational_operators/SampleOperator.cpp
----------------------------------------------------------------------
diff --git a/relational_operators/SampleOperator.cpp b/relational_operators/SampleOperator.cpp
index 6842b28..52bbbd4 100644
--- a/relational_operators/SampleOperator.cpp
+++ b/relational_operators/SampleOperator.cpp
@@ -1,6 +1,7 @@
 /**
  *   Copyright 2016, Quickstep Research Group, Computer Sciences Department,
  *     University of Wisconsin—Madison.
+ *   Copyright 2016 Pivotal Software, Inc.
  *
  *   Licensed under the Apache License, Version 2.0 (the "License");
  *   you may not use this file except in compliance with the License.
@@ -18,10 +19,13 @@
 #include "relational_operators/SampleOperator.hpp"
 
 #include <memory>
+#include <random>
 #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/InsertDestination.hpp"
 #include "storage/StorageBlock.hpp"
 #include "storage/StorageBlockInfo.hpp"
@@ -81,35 +85,97 @@ bool SampleOperator::getAllWorkOrders(
     }
     return started_;
   } else {
-      if (is_block_sample_) {
-          while (num_workorders_generated_ < input_relation_block_ids_.size()) {
-            if (distribution(generator) <= probability) {
-              container->addNormalWorkOrder(
-                  new SampleWorkOrder(input_relation_,
-                                      input_relation_block_ids_[num_workorders_generated_],
-                                      is_block_sample_,
-                                      percentage_,
-                                      output_destination,
-                                      storage_manager),
-                  op_index_);
-            ++num_workorders_generated_;
-          }
-        }
-      } else  {
-        while (num_workorders_generated_ < input_relation_block_ids_.size()) {
+    if (is_block_sample_) {
+      while (num_workorders_generated_ < input_relation_block_ids_.size()) {
+        if (distribution(generator) <= probability) {
           container->addNormalWorkOrder(
               new SampleWorkOrder(input_relation_,
                                   input_relation_block_ids_[num_workorders_generated_],
                                   is_block_sample_,
-                                  percentage_, output_destination,
+                                  percentage_,
+                                  output_destination,
                                   storage_manager),
               op_index_);
           ++num_workorders_generated_;
         }
       }
+    } else {
+      while (num_workorders_generated_ < input_relation_block_ids_.size()) {
+        container->addNormalWorkOrder(
+            new SampleWorkOrder(input_relation_,
+                                input_relation_block_ids_[num_workorders_generated_],
+                                is_block_sample_,
+                                percentage_,
+                                output_destination,
+                                storage_manager),
+            op_index_);
+        ++num_workorders_generated_;
+      }
+    }
+    return done_feeding_input_relation_;
+  }
+}
+
+bool SampleOperator::getAllWorkOrderProtos(WorkOrderProtosContainer *container) {
+  std::random_device random_device;
+  std::mt19937 generator(random_device());
+  std::uniform_real_distribution<> distribution(0, 1);
+  const double probability = static_cast<double>(percentage_) / 100;
+
+  if (input_relation_is_stored_) {
+    if (!started_) {
+      // If the sampling is by block choose blocks randomly
+      if (is_block_sample_) {
+        for (const block_id input_block_id : input_relation_block_ids_) {
+          if (distribution(generator) <= probability) {
+            container->addWorkOrderProto(createWorkOrderProto(input_block_id), op_index_);
+          }
+        }
+      } else {
+        // Add all the blocks for tuple sampling which would handle
+        // the sampling from each block
+        for (const block_id input_block_id : input_relation_block_ids_) {
+          container->addWorkOrderProto(createWorkOrderProto(input_block_id), op_index_);
+        }
+      }
+      started_ = true;
+    }
+    return true;
+  } else {
+    if (is_block_sample_) {
+      while (num_workorders_generated_ < input_relation_block_ids_.size()) {
+        if (distribution(generator) <= probability) {
+          container->addWorkOrderProto(
+              createWorkOrderProto(input_relation_block_ids_[num_workorders_generated_]),
+              op_index_);
+          ++num_workorders_generated_;
+        }
+      }
+    } else {
+      while (num_workorders_generated_ < input_relation_block_ids_.size()) {
+        container->addWorkOrderProto(
+            createWorkOrderProto(input_relation_block_ids_[num_workorders_generated_]),
+            op_index_);
+        ++num_workorders_generated_;
+      }
+    }
     return done_feeding_input_relation_;
   }
 }
+
+serialization::WorkOrder* SampleOperator::createWorkOrderProto(const block_id block) {
+  serialization::WorkOrder *proto = new serialization::WorkOrder;
+  proto->set_work_order_type(serialization::SAMPLE);
+
+  proto->SetExtension(serialization::SampleWorkOrder::relation_id, input_relation_.getID());
+  proto->SetExtension(serialization::SampleWorkOrder::block_id, block);
+  proto->SetExtension(serialization::SampleWorkOrder::is_block_sample, is_block_sample_);
+  proto->SetExtension(serialization::SampleWorkOrder::percentage, percentage_);
+  proto->SetExtension(serialization::SampleWorkOrder::insert_destination_index, output_destination_index_);
+
+  return proto;
+}
+
 void SampleWorkOrder::execute() {
   BlockReference block(
       storage_manager_->getBlock(input_block_id_, input_relation_));

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/3e35844f/relational_operators/SampleOperator.hpp
----------------------------------------------------------------------
diff --git a/relational_operators/SampleOperator.hpp b/relational_operators/SampleOperator.hpp
index 305de34..8ebe17f 100644
--- a/relational_operators/SampleOperator.hpp
+++ b/relational_operators/SampleOperator.hpp
@@ -41,8 +41,11 @@ class CatalogDatabase;
 class CatalogRelationSchema;
 class InsertDestination;
 class StorageManager;
+class WorkOrderProtosContainer;
 class WorkOrdersContainer;
 
+namespace serialization { class WorkOrder; }
+
 /** \addtogroup RelationalOperators
  *  @{
  */
@@ -91,6 +94,8 @@ class SampleOperator : public RelationalOperator {
                         const tmb::client_id scheduler_client_id,
                         tmb::MessageBus *bus) override;
 
+  bool getAllWorkOrderProtos(WorkOrderProtosContainer *container) override;
+
   void feedInputBlock(const block_id input_block_id, const relation_id input_relation_id) override {
     input_relation_block_ids_.push_back(input_block_id);
   }
@@ -110,6 +115,13 @@ class SampleOperator : public RelationalOperator {
   }
 
  private:
+  /**
+   * @brief Create Work Order proto.
+   *
+   * @param block The block id used in the Work Order.
+   **/
+  serialization::WorkOrder* createWorkOrderProto(const block_id block);
+
   const CatalogRelation &input_relation_;
   const CatalogRelationSchema &output_relation_;
   const QueryContext::insert_destination_id output_destination_index_;

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/3e35844f/relational_operators/SaveBlocksOperator.cpp
----------------------------------------------------------------------
diff --git a/relational_operators/SaveBlocksOperator.cpp b/relational_operators/SaveBlocksOperator.cpp
index ac61407..c054b1a 100644
--- a/relational_operators/SaveBlocksOperator.cpp
+++ b/relational_operators/SaveBlocksOperator.cpp
@@ -19,7 +19,9 @@
 
 #include <vector>
 
+#include "query_execution/WorkOrderProtosContainer.hpp"
 #include "query_execution/WorkOrdersContainer.hpp"
+#include "relational_operators/WorkOrder.pb.h"
 #include "storage/StorageBlockInfo.hpp"
 #include "storage/StorageManager.hpp"
 
@@ -45,6 +47,21 @@ bool SaveBlocksOperator::getAllWorkOrders(
   return done_feeding_input_relation_;
 }
 
+bool SaveBlocksOperator::getAllWorkOrderProtos(WorkOrderProtosContainer *container) {
+  while (num_workorders_generated_ < destination_block_ids_.size()) {
+    serialization::WorkOrder *proto = new serialization::WorkOrder;
+    proto->set_work_order_type(serialization::SAVE_BLOCKS);
+    proto->SetExtension(serialization::SaveBlocksWorkOrder::block_id,
+                        destination_block_ids_[num_workorders_generated_]);
+    proto->SetExtension(serialization::SaveBlocksWorkOrder::force, force_);
+
+    container->addWorkOrderProto(proto, op_index_);
+
+    ++num_workorders_generated_;
+  }
+  return done_feeding_input_relation_;
+}
+
 void SaveBlocksOperator::feedInputBlock(const block_id input_block_id, const relation_id input_relation_id) {
   destination_block_ids_.push_back(input_block_id);
 }

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/3e35844f/relational_operators/SaveBlocksOperator.hpp
----------------------------------------------------------------------
diff --git a/relational_operators/SaveBlocksOperator.hpp b/relational_operators/SaveBlocksOperator.hpp
index 49195ea..f50176f 100644
--- a/relational_operators/SaveBlocksOperator.hpp
+++ b/relational_operators/SaveBlocksOperator.hpp
@@ -36,6 +36,7 @@ namespace quickstep {
 
 class QueryContext;
 class StorageManager;
+class WorkOrderProtosContainer;
 class WorkOrdersContainer;
 
 /** \addtogroup RelationalOperators
@@ -65,6 +66,8 @@ class SaveBlocksOperator : public RelationalOperator {
                         const tmb::client_id scheduler_client_id,
                         tmb::MessageBus *bus) override;
 
+  bool getAllWorkOrderProtos(WorkOrderProtosContainer *container) override;
+
   void feedInputBlock(const block_id input_block_id, const relation_id input_relation_id) override;
 
   void feedInputBlocks(const relation_id rel_id, std::vector<block_id> *partially_filled_blocks) override {

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/3e35844f/relational_operators/SelectOperator.cpp
----------------------------------------------------------------------
diff --git a/relational_operators/SelectOperator.cpp b/relational_operators/SelectOperator.cpp
index 69bb434..97f9166 100644
--- a/relational_operators/SelectOperator.cpp
+++ b/relational_operators/SelectOperator.cpp
@@ -21,7 +21,9 @@
 #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/InsertDestination.hpp"
 #include "storage/StorageBlock.hpp"
 #include "storage/StorageBlockInfo.hpp"
@@ -167,6 +169,46 @@ bool SelectOperator::getAllWorkOrders(
   }
 }
 
+bool SelectOperator::getAllWorkOrderProtos(WorkOrderProtosContainer *container) {
+  if (input_relation_is_stored_) {
+    if (!started_) {
+      for (const block_id input_block_id : input_relation_block_ids_) {
+        container->addWorkOrderProto(createWorkOrderProto(input_block_id), op_index_);
+      }
+      started_ = true;
+    }
+    return true;
+  } else {
+    while (num_workorders_generated_ < input_relation_block_ids_.size()) {
+      container->addWorkOrderProto(
+          createWorkOrderProto(input_relation_block_ids_[num_workorders_generated_]),
+          op_index_);
+      ++num_workorders_generated_;
+    }
+    return done_feeding_input_relation_;
+  }
+}
+
+serialization::WorkOrder* SelectOperator::createWorkOrderProto(const block_id block) {
+  serialization::WorkOrder *proto = new serialization::WorkOrder;
+  proto->set_work_order_type(serialization::SELECT);
+
+  proto->SetExtension(serialization::SelectWorkOrder::relation_id, input_relation_.getID());
+  proto->SetExtension(serialization::SelectWorkOrder::insert_destination_index, output_destination_index_);
+  proto->SetExtension(serialization::SelectWorkOrder::predicate_index, predicate_index_);
+  proto->SetExtension(serialization::SelectWorkOrder::block_id, block);
+  proto->SetExtension(serialization::SelectWorkOrder::simple_projection, simple_projection_);
+  if (simple_projection_) {
+    for (const attribute_id attr_id : simple_selection_) {
+      proto->AddExtension(serialization::SelectWorkOrder::simple_selection, attr_id);
+    }
+  }
+  proto->SetExtension(serialization::SelectWorkOrder::selection_index, selection_index_);
+
+  return proto;
+}
+
+
 void SelectWorkOrder::execute() {
   BlockReference block(
       storage_manager_->getBlock(input_block_id_, input_relation_, getPreferredNUMANodes()[0]));



Mime
View raw message