diff --git a/catalog/CMakeLists.txt b/catalog/CMakeLists.txt
index 0ca014e..1aa2b41 100644
--- a/catalog/CMakeLists.txt
+++ b/catalog/CMakeLists.txt
@@ -108,6 +108,7 @@
                       quickstep_catalog_Catalog_proto
                       quickstep_catalog_IndexScheme
                       quickstep_catalog_PartitionScheme
+                      quickstep_catalog_PartitionSchemeHeader
                       quickstep_storage_StorageBlockInfo
                       quickstep_storage_StorageBlockLayout
                       quickstep_storage_StorageBlockLayout_proto
diff --git a/catalog/CatalogRelation.cpp b/catalog/CatalogRelation.cpp
index 793fc2d..38c86f6 100644
--- a/catalog/CatalogRelation.cpp
+++ b/catalog/CatalogRelation.cpp
@@ -33,6 +33,7 @@
 #endif
 
 #include "catalog/PartitionScheme.hpp"
+#include "catalog/PartitionSchemeHeader.hpp"
 #include "storage/StorageBlockInfo.hpp"
 #include "storage/StorageBlockLayout.hpp"
 #include "storage/StorageBlockLayout.pb.h"
@@ -180,6 +181,10 @@
 void CatalogRelation::setPartitionScheme(PartitionScheme* partition_scheme) {
   DCHECK_EQ(0u, size_blocks());
   partition_scheme_.reset(partition_scheme);
+
+  if (partition_scheme_) {
+    num_partitions_ = partition_scheme_->getPartitionSchemeHeader().getNumPartitions();
+  }
 }
 
 void CatalogRelation::setDefaultStorageBlockLayout(StorageBlockLayout *default_layout) {
diff --git a/catalog/CatalogRelation.hpp b/catalog/CatalogRelation.hpp
index af7f273..e40b599 100644
--- a/catalog/CatalogRelation.hpp
+++ b/catalog/CatalogRelation.hpp
@@ -81,6 +81,7 @@
                   bool temporary = false)
       : CatalogRelationSchema(parent, name, id, temporary),
         default_layout_(nullptr),
+        num_partitions_(1u),
         statistics_(new CatalogRelationStatistics()) {
   }
 
@@ -138,6 +139,15 @@
   }
 
   /**
+   * @brief Get the number of partitions for the relation.
+   *
+   * @return The number of partitions the relation is partitioned into.
+   **/
+  std::size_t getNumPartitions() const {
+    return num_partitions_;
+  }
+
+  /**
    * @brief Check if a NUMA placement scheme is available for the relation.
    *
    * @return True if the relation has a NUMA placement scheme, false otherwise.
@@ -423,6 +433,7 @@
   // A relation may or may not have a Partition Scheme
   // assosiated with it.
   std::unique_ptr<PartitionScheme> partition_scheme_;
+  std::size_t num_partitions_;
 
   // Index scheme associated with this relation.
   // Defines a set of indices defined for this relation.
diff --git a/query_optimizer/ExecutionGenerator.cpp b/query_optimizer/ExecutionGenerator.cpp
index 9bfd136..d43432c 100644
--- a/query_optimizer/ExecutionGenerator.cpp
+++ b/query_optimizer/ExecutionGenerator.cpp
@@ -747,13 +747,6 @@
                                  insert_destination_proto);
 
   // Create and add a Select operator.
-  const PartitionScheme *input_partition_scheme = input_relation.getPartitionScheme();
-
-  const std::size_t num_partitions =
-      input_partition_scheme
-          ? input_partition_scheme->getPartitionSchemeHeader().getNumPartitions()
-          : 1u;
-
   // Use the "simple" form of the selection operator (a pure projection that
   // doesn't require any expression evaluation or intermediate copies) if
   // possible.
@@ -766,16 +759,14 @@
                                insert_destination_index,
                                execution_predicate_index,
                                move(attributes),
-                               input_relation_info->isStoredRelation(),
-                               num_partitions)
+                               input_relation_info->isStoredRelation())
           : new SelectOperator(query_handle_->query_id(),
                                input_relation,
                                *output_relation,
                                insert_destination_index,
                                execution_predicate_index,
                                project_expressions_group_index,
-                               input_relation_info->isStoredRelation(),
-                               num_partitions);
+                               input_relation_info->isStoredRelation());
 
   const QueryPlan::DAGNodeIndex select_index =
       execution_plan_->addRelationalOperator(op);
@@ -847,12 +838,6 @@
       findRelationInfoOutputByPhysical(build_physical);
 
   const CatalogRelation &build_relation = *build_relation_info->relation;
-  const PartitionScheme *build_partition_scheme = build_relation.getPartitionScheme();
-
-  const std::size_t build_num_partitions =
-      build_partition_scheme
-          ? build_partition_scheme->getPartitionSchemeHeader().getNumPartitions()
-          : 1u;
 
   // Create a BuildLIPFilterOperator for the FilterJoin. This operator builds
   // LIP filters that are applied properly in downstream operators to achieve
@@ -862,7 +847,6 @@
           new BuildLIPFilterOperator(
               query_handle_->query_id(),
               build_relation,
-              build_num_partitions,
               build_side_predicate_index,
               build_relation_info->isStoredRelation()));
 
@@ -954,7 +938,7 @@
 
   const CatalogRelationInfo *build_relation_info =
       findRelationInfoOutputByPhysical(build_physical);
-  const CatalogRelationInfo *probe_operator_info =
+  const CatalogRelationInfo *probe_relation_info =
       findRelationInfoOutputByPhysical(probe_physical);
 
   // Create a vector that indicates whether each project expression is using
@@ -970,9 +954,10 @@
   }
 
   const CatalogRelation *build_relation = build_relation_info->relation;
+  const CatalogRelation *probe_relation = probe_relation_info->relation;
 
   // FIXME(quickstep-team): Add support for self-join.
-  if (build_relation == probe_operator_info->relation) {
+  if (build_relation == probe_relation) {
     THROW_SQL_ERROR() << "Self-join is not supported";
   }
 
@@ -982,9 +967,7 @@
   S::QueryContext::HashTableContext *hash_table_context_proto =
       query_context_proto_->add_join_hash_tables();
 
-  const P::PartitionSchemeHeader *probe_partition_scheme_header = probe_physical->getOutputPartitionSchemeHeader();
-  const std::size_t probe_num_partitions =
-      probe_partition_scheme_header ? probe_partition_scheme_header->num_partitions : 1u;
+  const std::size_t probe_num_partitions = probe_relation->getNumPartitions();
   hash_table_context_proto->set_num_partitions(probe_num_partitions);
 
   S::HashTable *hash_table_proto = hash_table_context_proto->mutable_join_hash_table();
@@ -1052,8 +1035,8 @@
           new HashJoinOperator(
               query_handle_->query_id(),
               *build_relation,
-              *probe_operator_info->relation,
-              probe_operator_info->isStoredRelation(),
+              *probe_relation,
+              probe_relation_info->isStoredRelation(),
               probe_attribute_ids,
               any_probe_attributes_nullable,
               probe_num_partitions,
@@ -1083,9 +1066,9 @@
                                          build_relation_info->producer_operator_index,
                                          true /* is_pipeline_breaker */);
   }
-  if (!probe_operator_info->isStoredRelation()) {
+  if (!probe_relation_info->isStoredRelation()) {
     execution_plan_->addDirectDependency(join_operator_index,
-                                         probe_operator_info->producer_operator_index,
+                                         probe_relation_info->producer_operator_index,
                                          false /* is_pipeline_breaker */);
   }
   execution_plan_->addDirectDependency(join_operator_index,
@@ -1140,15 +1123,11 @@
     THROW_SQL_ERROR() << "NestedLoopsJoin does not support self-join yet";
   }
 
-  const PartitionScheme *left_partition_scheme = left_relation.getPartitionScheme();
-  const std::size_t num_partitions =
-      left_partition_scheme ? left_partition_scheme->getPartitionSchemeHeader().getNumPartitions()
-                            : 1u;
+  const std::size_t num_partitions = left_relation.getNumPartitions();
 
 #ifdef QUICKSTEP_DEBUG
-  const PartitionScheme *right_partition_scheme = right_relation.getPartitionScheme();
-  if (right_partition_scheme) {
-    DCHECK_EQ(num_partitions, right_partition_scheme->getPartitionSchemeHeader().getNumPartitions());
+  if (right_relation.hasPartitionScheme()) {
+    DCHECK_EQ(num_partitions, right_relation.getNumPartitions());
   }
 #endif
 
@@ -1549,12 +1528,6 @@
   const CatalogRelationInfo *selection_relation_info =
       findRelationInfoOutputByPhysical(physical_plan->selection());
   const CatalogRelation &selection_relation = *selection_relation_info->relation;
-  const PartitionScheme *selection_partition_scheme = selection_relation.getPartitionScheme();
-
-  const std::size_t num_partitions =
-      selection_partition_scheme
-          ? selection_partition_scheme->getPartitionSchemeHeader().getNumPartitions()
-          : 1u;
 
   // Prepare the attributes, which are output columns of the selection relation.
   std::vector<attribute_id> attributes;
@@ -1580,8 +1553,7 @@
                          insert_destination_index,
                          QueryContext::kInvalidPredicateId,
                          move(attributes),
-                         selection_relation_info->isStoredRelation(),
-                         num_partitions);
+                         selection_relation_info->isStoredRelation());
 
   const QueryPlan::DAGNodeIndex insert_selection_index =
       execution_plan_->addRelationalOperator(insert_selection_op);
@@ -1756,11 +1728,7 @@
   const CatalogRelationInfo *input_relation_info =
       findRelationInfoOutputByPhysical(physical_plan->input());
   const CatalogRelation &input_relation = *input_relation_info->relation;
-  const PartitionScheme *input_partition_scheme = input_relation.getPartitionScheme();
-  const size_t num_partitions =
-      input_partition_scheme
-          ? input_partition_scheme->getPartitionSchemeHeader().getNumPartitions()
-          : 1u;
+  const size_t num_partitions = input_relation.getNumPartitions();
 
   // Create aggr state proto.
   const QueryContext::aggregation_state_id aggr_state_index =
@@ -2312,7 +2280,9 @@
   // Get input.
   const CatalogRelationInfo *input_relation_info =
       findRelationInfoOutputByPhysical(physical_plan->input());
-  window_aggr_state_proto->set_input_relation_id(input_relation_info->relation->getID());
+  const CatalogRelation &input_relation = *input_relation_info->relation;
+  DCHECK_EQ(1u, input_relation.getNumPartitions());
+  window_aggr_state_proto->set_input_relation_id(input_relation.getID());
 
   // Get window aggregate function expression.
   const E::AliasPtr &named_window_aggregate_expression =
@@ -2382,7 +2352,7 @@
   const QueryPlan::DAGNodeIndex window_aggregation_operator_index =
       execution_plan_->addRelationalOperator(
           new WindowAggregationOperator(query_handle_->query_id(),
-                                        *input_relation_info->relation,
+                                        input_relation,
                                         *output_relation,
                                         window_aggr_state_index,
                                         insert_destination_index));
diff --git a/relational_operators/AggregationOperator.hpp b/relational_operators/AggregationOperator.hpp
index 2fb620b..c673ff5 100644
--- a/relational_operators/AggregationOperator.hpp
+++ b/relational_operators/AggregationOperator.hpp
@@ -76,11 +76,10 @@
                       bool input_relation_is_stored,
                       const QueryContext::aggregation_state_id aggr_state_index,
                       const std::size_t num_partitions)
-      : RelationalOperator(query_id),
+      : RelationalOperator(query_id, num_partitions),
         input_relation_(input_relation),
         input_relation_is_stored_(input_relation_is_stored),
         aggr_state_index_(aggr_state_index),
-        num_partitions_(num_partitions),
         input_relation_block_ids_(num_partitions),
         num_workorders_generated_(num_partitions),
         started_(false) {
@@ -136,7 +135,6 @@
   const CatalogRelation &input_relation_;
   const bool input_relation_is_stored_;
   const QueryContext::aggregation_state_id aggr_state_index_;
-  const std::size_t num_partitions_;
 
   // The index is the partition id.
   std::vector<BlocksInPartition> input_relation_block_ids_;
diff --git a/relational_operators/BuildAggregationExistenceMapOperator.hpp b/relational_operators/BuildAggregationExistenceMapOperator.hpp
index 8791a38..9cfd93d 100644
--- a/relational_operators/BuildAggregationExistenceMapOperator.hpp
+++ b/relational_operators/BuildAggregationExistenceMapOperator.hpp
@@ -79,12 +79,11 @@
                                        const bool input_relation_is_stored,
                                        const QueryContext::aggregation_state_id aggr_state_index,
                                        const std::size_t num_partitions)
-      : RelationalOperator(query_id),
+      : RelationalOperator(query_id, num_partitions),
         input_relation_(input_relation),
         build_attribute_(build_attribute),
         input_relation_is_stored_(input_relation_is_stored),
         aggr_state_index_(aggr_state_index),
-        num_partitions_(num_partitions),
         input_relation_block_ids_(num_partitions),
         num_workorders_generated_(num_partitions),
         started_(false) {
@@ -138,7 +137,6 @@
   const attribute_id build_attribute_;
   const bool input_relation_is_stored_;
   const QueryContext::aggregation_state_id aggr_state_index_;
-  const std::size_t num_partitions_;
 
   // The index is the partition id.
   std::vector<BlocksInPartition> input_relation_block_ids_;
diff --git a/relational_operators/BuildHashOperator.hpp b/relational_operators/BuildHashOperator.hpp
index d18d9fb..dfb6dfe 100644
--- a/relational_operators/BuildHashOperator.hpp
+++ b/relational_operators/BuildHashOperator.hpp
@@ -89,12 +89,11 @@
                     const bool any_join_key_attributes_nullable,
                     const std::size_t num_partitions,
                     const QueryContext::join_hash_table_id hash_table_index)
-      : RelationalOperator(query_id),
+      : RelationalOperator(query_id, num_partitions),
         input_relation_(input_relation),
         input_relation_is_stored_(input_relation_is_stored),
         join_key_attributes_(join_key_attributes),
         any_join_key_attributes_nullable_(any_join_key_attributes_nullable),
-        num_partitions_(num_partitions),
         is_broadcast_join_(num_partitions > 1u && !input_relation.hasPartitionScheme()),
         hash_table_index_(hash_table_index),
         input_relation_block_ids_(num_partitions),
@@ -102,8 +101,9 @@
         started_(false) {
     if (input_relation_is_stored) {
       if (input_relation.hasPartitionScheme()) {
+        DCHECK_EQ(num_partitions_, input_relation.getNumPartitions());
+
         const PartitionScheme &part_scheme = *input_relation.getPartitionScheme();
-        DCHECK_EQ(part_scheme.getPartitionSchemeHeader().getNumPartitions(), num_partitions_);
         for (partition_id part_id = 0; part_id < num_partitions_; ++part_id) {
           input_relation_block_ids_[part_id] = part_scheme.getBlocksInPartition(part_id);
         }
@@ -162,7 +162,6 @@
   const bool input_relation_is_stored_;
   const std::vector<attribute_id> join_key_attributes_;
   const bool any_join_key_attributes_nullable_;
-  const std::size_t num_partitions_;
   const bool is_broadcast_join_;
   const QueryContext::join_hash_table_id hash_table_index_;
 
diff --git a/relational_operators/BuildLIPFilterOperator.hpp b/relational_operators/BuildLIPFilterOperator.hpp
index 9b23dd9..41c3294 100644
--- a/relational_operators/BuildLIPFilterOperator.hpp
+++ b/relational_operators/BuildLIPFilterOperator.hpp
@@ -28,7 +28,6 @@
 #include "catalog/CatalogRelation.hpp"
 #include "catalog/CatalogTypedefs.hpp"
 #include "catalog/PartitionScheme.hpp"
-#include "catalog/PartitionSchemeHeader.hpp"
 #include "query_execution/QueryContext.hpp"
 #include "relational_operators/RelationalOperator.hpp"
 #include "relational_operators/WorkOrder.hpp"
@@ -70,8 +69,6 @@
    *
    * @param query_id The ID of the query to which this operator belongs.
    * @param input_relation The relation to build LIP filters on.
-   * @param num_partitions The number of partitions in 'input_relation'.
-   *        If no partitions, it is one.
    * @param build_side_predicate_index The index of the predicate in QueryContext
    *        where the predicate is to be applied to the input relation before
    *        building the LIP filters (or kInvalidPredicateId if no predicate is
@@ -82,21 +79,18 @@
    **/
   BuildLIPFilterOperator(const std::size_t query_id,
                          const CatalogRelation &input_relation,
-                         const std::size_t num_partitions,
                          const QueryContext::predicate_id build_side_predicate_index,
                          const bool input_relation_is_stored)
-    : RelationalOperator(query_id),
+    : RelationalOperator(query_id, input_relation.getNumPartitions()),
       input_relation_(input_relation),
-      num_partitions_(num_partitions),
       build_side_predicate_index_(build_side_predicate_index),
       input_relation_is_stored_(input_relation_is_stored),
-      input_relation_block_ids_(num_partitions),
-      num_workorders_generated_(num_partitions),
+      input_relation_block_ids_(num_partitions_),
+      num_workorders_generated_(num_partitions_),
       started_(false) {
     if (input_relation_is_stored) {
       if (input_relation.hasPartitionScheme()) {
         const PartitionScheme &part_scheme = *input_relation.getPartitionScheme();
-        DCHECK_EQ(part_scheme.getPartitionSchemeHeader().getNumPartitions(), num_partitions_);
         for (partition_id part_id = 0; part_id < num_partitions_; ++part_id) {
           input_relation_block_ids_[part_id] = part_scheme.getBlocksInPartition(part_id);
         }
@@ -148,7 +142,6 @@
   serialization::WorkOrder* createWorkOrderProto(const partition_id part_id, const block_id block);
 
   const CatalogRelation &input_relation_;
-  const std::size_t num_partitions_;
   const QueryContext::predicate_id build_side_predicate_index_;
   const bool input_relation_is_stored_;
 
diff --git a/relational_operators/CMakeLists.txt b/relational_operators/CMakeLists.txt
index c0def6f..57ba9f9 100644
--- a/relational_operators/CMakeLists.txt
+++ b/relational_operators/CMakeLists.txt
@@ -153,7 +153,6 @@
                       quickstep_catalog_CatalogRelation
                       quickstep_catalog_CatalogTypedefs
                       quickstep_catalog_PartitionScheme
-                      quickstep_catalog_PartitionSchemeHeader
                       quickstep_queryexecution_QueryContext
                       quickstep_queryexecution_WorkOrderProtosContainer
                       quickstep_queryexecution_WorkOrdersContainer
@@ -190,6 +189,7 @@
                       quickstep_catalog_CatalogRelation
                       quickstep_catalog_CatalogRelationSchema
                       quickstep_catalog_CatalogTypedefs
+                      quickstep_catalog_PartitionScheme
                       quickstep_queryexecution_QueryContext
                       quickstep_queryexecution_QueryExecutionMessages_proto
                       quickstep_queryexecution_QueryExecutionTypedefs
@@ -394,7 +394,7 @@
                       glog
                       quickstep_catalog_CatalogRelation
                       quickstep_catalog_CatalogTypedefs
-                      quickstep_catalog_PartitionSchemeHeader
+                      quickstep_catalog_PartitionScheme
                       quickstep_queryexecution_QueryContext
                       quickstep_queryexecution_WorkOrderProtosContainer
                       quickstep_queryexecution_WorkOrdersContainer
@@ -534,6 +534,7 @@
                       quickstep_catalog_CatalogRelation
                       quickstep_catalog_CatalogRelationSchema
                       quickstep_catalog_CatalogTypedefs
+                      quickstep_catalog_PartitionScheme
                       quickstep_queryexecution_QueryContext
                       quickstep_queryexecution_QueryExecutionMessages_proto
                       quickstep_queryexecution_QueryExecutionTypedefs
diff --git a/relational_operators/CreateIndexOperator.hpp b/relational_operators/CreateIndexOperator.hpp
index 09d5d81..9449f53 100644
--- a/relational_operators/CreateIndexOperator.hpp
+++ b/relational_operators/CreateIndexOperator.hpp
@@ -64,7 +64,7 @@
                       CatalogRelation *relation,
                       const std::string &index_name,
                       IndexSubBlockDescription &&index_description)  // NOLINT(whitespace/operators)
-      : RelationalOperator(query_id),
+      : RelationalOperator(query_id, 0u),
         relation_(DCHECK_NOTNULL(relation)),
         index_name_(index_name),
         index_description_(index_description) {}
diff --git a/relational_operators/CreateTableOperator.hpp b/relational_operators/CreateTableOperator.hpp
index 52c7846..827ed72 100644
--- a/relational_operators/CreateTableOperator.hpp
+++ b/relational_operators/CreateTableOperator.hpp
@@ -64,7 +64,7 @@
   CreateTableOperator(const std::size_t query_id,
                       CatalogRelation *relation,
                       CatalogDatabase *database)
-      : RelationalOperator(query_id),
+      : RelationalOperator(query_id, 0u),
         relation_(DCHECK_NOTNULL(relation)),
         database_(DCHECK_NOTNULL(database)) {}
 
diff --git a/relational_operators/DeleteOperator.cpp b/relational_operators/DeleteOperator.cpp
index a8723a9..4958024 100644
--- a/relational_operators/DeleteOperator.cpp
+++ b/relational_operators/DeleteOperator.cpp
@@ -25,6 +25,7 @@
 #include <vector>
 
 #include "catalog/CatalogRelationSchema.hpp"
+#include "catalog/CatalogTypedefs.hpp"
 #include "query_execution/QueryContext.hpp"
 #include "query_execution/QueryExecutionMessages.pb.h"
 #include "query_execution/QueryExecutionUtil.hpp"
@@ -58,36 +59,42 @@
       return true;
     }
 
-    for (const block_id input_block_id : relation_block_ids_) {
-      container->addNormalWorkOrder(
-          new DeleteWorkOrder(query_id_,
-                              relation_,
-                              input_block_id,
-                              predicate,
-                              storage_manager,
-                              op_index_,
-                              scheduler_client_id,
-                              bus),
-          op_index_);
+    for (partition_id part_id = 0; part_id < num_partitions_; ++part_id) {
+      for (const block_id input_block_id : relation_block_ids_[part_id]) {
+        container->addNormalWorkOrder(
+            new DeleteWorkOrder(query_id_,
+                                relation_,
+                                part_id,
+                                input_block_id,
+                                predicate,
+                                storage_manager,
+                                op_index_,
+                                scheduler_client_id,
+                                bus),
+            op_index_);
+      }
     }
     started_ = true;
     return true;
-  } else {
-    while (num_workorders_generated_ < relation_block_ids_.size()) {
+  }
+
+  for (partition_id part_id = 0; part_id < num_partitions_; ++part_id) {
+    while (num_workorders_generated_[part_id] < relation_block_ids_[part_id].size()) {
       container->addNormalWorkOrder(
           new DeleteWorkOrder(query_id_,
                               relation_,
-                              relation_block_ids_[num_workorders_generated_],
+                              part_id,
+                              relation_block_ids_[part_id][num_workorders_generated_[part_id]],
                               predicate,
                               storage_manager,
                               op_index_,
                               scheduler_client_id,
                               bus),
           op_index_);
-      ++num_workorders_generated_;
+      ++num_workorders_generated_[part_id];
     }
-    return done_feeding_input_relation_;
   }
+  return done_feeding_input_relation_;
 }
 
 bool DeleteOperator::getAllWorkOrderProtos(WorkOrderProtosContainer *container) {
@@ -97,23 +104,27 @@
       return true;
     }
 
-    for (const block_id input_block_id : relation_block_ids_) {
-      container->addWorkOrderProto(createWorkOrderProto(input_block_id), op_index_);
+    for (partition_id part_id = 0; part_id < num_partitions_; ++part_id) {
+      for (const block_id input_block_id : relation_block_ids_[part_id]) {
+        container->addWorkOrderProto(createWorkOrderProto(part_id, 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_;
   }
+
+  for (partition_id part_id = 0; part_id < num_partitions_; ++part_id) {
+    while (num_workorders_generated_[part_id] < relation_block_ids_[part_id].size()) {
+      container->addWorkOrderProto(
+          createWorkOrderProto(part_id, relation_block_ids_[part_id][num_workorders_generated_[part_id]]),
+          op_index_);
+      ++num_workorders_generated_[part_id];
+    }
+  }
+  return done_feeding_input_relation_;
 }
 
-serialization::WorkOrder* DeleteOperator::createWorkOrderProto(const block_id block) {
+serialization::WorkOrder* DeleteOperator::createWorkOrderProto(const partition_id part_id, const block_id block) {
   serialization::WorkOrder *proto = new serialization::WorkOrder;
   proto->set_work_order_type(serialization::DELETE);
   proto->set_query_id(query_id_);
@@ -122,6 +133,7 @@
   proto->SetExtension(serialization::DeleteWorkOrder::relation_id, relation_.getID());
   proto->SetExtension(serialization::DeleteWorkOrder::predicate_index, predicate_index_);
   proto->SetExtension(serialization::DeleteWorkOrder::block_id, block);
+  proto->SetExtension(serialization::DeleteWorkOrder::partition_id, part_id);
 
   return proto;
 }
@@ -139,6 +151,7 @@
   proto.set_block_id(input_block_id_);
   proto.set_relation_id(input_relation_.getID());
   proto.set_query_id(query_id_);
+  proto.set_partition_id(partition_id_);
 
   // NOTE(zuyu): Using the heap memory to serialize proto as a c-like string.
   const std::size_t proto_length = proto.ByteSize();
diff --git a/relational_operators/DeleteOperator.hpp b/relational_operators/DeleteOperator.hpp
index aed37f6..b084560 100644
--- a/relational_operators/DeleteOperator.hpp
+++ b/relational_operators/DeleteOperator.hpp
@@ -26,6 +26,7 @@
 
 #include "catalog/CatalogRelation.hpp"
 #include "catalog/CatalogTypedefs.hpp"
+#include "catalog/PartitionScheme.hpp"
 #include "query_execution/QueryContext.hpp"
 #include "query_execution/QueryExecutionTypedefs.hpp"
 #include "relational_operators/RelationalOperator.hpp"
@@ -73,14 +74,26 @@
                  const CatalogRelation &relation,
                  const QueryContext::predicate_id predicate_index,
                  const bool relation_is_stored)
-      : RelationalOperator(query_id),
+      : RelationalOperator(query_id, relation.getNumPartitions()),
         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_block_ids_(num_partitions_),
+        num_workorders_generated_(num_partitions_) {
+    if (relation_is_stored) {
+      if (relation.hasPartitionScheme()) {
+        const PartitionScheme &part_scheme = *relation.getPartitionScheme();
+
+        for (partition_id part_id = 0; part_id < num_partitions_; ++part_id) {
+          relation_block_ids_[part_id] = part_scheme.getBlocksInPartition(part_id);
+        }
+      } else {
+        DCHECK_EQ(1u, num_partitions_);
+        relation_block_ids_[0] = relation.getBlocksSnapshot();
+      }
+    }
+  }
 
   ~DeleteOperator() override {}
 
@@ -107,16 +120,17 @@
   void feedInputBlock(const block_id input_block_id, const relation_id input_relation_id,
                       const partition_id part_id) override {
     DCHECK(!relation_is_stored_);
-    relation_block_ids_.push_back(input_block_id);
+    relation_block_ids_[part_id].push_back(input_block_id);
   }
 
  private:
   /**
    * @brief Create Work Order proto.
    *
+   * @param part_id The partition id.
    * @param block The block id used in the Work Order.
    **/
-  serialization::WorkOrder* createWorkOrderProto(const block_id block);
+  serialization::WorkOrder* createWorkOrderProto(const partition_id part_id, const block_id block);
 
   const CatalogRelation &relation_;
   const QueryContext::predicate_id predicate_index_;
@@ -125,8 +139,8 @@
 
   bool started_;
 
-  std::vector<block_id> relation_block_ids_;
-  std::vector<block_id>::size_type num_workorders_generated_;
+  std::vector<std::vector<block_id>> relation_block_ids_;
+  std::vector<std::size_t> num_workorders_generated_;
 
   DISALLOW_COPY_AND_ASSIGN(DeleteOperator);
 };
@@ -141,6 +155,7 @@
    *
    * @param query_id The ID of the query to which this workorder belongs.
    * @param input_relation The relation to perform the DELETE over.
+   * @param part_id The partition id.
    * @param input_block_id The block Id.
    * @param predicate All tuples matching \c predicate will be deleted (If
    *        NULL, then all tuples will be deleted).
@@ -152,13 +167,14 @@
    **/
   DeleteWorkOrder(const std::size_t query_id,
                   const CatalogRelationSchema &input_relation,
+                  const partition_id part_id,
                   const block_id input_block_id,
                   const Predicate *predicate,
                   StorageManager *storage_manager,
                   const std::size_t delete_operator_index,
                   const tmb::client_id scheduler_client_id,
                   MessageBus *bus)
-      : WorkOrder(query_id),
+      : WorkOrder(query_id, part_id),
         input_relation_(input_relation),
         input_block_id_(input_block_id),
         predicate_(predicate),
diff --git a/relational_operators/DestroyAggregationStateOperator.hpp b/relational_operators/DestroyAggregationStateOperator.hpp
index c5b092f..f285957 100644
--- a/relational_operators/DestroyAggregationStateOperator.hpp
+++ b/relational_operators/DestroyAggregationStateOperator.hpp
@@ -61,9 +61,8 @@
       const std::size_t query_id,
       const QueryContext::aggregation_state_id aggr_state_index,
       const std::size_t num_partitions)
-      : RelationalOperator(query_id),
+      : RelationalOperator(query_id, num_partitions),
         aggr_state_index_(aggr_state_index),
-        num_partitions_(num_partitions),
         work_generated_(false) {}
 
   ~DestroyAggregationStateOperator() override {}
@@ -86,7 +85,6 @@
 
  private:
   const QueryContext::aggregation_state_id aggr_state_index_;
-  const std::size_t num_partitions_;
   bool work_generated_;
 
   DISALLOW_COPY_AND_ASSIGN(DestroyAggregationStateOperator);
diff --git a/relational_operators/DestroyHashOperator.cpp b/relational_operators/DestroyHashOperator.cpp
index 4b93f0c..490c316 100644
--- a/relational_operators/DestroyHashOperator.cpp
+++ b/relational_operators/DestroyHashOperator.cpp
@@ -38,7 +38,7 @@
     return true;
   }
 
-  for (std::size_t part_id = 0; part_id < build_num_partitions_; ++part_id) {
+  for (std::size_t part_id = 0; part_id < num_partitions_; ++part_id) {
     container->addNormalWorkOrder(
         new DestroyHashWorkOrder(query_id_, hash_table_index_, part_id, query_context),
         op_index_);
@@ -52,7 +52,7 @@
     return true;
   }
 
-  for (std::size_t part_id = 0; part_id < build_num_partitions_; ++part_id) {
+  for (std::size_t part_id = 0; part_id < num_partitions_; ++part_id) {
     serialization::WorkOrder *proto = new serialization::WorkOrder;
     proto->set_work_order_type(serialization::DESTROY_HASH);
     proto->set_query_id(query_id_);
diff --git a/relational_operators/DestroyHashOperator.hpp b/relational_operators/DestroyHashOperator.hpp
index 3a6cf0e..6c4c997 100644
--- a/relational_operators/DestroyHashOperator.hpp
+++ b/relational_operators/DestroyHashOperator.hpp
@@ -53,14 +53,13 @@
    * @brief Constructor.
    *
    * @param query_id The ID of the query to which this operator belongs.
-   * @param build_num_partitions The number of partitions in 'build_relation'.
+   * @param num_partitions The number of partitions.
    * @param hash_table_index The index of the JoinHashTable in QueryContext.
    **/
   DestroyHashOperator(const std::size_t query_id,
-                      const std::size_t build_num_partitions,
+                      const std::size_t num_partitions,
                       const QueryContext::join_hash_table_id hash_table_index)
-      : RelationalOperator(query_id),
-        build_num_partitions_(build_num_partitions),
+      : RelationalOperator(query_id, num_partitions),
         hash_table_index_(hash_table_index),
         work_generated_(false) {}
 
@@ -83,7 +82,6 @@
   bool getAllWorkOrderProtos(WorkOrderProtosContainer *container) override;
 
  private:
-  const std::size_t build_num_partitions_;
   const QueryContext::join_hash_table_id hash_table_index_;
   bool work_generated_;
 
diff --git a/relational_operators/FinalizeAggregationOperator.hpp b/relational_operators/FinalizeAggregationOperator.hpp
index a6f1ef6..382a7da 100644
--- a/relational_operators/FinalizeAggregationOperator.hpp
+++ b/relational_operators/FinalizeAggregationOperator.hpp
@@ -72,9 +72,8 @@
       const std::size_t aggr_state_num_partitions,
       const CatalogRelation &output_relation,
       const QueryContext::insert_destination_id output_destination_index)
-      : RelationalOperator(query_id),
+      : RelationalOperator(query_id, num_partitions),
         aggr_state_index_(aggr_state_index),
-        num_partitions_(num_partitions),
         aggr_state_num_partitions_(aggr_state_num_partitions),
         output_relation_(output_relation),
         output_destination_index_(output_destination_index),
@@ -108,7 +107,7 @@
 
  private:
   const QueryContext::aggregation_state_id aggr_state_index_;
-  const std::size_t num_partitions_, aggr_state_num_partitions_;
+  const std::size_t aggr_state_num_partitions_;
   const CatalogRelation &output_relation_;
   const QueryContext::insert_destination_id output_destination_index_;
   bool started_;
diff --git a/relational_operators/HashJoinOperator.hpp b/relational_operators/HashJoinOperator.hpp
index 316be66..519718c 100644
--- a/relational_operators/HashJoinOperator.hpp
+++ b/relational_operators/HashJoinOperator.hpp
@@ -137,13 +137,12 @@
       const QueryContext::scalar_group_id selection_index,
       const std::vector<bool> *is_selection_on_build = nullptr,
       const JoinType join_type = JoinType::kInnerJoin)
-      : RelationalOperator(query_id),
+      : RelationalOperator(query_id, num_partitions),
         build_relation_(build_relation),
         probe_relation_(probe_relation),
         probe_relation_is_stored_(probe_relation_is_stored),
         join_key_attributes_(join_key_attributes),
         any_join_key_attributes_nullable_(any_join_key_attributes_nullable),
-        num_partitions_(num_partitions),
         output_relation_(output_relation),
         output_destination_index_(output_destination_index),
         hash_table_index_(hash_table_index),
@@ -276,7 +275,6 @@
   const bool probe_relation_is_stored_;
   const std::vector<attribute_id> join_key_attributes_;
   const bool any_join_key_attributes_nullable_;
-  const std::size_t num_partitions_;
   const CatalogRelation &output_relation_;
   const QueryContext::insert_destination_id output_destination_index_;
   const QueryContext::join_hash_table_id hash_table_index_;
diff --git a/relational_operators/InitializeAggregationOperator.hpp b/relational_operators/InitializeAggregationOperator.hpp
index 0b052c7..cf9abe5 100644
--- a/relational_operators/InitializeAggregationOperator.hpp
+++ b/relational_operators/InitializeAggregationOperator.hpp
@@ -65,9 +65,8 @@
                                 const QueryContext::aggregation_state_id aggr_state_index,
                                 const std::size_t num_partitions,
                                 const std::size_t aggr_state_num_init_partitions)
-      : RelationalOperator(query_id),
+      : RelationalOperator(query_id, num_partitions),
         aggr_state_index_(aggr_state_index),
-        num_partitions_(num_partitions),
         aggr_state_num_init_partitions_(aggr_state_num_init_partitions),
         started_(false) {}
 
@@ -91,7 +90,7 @@
 
  private:
   const QueryContext::aggregation_state_id aggr_state_index_;
-  const std::size_t num_partitions_, aggr_state_num_init_partitions_;
+  const std::size_t aggr_state_num_init_partitions_;
   bool started_;
 
   DISALLOW_COPY_AND_ASSIGN(InitializeAggregationOperator);
diff --git a/relational_operators/NestedLoopsJoinOperator.hpp b/relational_operators/NestedLoopsJoinOperator.hpp
index 006d496..1b14638 100644
--- a/relational_operators/NestedLoopsJoinOperator.hpp
+++ b/relational_operators/NestedLoopsJoinOperator.hpp
@@ -100,11 +100,10 @@
       const QueryContext::scalar_group_id selection_index,
       const bool left_relation_is_stored,
       const bool right_relation_is_stored)
-      : RelationalOperator(query_id),
+      : RelationalOperator(query_id, num_partitions),
         nested_loops_join_index_(nested_loops_join_index),
         left_input_relation_(left_input_relation),
         right_input_relation_(right_input_relation),
-        num_partitions_(num_partitions),
         output_relation_(output_relation),
         output_destination_index_(output_destination_index),
         join_predicate_index_(join_predicate_index),
@@ -296,8 +295,6 @@
   const CatalogRelation &left_input_relation_;
   const CatalogRelation &right_input_relation_;
 
-  const std::size_t num_partitions_;
-
   const CatalogRelation &output_relation_;
   const QueryContext::insert_destination_id output_destination_index_;
 
diff --git a/relational_operators/RelationalOperator.hpp b/relational_operators/RelationalOperator.hpp
index 5de7eb5..29ae194 100644
--- a/relational_operators/RelationalOperator.hpp
+++ b/relational_operators/RelationalOperator.hpp
@@ -260,6 +260,15 @@
   }
 
   /**
+   * @brief Get the number of partitions of the input relation in the operator.
+   *
+   * @return The number of partitions of the input relation.
+   */
+  std::size_t getNumPartitions() const {
+    return num_partitions_;
+  }
+
+  /**
    * @brief Deploy a group of LIPFilters to this operator.
    */
   void deployLIPFilters(const QueryContext::lip_deployment_id lip_deployment_index,
@@ -273,15 +282,18 @@
    * @brief Constructor
    *
    * @param query_id The ID of the query to which this operator belongs.
-   * @param blocking_dependencies_met If those dependencies which break the
-   *        pipeline have been met.
+   * @param num_partitions The number of partitions of the input relation.
+   *        If create table / index, return zero. If no partitions, return one.
    **/
-  explicit RelationalOperator(const std::size_t query_id)
+  explicit RelationalOperator(const std::size_t query_id,
+                              const std::size_t num_partitions = 1u)
       : query_id_(query_id),
+        num_partitions_(num_partitions),
         done_feeding_input_relation_(false),
         lip_deployment_index_(QueryContext::kInvalidLIPDeploymentId) {}
 
   const std::size_t query_id_;
+  const std::size_t num_partitions_;
 
   bool done_feeding_input_relation_;
   std::size_t op_index_;
diff --git a/relational_operators/SaveBlocksOperator.cpp b/relational_operators/SaveBlocksOperator.cpp
index 9d6c3f6..591de12 100644
--- a/relational_operators/SaveBlocksOperator.cpp
+++ b/relational_operators/SaveBlocksOperator.cpp
@@ -22,6 +22,7 @@
 #include <vector>
 
 #include "catalog/CatalogRelation.hpp"
+#include "catalog/CatalogTypedefs.hpp"
 #include "query_execution/WorkOrderProtosContainer.hpp"
 #include "query_execution/WorkOrdersContainer.hpp"
 #include "relational_operators/WorkOrder.pb.h"
@@ -38,31 +39,37 @@
     StorageManager *storage_manager,
     const tmb::client_id scheduler_client_id,
     tmb::MessageBus *bus) {
-  while (num_workorders_generated_ < destination_block_ids_.size()) {
-    container->addNormalWorkOrder(
-        new SaveBlocksWorkOrder(
-            query_id_,
-            destination_block_ids_[num_workorders_generated_],
-            force_,
-            storage_manager),
-        op_index_);
-    ++num_workorders_generated_;
+  for (partition_id part_id = 0; part_id < num_partitions_; ++part_id) {
+    while (num_workorders_generated_[part_id] < destination_block_ids_[part_id].size()) {
+      container->addNormalWorkOrder(
+          new SaveBlocksWorkOrder(
+              query_id_,
+              part_id,
+              destination_block_ids_[part_id][num_workorders_generated_[part_id]],
+              force_,
+              storage_manager),
+          op_index_);
+      ++num_workorders_generated_[part_id];
+    }
   }
   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->set_query_id(query_id_);
-    proto->SetExtension(serialization::SaveBlocksWorkOrder::block_id,
-                        destination_block_ids_[num_workorders_generated_]);
-    proto->SetExtension(serialization::SaveBlocksWorkOrder::force, force_);
+  for (partition_id part_id = 0; part_id < num_partitions_; ++part_id) {
+    while (num_workorders_generated_[part_id] < destination_block_ids_[part_id].size()) {
+      serialization::WorkOrder *proto = new serialization::WorkOrder;
+      proto->set_work_order_type(serialization::SAVE_BLOCKS);
+      proto->set_query_id(query_id_);
+      proto->SetExtension(serialization::SaveBlocksWorkOrder::block_id,
+                          destination_block_ids_[part_id][num_workorders_generated_[part_id]]);
+      proto->SetExtension(serialization::SaveBlocksWorkOrder::force, force_);
+      proto->SetExtension(serialization::SaveBlocksWorkOrder::partition_id, part_id);
 
-    container->addWorkOrderProto(proto, op_index_);
+      container->addWorkOrderProto(proto, op_index_);
 
-    ++num_workorders_generated_;
+      ++num_workorders_generated_[part_id];
+    }
   }
   return done_feeding_input_relation_;
 }
diff --git a/relational_operators/SaveBlocksOperator.hpp b/relational_operators/SaveBlocksOperator.hpp
index 4489299..cea2d7e 100644
--- a/relational_operators/SaveBlocksOperator.hpp
+++ b/relational_operators/SaveBlocksOperator.hpp
@@ -24,6 +24,7 @@
 #include <string>
 #include <vector>
 
+#include "catalog/CatalogRelation.hpp"
 #include "catalog/CatalogTypedefs.hpp"
 #include "relational_operators/RelationalOperator.hpp"
 #include "relational_operators/WorkOrder.hpp"
@@ -38,7 +39,6 @@
 
 namespace quickstep {
 
-class CatalogRelation;
 class QueryContext;
 class StorageManager;
 class WorkOrderProtosContainer;
@@ -64,10 +64,11 @@
   SaveBlocksOperator(const std::size_t query_id,
                      CatalogRelation *relation,
                      const bool force = false)
-      : RelationalOperator(query_id),
+      : RelationalOperator(query_id, relation->getNumPartitions()),
         force_(force),
         relation_(relation),
-        num_workorders_generated_(0) {}
+        destination_block_ids_(num_partitions_),
+        num_workorders_generated_(num_partitions_) {}
 
   ~SaveBlocksOperator() override {}
 
@@ -89,7 +90,7 @@
 
   void feedInputBlock(const block_id input_block_id, const relation_id input_relation_id,
                       const partition_id part_id) override {
-    destination_block_ids_.push_back(input_block_id);
+    destination_block_ids_[part_id].push_back(input_block_id);
   }
 
   void updateCatalogOnCompletion() override;
@@ -98,9 +99,9 @@
   const bool force_;
 
   CatalogRelation *relation_;
-  std::vector<block_id> destination_block_ids_;
+  std::vector<std::vector<block_id>> destination_block_ids_;
 
-  std::vector<block_id>::size_type num_workorders_generated_;
+  std::vector<std::size_t> num_workorders_generated_;
 
   DISALLOW_COPY_AND_ASSIGN(SaveBlocksOperator);
 };
@@ -114,16 +115,18 @@
    * @brief Constructor.
    *
    * @param query_id The ID of the query to which this operator belongs.
+   * @param part_id The partition id.
    * @param save_block_id The id of the block to save.
    * @param force If true, force writing of all blocks to disk, otherwise only
    *        write dirty blocks.
    * @param storage_manager The StorageManager to use.
    **/
   SaveBlocksWorkOrder(const std::size_t query_id,
+                      const partition_id part_id,
                       const block_id save_block_id,
                       const bool force,
                       StorageManager *storage_manager)
-      : WorkOrder(query_id),
+      : WorkOrder(query_id, part_id),
         save_block_id_(save_block_id),
         force_(force),
         storage_manager_(DCHECK_NOTNULL(storage_manager)) {}
diff --git a/relational_operators/SelectOperator.hpp b/relational_operators/SelectOperator.hpp
index f4937b3..a71f4c1 100644
--- a/relational_operators/SelectOperator.hpp
+++ b/relational_operators/SelectOperator.hpp
@@ -32,7 +32,7 @@
 #include "catalog/NUMAPlacementScheme.hpp"
 #endif
 
-#include "catalog/PartitionSchemeHeader.hpp"
+#include "catalog/PartitionScheme.hpp"
 #include "query_execution/QueryContext.hpp"
 #include "relational_operators/RelationalOperator.hpp"
 #include "relational_operators/WorkOrder.hpp"
@@ -85,8 +85,6 @@
    * @param input_relation_is_stored If input_relation is a stored relation and
    *        is fully available to the operator before it can start generating
    *        workorders.
-   * @param num_partitions The number of partitions in 'input_relation'.
-   *        If no partitions, it is one.
    **/
   SelectOperator(
       const std::size_t query_id,
@@ -95,17 +93,15 @@
       const QueryContext::insert_destination_id output_destination_index,
       const QueryContext::predicate_id predicate_index,
       const QueryContext::scalar_group_id selection_index,
-      const bool input_relation_is_stored,
-      const std::size_t num_partitions)
-      : RelationalOperator(query_id),
+      const bool input_relation_is_stored)
+      : RelationalOperator(query_id, input_relation.getNumPartitions()),
         input_relation_(input_relation),
         output_relation_(output_relation),
         output_destination_index_(output_destination_index),
         predicate_index_(predicate_index),
         selection_index_(selection_index),
-        num_partitions_(num_partitions),
-        input_relation_block_ids_(num_partitions),
-        num_workorders_generated_(num_partitions),
+        input_relation_block_ids_(num_partitions_),
+        num_workorders_generated_(num_partitions_),
         simple_projection_(false),
         input_relation_is_stored_(input_relation_is_stored),
         started_(false) {
@@ -120,6 +116,7 @@
           input_relation_block_ids_[part_id] = part_scheme.getBlocksInPartition(part_id);
         }
       } else {
+        DCHECK_EQ(1u, num_partitions_);
         input_relation_block_ids_[0] = input_relation.getBlocksSnapshot();
       }
     }
@@ -143,8 +140,6 @@
    * @param input_relation_is_stored If input_relation is a stored relation and
    *        is fully available to the operator before it can start generating
    *        workorders.
-   * @param num_partitions The number of partitions in 'input_relation'.
-   *        If no partitions, it is one.
    **/
   SelectOperator(
       const std::size_t query_id,
@@ -153,18 +148,16 @@
       const QueryContext::insert_destination_id output_destination_index,
       const QueryContext::predicate_id predicate_index,
       std::vector<attribute_id> &&selection,
-      const bool input_relation_is_stored,
-      const std::size_t num_partitions)
-      : RelationalOperator(query_id),
+      const bool input_relation_is_stored)
+      : RelationalOperator(query_id, input_relation.getNumPartitions()),
         input_relation_(input_relation),
         output_relation_(output_relation),
         output_destination_index_(output_destination_index),
         predicate_index_(predicate_index),
         selection_index_(QueryContext::kInvalidScalarGroupId),
         simple_selection_(std::move(selection)),
-        num_partitions_(num_partitions),
-        input_relation_block_ids_(num_partitions),
-        num_workorders_generated_(num_partitions),
+        input_relation_block_ids_(num_partitions_),
+        num_workorders_generated_(num_partitions_),
         simple_projection_(true),
         input_relation_is_stored_(input_relation_is_stored),
         started_(false) {
@@ -179,6 +172,7 @@
           input_relation_block_ids_[part_id] = part_scheme.getBlocksInPartition(part_id);
         }
       } else {
+        DCHECK_EQ(1u, num_partitions_);
         input_relation_block_ids_[0] = input_relation.getBlocksSnapshot();
       }
     }
@@ -239,7 +233,6 @@
   const QueryContext::scalar_group_id selection_index_;
   const std::vector<attribute_id> simple_selection_;
 
-  const std::size_t num_partitions_;
   // A vector of vectors V where V[i] indicates the list of block IDs of the
   // input relation that belong to the partition i.
   std::vector<std::vector<block_id>> input_relation_block_ids_;
diff --git a/relational_operators/UpdateOperator.cpp b/relational_operators/UpdateOperator.cpp
index 08bfa59..0f7f999 100644
--- a/relational_operators/UpdateOperator.cpp
+++ b/relational_operators/UpdateOperator.cpp
@@ -25,6 +25,7 @@
 #include <utility>
 
 #include "catalog/CatalogRelationSchema.hpp"
+#include "catalog/CatalogTypedefs.hpp"
 #include "query_execution/QueryContext.hpp"
 #include "query_execution/QueryExecutionMessages.pb.h"
 #include "query_execution/QueryExecutionUtil.hpp"
@@ -57,21 +58,24 @@
   }
 
   DCHECK(query_context != nullptr);
-  for (const block_id input_block_id : input_blocks_) {
-    container->addNormalWorkOrder(
-        new UpdateWorkOrder(
-            query_id_,
-            relation_,
-            input_block_id,
-            query_context->getPredicate(predicate_index_),
-            query_context->getUpdateGroup(update_group_index_),
-            query_context->getInsertDestination(
-                relocation_destination_index_),
-            storage_manager,
-            op_index_,
-            scheduler_client_id,
-            bus),
-        op_index_);
+  for (partition_id part_id = 0; part_id < num_partitions_; ++part_id) {
+    for (const block_id input_block_id : input_blocks_[part_id]) {
+      container->addNormalWorkOrder(
+          new UpdateWorkOrder(
+              query_id_,
+              relation_,
+              part_id,
+              input_block_id,
+              query_context->getPredicate(predicate_index_),
+              query_context->getUpdateGroup(update_group_index_),
+              query_context->getInsertDestination(
+                  relocation_destination_index_),
+              storage_manager,
+              op_index_,
+              scheduler_client_id,
+              bus),
+          op_index_);
+    }
   }
   started_ = true;
   return true;
@@ -82,19 +86,22 @@
     return true;
   }
 
-  for (const block_id input_block_id : input_blocks_) {
-    serialization::WorkOrder *proto = new serialization::WorkOrder;
-    proto->set_work_order_type(serialization::UPDATE);
-    proto->set_query_id(query_id_);
+  for (std::size_t part_id = 0; part_id < num_partitions_; ++part_id) {
+    for (const block_id input_block_id : input_blocks_[part_id]) {
+      serialization::WorkOrder *proto = new serialization::WorkOrder;
+      proto->set_work_order_type(serialization::UPDATE);
+      proto->set_query_id(query_id_);
 
-    proto->SetExtension(serialization::UpdateWorkOrder::operator_index, op_index_);
-    proto->SetExtension(serialization::UpdateWorkOrder::relation_id, relation_.getID());
-    proto->SetExtension(serialization::UpdateWorkOrder::insert_destination_index, relocation_destination_index_);
-    proto->SetExtension(serialization::UpdateWorkOrder::predicate_index, predicate_index_);
-    proto->SetExtension(serialization::UpdateWorkOrder::update_group_index, update_group_index_);
-    proto->SetExtension(serialization::UpdateWorkOrder::block_id, input_block_id);
+      proto->SetExtension(serialization::UpdateWorkOrder::operator_index, op_index_);
+      proto->SetExtension(serialization::UpdateWorkOrder::relation_id, relation_.getID());
+      proto->SetExtension(serialization::UpdateWorkOrder::insert_destination_index, relocation_destination_index_);
+      proto->SetExtension(serialization::UpdateWorkOrder::predicate_index, predicate_index_);
+      proto->SetExtension(serialization::UpdateWorkOrder::update_group_index, update_group_index_);
+      proto->SetExtension(serialization::UpdateWorkOrder::block_id, input_block_id);
+      proto->SetExtension(serialization::UpdateWorkOrder::partition_id, part_id);
 
-    container->addWorkOrderProto(proto, op_index_);
+      container->addWorkOrderProto(proto, op_index_);
+    }
   }
   started_ = true;
   return true;
@@ -120,6 +127,7 @@
   proto.set_block_id(input_block_id_);
   proto.set_relation_id(relation_.getID());
   proto.set_query_id(query_id_);
+  proto.set_partition_id(partition_id_);
 
   // NOTE(zuyu): Using the heap memory to serialize proto as a c-like string.
   const std::size_t proto_length = proto.ByteSize();
diff --git a/relational_operators/UpdateOperator.hpp b/relational_operators/UpdateOperator.hpp
index 8e020d8..fd2096b 100644
--- a/relational_operators/UpdateOperator.hpp
+++ b/relational_operators/UpdateOperator.hpp
@@ -28,6 +28,7 @@
 
 #include "catalog/CatalogRelation.hpp"
 #include "catalog/CatalogTypedefs.hpp"
+#include "catalog/PartitionScheme.hpp"
 #include "query_execution/QueryContext.hpp"
 #include "query_execution/QueryExecutionTypedefs.hpp"
 #include "relational_operators/RelationalOperator.hpp"
@@ -87,13 +88,24 @@
       const QueryContext::insert_destination_id relocation_destination_index,
       const QueryContext::predicate_id predicate_index,
       const QueryContext::update_group_id update_group_index)
-      : RelationalOperator(query_id),
+      : RelationalOperator(query_id, relation.getNumPartitions()),
         relation_(relation),
         relocation_destination_index_(relocation_destination_index),
         predicate_index_(predicate_index),
         update_group_index_(update_group_index),
-        input_blocks_(relation.getBlocksSnapshot()),
-        started_(false) {}
+        input_blocks_(num_partitions_),
+        started_(false) {
+    if (relation.hasPartitionScheme()) {
+      const PartitionScheme &part_scheme = *relation.getPartitionScheme();
+
+      for (partition_id part_id = 0; part_id < num_partitions_; ++part_id) {
+        input_blocks_[part_id] = part_scheme.getBlocksInPartition(part_id);
+      }
+    } else {
+      DCHECK_EQ(1u, num_partitions_);
+      input_blocks_[0] = relation.getBlocksSnapshot();
+    }
+  }
 
   ~UpdateOperator() override {}
 
@@ -127,7 +139,7 @@
   const QueryContext::predicate_id predicate_index_;
   const QueryContext::update_group_id update_group_index_;
 
-  const std::vector<block_id> input_blocks_;
+  std::vector<std::vector<block_id>> input_blocks_;
 
   bool started_;
 
@@ -144,6 +156,7 @@
    *
    * @param query_id The ID of the query to which this WorkOrder belongs.
    * @param relation The relation to perform the UPDATE over.
+   * @param part_id The partition id.
    * @param predicate All tuples matching \c predicate will be updated (or NULL
    *        to update all tuples).
    * @param assignments The assignments (the map of attribute_ids to Scalars)
@@ -161,6 +174,7 @@
   UpdateWorkOrder(
       const std::size_t query_id,
       const CatalogRelationSchema &relation,
+      const partition_id part_id,
       const block_id input_block_id,
       const Predicate *predicate,
       const std::unordered_map<attribute_id, std::unique_ptr<const Scalar>>
@@ -170,7 +184,7 @@
       const std::size_t update_operator_index,
       const tmb::client_id scheduler_client_id,
       MessageBus *bus)
-      : WorkOrder(query_id),
+      : WorkOrder(query_id, part_id),
         relation_(relation),
         input_block_id_(input_block_id),
         predicate_(predicate),
diff --git a/relational_operators/WorkOrder.proto b/relational_operators/WorkOrder.proto
index 42a0e7d..6dafbe0 100644
--- a/relational_operators/WorkOrder.proto
+++ b/relational_operators/WorkOrder.proto
@@ -116,6 +116,7 @@
     optional int32 relation_id = 97;
     optional int32 predicate_index = 98;
     optional fixed64 block_id = 99;
+    optional uint64 partition_id = 100;
   }
 }
 
@@ -235,6 +236,7 @@
     // All required.
     optional fixed64 block_id = 224;
     optional bool force = 225;
+    optional uint64 partition_id = 226;
   }
 }
 
@@ -320,6 +322,7 @@
     optional int32 predicate_index = 323;
     optional uint32 update_group_index = 324;
     optional fixed64 block_id = 325;
+    optional uint64 partition_id = 326;
   }
 }
 
diff --git a/relational_operators/WorkOrderFactory.cpp b/relational_operators/WorkOrderFactory.cpp
index 68f7f55..5baa21b 100644
--- a/relational_operators/WorkOrderFactory.cpp
+++ b/relational_operators/WorkOrderFactory.cpp
@@ -170,11 +170,16 @@
               proto.GetExtension(serialization::BuildHashWorkOrder::lip_deployment_index), query_context));
     }
     case serialization::DELETE: {
-      LOG(INFO) << "Creating DeleteWorkOrder for Query " << query_id << " in Shiftboss " << shiftboss_index;
+      const partition_id part_id =
+          proto.GetExtension(serialization::DeleteWorkOrder::partition_id);
+
+      LOG(INFO) << "Creating DeleteWorkOrder (Partition " << part_id << ") for Query " << query_id
+                << " in Shiftboss " << shiftboss_index;
       return new DeleteWorkOrder(
           query_id,
           catalog_database->getRelationSchemaById(
               proto.GetExtension(serialization::DeleteWorkOrder::relation_id)),
+          part_id,
           proto.GetExtension(serialization::DeleteWorkOrder::block_id),
           query_context->getPredicate(
               proto.GetExtension(serialization::DeleteWorkOrder::predicate_index)),
@@ -436,10 +441,14 @@
           storage_manager);
     }
     case serialization::SAVE_BLOCKS: {
-      LOG(INFO) << "Creating SaveBlocksWorkOrder for Query " << query_id
+      const partition_id part_id =
+          proto.GetExtension(serialization::SaveBlocksWorkOrder::partition_id);
+
+      LOG(INFO) << "Creating SaveBlocksWorkOrder (Partition " << part_id << ") for Query " << query_id
                 << " in Shiftboss " << shiftboss_index;
       return new SaveBlocksWorkOrder(
           query_id,
+          part_id,
           proto.GetExtension(serialization::SaveBlocksWorkOrder::block_id),
           proto.GetExtension(serialization::SaveBlocksWorkOrder::force),
           storage_manager);
@@ -563,11 +572,16 @@
           storage_manager);
     }
     case serialization::UPDATE: {
-      LOG(INFO) << "Creating UpdateWorkOrder for Query " << query_id << " in Shiftboss " << shiftboss_index;
+      const partition_id part_id =
+          proto.GetExtension(serialization::UpdateWorkOrder::partition_id);
+
+      LOG(INFO) << "Creating UpdateWorkOrder (Partition " << part_id << ") for Query " << query_id
+                << " in Shiftboss " << shiftboss_index;
       return new UpdateWorkOrder(
           query_id,
           catalog_database->getRelationSchemaById(
               proto.GetExtension(serialization::UpdateWorkOrder::relation_id)),
+          part_id,
           proto.GetExtension(serialization::UpdateWorkOrder::block_id),
           query_context->getPredicate(
               proto.GetExtension(serialization::UpdateWorkOrder::predicate_index)),
@@ -600,7 +614,7 @@
     default:
       LOG(FATAL) << "Unknown WorkOrder Type in WorkOrderFactory::ReconstructFromProto in Shiftboss" << shiftboss_index;
   }
-}
+}  // NOLINT(readability/fn_size)
 
 bool WorkOrderFactory::ProtoIsValid(const serialization::WorkOrder &proto,
                                     const CatalogDatabaseLite &catalog_database,
@@ -726,7 +740,8 @@
              query_context.isValidPredicate(
                  proto.GetExtension(serialization::DeleteWorkOrder::predicate_index)) &&
              proto.HasExtension(serialization::DeleteWorkOrder::block_id) &&
-             proto.HasExtension(serialization::DeleteWorkOrder::operator_index);
+             proto.HasExtension(serialization::DeleteWorkOrder::operator_index) &&
+             proto.GetExtension(serialization::DeleteWorkOrder::partition_id);
     }
     case serialization::DESTROY_AGGREGATION_STATE: {
       return proto.HasExtension(serialization::DestroyAggregationStateWorkOrder::aggr_state_index) &&
@@ -885,7 +900,8 @@
     }
     case serialization::SAVE_BLOCKS: {
       return proto.HasExtension(serialization::SaveBlocksWorkOrder::block_id) &&
-             proto.HasExtension(serialization::SaveBlocksWorkOrder::force);
+             proto.HasExtension(serialization::SaveBlocksWorkOrder::force) &&
+             proto.HasExtension(serialization::SaveBlocksWorkOrder::partition_id);
     }
     case serialization::SELECT: {
       if (!proto.HasExtension(serialization::SelectWorkOrder::relation_id) ||
@@ -1016,7 +1032,8 @@
              query_context.isValidUpdateGroupId(
                  proto.GetExtension(serialization::UpdateWorkOrder::update_group_index)) &&
              proto.HasExtension(serialization::UpdateWorkOrder::operator_index) &&
-             proto.HasExtension(serialization::UpdateWorkOrder::block_id);
+             proto.HasExtension(serialization::UpdateWorkOrder::block_id) &&
+             proto.GetExtension(serialization::UpdateWorkOrder::partition_id);
     }
     case serialization::WINDOW_AGGREGATION: {
       return proto.HasExtension(serialization::WindowAggregationWorkOrder::window_aggr_state_index) &&