diff --git a/query_execution/QueryContext.hpp b/query_execution/QueryContext.hpp
index 7876821..e65f096 100644
--- a/query_execution/QueryContext.hpp
+++ b/query_execution/QueryContext.hpp
@@ -489,6 +489,22 @@
   }
 
   /**
+   * @brief Whether the given vector of Tuple ids is valid.
+   *
+   * @param ids The vector of Tuple ids.
+   *
+   * @return True if valid, otherwise false.
+   **/
+  bool areValidTupleIds(const std::vector<tuple_id> &ids) const {
+    for (const tuple_id id : ids) {
+      if (id >= tuples_.size()) {
+        return false;
+      }
+    }
+    return true;
+  }
+
+  /**
    * @brief Release the ownership of the Tuple referenced by the id.
    *
    * @note Each id should use only once.
diff --git a/query_optimizer/ExecutionGenerator.cpp b/query_optimizer/ExecutionGenerator.cpp
index 14d8949..b0d3c48 100644
--- a/query_optimizer/ExecutionGenerator.cpp
+++ b/query_optimizer/ExecutionGenerator.cpp
@@ -1461,72 +1461,75 @@
       *catalog_database_->getRelationById(
           input_relation_info->relation->getID());
 
-  for (const std::vector<expressions::ScalarLiteralPtr> &tuple : physical_plan->column_values()) {
-    // Construct the tuple proto to be inserted.
-    const QueryContext::tuple_id tuple_index = query_context_proto_->tuples_size();
 
+  // Construct the tuple proto to be inserted.
+  std::vector<QueryContext::tuple_id> tuple_indexes;
+
+  for (const std::vector<expressions::ScalarLiteralPtr> &tuple : physical_plan->column_values()) {
+    const QueryContext::tuple_id tuple_index = query_context_proto_->tuples_size();
     S::Tuple *tuple_proto = query_context_proto_->add_tuples();
     for (const E::ScalarLiteralPtr &literal : tuple) {
       tuple_proto->add_attribute_values()->CopyFrom(literal->value().getProto());
     }
-
-    // FIXME(qzeng): A better way is using a traits struct to look up whether a storage
-    //               block supports ad-hoc insertion instead of hard-coding the block types.
-    const StorageBlockLayout &storage_block_layout =
-        input_relation.getDefaultStorageBlockLayout();
-    if (storage_block_layout.getDescription().tuple_store_description().sub_block_type() ==
-        TupleStorageSubBlockDescription::COMPRESSED_COLUMN_STORE ||
-        storage_block_layout.getDescription().tuple_store_description().sub_block_type() ==
-              TupleStorageSubBlockDescription::COMPRESSED_PACKED_ROW_STORE) {
-      THROW_SQL_ERROR() << "INSERT statement is not supported for the relation "
-                        << input_relation.getName()
-                        << ", because its storage blocks do not support ad-hoc insertion";
-    }
-
-    // Create InsertDestination proto.
-    const QueryContext::insert_destination_id insert_destination_index =
-        query_context_proto_->insert_destinations_size();
-    S::InsertDestination *insert_destination_proto = query_context_proto_->add_insert_destinations();
-
-    insert_destination_proto->set_relation_id(input_relation.getID());
-    insert_destination_proto->mutable_layout()->MergeFrom(
-        input_relation.getDefaultStorageBlockLayout().getDescription());
-
-    if (input_relation.hasPartitionScheme()) {
-      insert_destination_proto->set_insert_destination_type(S::InsertDestinationType::PARTITION_AWARE);
-      insert_destination_proto->MutableExtension(S::PartitionAwareInsertDestination::partition_scheme)
-          ->MergeFrom(input_relation.getPartitionScheme()->getProto());
-    } else {
-      insert_destination_proto->set_insert_destination_type(S::InsertDestinationType::BLOCK_POOL);
-
-      const vector<block_id> blocks(input_relation.getBlocksSnapshot());
-      for (const block_id block : blocks) {
-        insert_destination_proto->AddExtension(S::BlockPoolInsertDestination::blocks, block);
-      }
-    }
-
-    const QueryPlan::DAGNodeIndex insert_operator_index =
-        execution_plan_->addRelationalOperator(
-            new InsertOperator(query_handle_->query_id(),
-                               input_relation,
-                               insert_destination_index,
-                               tuple_index));
-    insert_destination_proto->set_relational_op_index(insert_operator_index);
-
-    CatalogRelation *mutable_relation =
-        catalog_database_->getRelationByIdMutable(input_relation.getID());
-    const QueryPlan::DAGNodeIndex save_blocks_index =
-        execution_plan_->addRelationalOperator(
-            new SaveBlocksOperator(query_handle_->query_id(), mutable_relation));
-    if (!input_relation_info->isStoredRelation()) {
-      execution_plan_->addDirectDependency(insert_operator_index,
-                                           input_relation_info->producer_operator_index,
-                                           true /* is_pipeline_breaker */);
-    }
-    execution_plan_->addDirectDependency(save_blocks_index,
-                                         insert_operator_index,
-                                         false /* is_pipeline_breaker */);
+    tuple_indexes.push_back(tuple_index);
   }
+
+  // FIXME(qzeng): A better way is using a traits struct to look up whether a storage
+  //               block supports ad-hoc insertion instead of hard-coding the block types.
+  const StorageBlockLayout &storage_block_layout =
+      input_relation.getDefaultStorageBlockLayout();
+  if (storage_block_layout.getDescription().tuple_store_description().sub_block_type() ==
+      TupleStorageSubBlockDescription::COMPRESSED_COLUMN_STORE ||
+      storage_block_layout.getDescription().tuple_store_description().sub_block_type() ==
+            TupleStorageSubBlockDescription::COMPRESSED_PACKED_ROW_STORE) {
+    THROW_SQL_ERROR() << "INSERT statement is not supported for the relation "
+                      << input_relation.getName()
+                      << ", because its storage blocks do not support ad-hoc insertion";
+  }
+
+  // Create InsertDestination proto.
+  const QueryContext::insert_destination_id insert_destination_index =
+      query_context_proto_->insert_destinations_size();
+  S::InsertDestination *insert_destination_proto = query_context_proto_->add_insert_destinations();
+
+  insert_destination_proto->set_relation_id(input_relation.getID());
+  insert_destination_proto->mutable_layout()->MergeFrom(
+      input_relation.getDefaultStorageBlockLayout().getDescription());
+
+  if (input_relation.hasPartitionScheme()) {
+    insert_destination_proto->set_insert_destination_type(S::InsertDestinationType::PARTITION_AWARE);
+    insert_destination_proto->MutableExtension(S::PartitionAwareInsertDestination::partition_scheme)
+        ->MergeFrom(input_relation.getPartitionScheme()->getProto());
+  } else {
+    insert_destination_proto->set_insert_destination_type(S::InsertDestinationType::BLOCK_POOL);
+
+    const vector<block_id> blocks(input_relation.getBlocksSnapshot());
+    for (const block_id block : blocks) {
+      insert_destination_proto->AddExtension(S::BlockPoolInsertDestination::blocks, block);
+    }
+  }
+
+  const QueryPlan::DAGNodeIndex insert_operator_index =
+      execution_plan_->addRelationalOperator(
+          new InsertOperator(query_handle_->query_id(),
+                             input_relation,
+                             insert_destination_index,
+                             tuple_indexes));
+  insert_destination_proto->set_relational_op_index(insert_operator_index);
+
+  CatalogRelation *mutable_relation =
+      catalog_database_->getRelationByIdMutable(input_relation.getID());
+  const QueryPlan::DAGNodeIndex save_blocks_index =
+      execution_plan_->addRelationalOperator(
+          new SaveBlocksOperator(query_handle_->query_id(), mutable_relation));
+  if (!input_relation_info->isStoredRelation()) {
+    execution_plan_->addDirectDependency(insert_operator_index,
+                                         input_relation_info->producer_operator_index,
+                                         true /* is_pipeline_breaker */);
+  }
+  execution_plan_->addDirectDependency(save_blocks_index,
+                                       insert_operator_index,
+                                       false /* is_pipeline_breaker */);
 }
 
 void ExecutionGenerator::convertInsertSelection(
diff --git a/query_optimizer/tests/execution_generator/Insert.test b/query_optimizer/tests/execution_generator/Insert.test
index 1be7be9..8131fb2 100644
--- a/query_optimizer/tests/execution_generator/Insert.test
+++ b/query_optimizer/tests/execution_generator/Insert.test
@@ -132,3 +132,18 @@
 |          5|                      2016-01-01T00:00:00|                    NULL|                             abc|
 +-----------+-----------------------------------------+------------------------+--------------------------------+
 ==
+
+CREATE TABLE bar5 (x INT NULL, y INT);
+
+INSERT INTO bar5 VALUES (1,2),(3,4),(5,6);
+
+SELECT * FROM bar5;
+--
++-----------+-----------+
+|x          |y          |
++-----------+-----------+
+|          1|          2|
+|          3|          4|
+|          5|          6|
++-----------+-----------+
+==
diff --git a/query_optimizer/tests/resolver/Insert.test b/query_optimizer/tests/resolver/Insert.test
index 88fff53..f21ce5c 100644
--- a/query_optimizer/tests/resolver/Insert.test
+++ b/query_optimizer/tests/resolver/Insert.test
@@ -153,3 +153,32 @@
 ERROR: Unrecognized relation undefined_table (1 : 13)
 insert into undefined_table values (1, 2)
             ^
+==
+
+insert into test values (null, 1, 2, 3, 'foo', 'foo'),(null, 4, 5, 6, 'foo', 'foo');
+--
+TopLevelPlan
++-plan=InsertTuple
+| +-input=TableReference[relation_name=Test,relation_alias=test]
+| | +-AttributeReference[id=0,name=int_col,relation=test,type=Int NULL]
+| | +-AttributeReference[id=1,name=long_col,relation=test,type=Long]
+| | +-AttributeReference[id=2,name=float_col,relation=test,type=Float]
+| | +-AttributeReference[id=3,name=double_col,relation=test,type=Double NULL]
+| | +-AttributeReference[id=4,name=char_col,relation=test,type=Char(20)]
+| | +-AttributeReference[id=5,name=vchar_col,relation=test,type=VarChar(20) NULL]
+| +-column_values=
+| | +-Literal[value=NULL,type=Int NULL]
+| | +-Literal[value=1,type=Long]
+| | +-Literal[value=2,type=Float]
+| | +-Literal[value=3,type=Double NULL]
+| | +-Literal[value=foo,type=Char(20)]
+| | +-Literal[value=foo,type=VarChar(20) NULL]
+| +-column_values=
+|   +-Literal[value=NULL,type=Int NULL]
+|   +-Literal[value=4,type=Long]
+|   +-Literal[value=5,type=Float]
+|   +-Literal[value=6,type=Double NULL]
+|   +-Literal[value=foo,type=Char(20)]
+|   +-Literal[value=foo,type=VarChar(20) NULL]
++-output_attributes=
+  +-[]
diff --git a/relational_operators/InsertOperator.cpp b/relational_operators/InsertOperator.cpp
index fbd3a07..b8c9f07 100644
--- a/relational_operators/InsertOperator.cpp
+++ b/relational_operators/InsertOperator.cpp
@@ -20,6 +20,7 @@
 #include "relational_operators/InsertOperator.hpp"
 
 #include <memory>
+#include <vector>
 
 #include "query_execution/QueryContext.hpp"
 #include "query_execution/WorkOrderProtosContainer.hpp"
@@ -43,12 +44,19 @@
     return true;
   }
 
+  std::vector<std::unique_ptr<Tuple>> tuples;
+
+  for (const QueryContext::tuple_id tuple_index : tuple_indexes_) {
+    std::unique_ptr<Tuple> newTuple(query_context->releaseTuple(tuple_index));
+    tuples.push_back(std::move(newTuple));
+  }
+
   DCHECK(query_context != nullptr);
   container->addNormalWorkOrder(
       new InsertWorkOrder(
           query_id_,
           query_context->getInsertDestination(output_destination_index_),
-          query_context->releaseTuple(tuple_index_)),
+          std::move(tuples)),
       op_index_);
 
   work_generated_ = true;
@@ -64,7 +72,9 @@
   proto->set_work_order_type(serialization::INSERT);
   proto->set_query_id(query_id_);
   proto->SetExtension(serialization::InsertWorkOrder::insert_destination_index, output_destination_index_);
-  proto->SetExtension(serialization::InsertWorkOrder::tuple_index, tuple_index_);
+  for (const QueryContext::tuple_id tuple_index : tuple_indexes_) {
+    proto->AddExtension(serialization::InsertWorkOrder::tuple_indexes, tuple_index);
+  }
 
   container->addWorkOrderProto(proto, op_index_);
 
@@ -74,7 +84,9 @@
 
 
 void InsertWorkOrder::execute() {
-  output_destination_->insertTuple(*tuple_);
+  for (const auto &tuple : tuples_) {
+    output_destination_->insertTuple(*tuple);
+  }
 }
 
 }  // namespace quickstep
diff --git a/relational_operators/InsertOperator.hpp b/relational_operators/InsertOperator.hpp
index b103538..3865a7f 100644
--- a/relational_operators/InsertOperator.hpp
+++ b/relational_operators/InsertOperator.hpp
@@ -23,6 +23,7 @@
 #include <cstddef>
 #include <string>
 #include <memory>
+#include <vector>
 
 #include "catalog/CatalogRelation.hpp"
 #include "catalog/CatalogTypedefs.hpp"
@@ -67,11 +68,11 @@
       const std::size_t query_id,
       const CatalogRelation &output_relation,
       const QueryContext::insert_destination_id output_destination_index,
-      const QueryContext::tuple_id tuple_index)
+      const std::vector<QueryContext::tuple_id> &tuple_indexes)
       : RelationalOperator(query_id, 1u, false, output_relation.getNumPartitions()),
         output_relation_(output_relation),
         output_destination_index_(output_destination_index),
-        tuple_index_(tuple_index),
+        tuple_indexes_(tuple_indexes),
         work_generated_(false) {}
 
   ~InsertOperator() override {}
@@ -103,7 +104,7 @@
  private:
   const CatalogRelation &output_relation_;
   const QueryContext::insert_destination_id output_destination_index_;
-  const QueryContext::tuple_id tuple_index_;
+  const std::vector<QueryContext::tuple_id> tuple_indexes_;
   bool work_generated_;
 
   DISALLOW_COPY_AND_ASSIGN(InsertOperator);
@@ -125,10 +126,10 @@
    **/
   InsertWorkOrder(const std::size_t query_id,
                   InsertDestination *output_destination,
-                  Tuple *tuple)
+                  std::vector<std::unique_ptr<Tuple>> &&tuples)
       : WorkOrder(query_id),
         output_destination_(DCHECK_NOTNULL(output_destination)),
-        tuple_(DCHECK_NOTNULL(tuple)) {}
+        tuples_(std::move(tuples)) {}
 
   ~InsertWorkOrder() override {}
 
@@ -140,7 +141,7 @@
 
  private:
   InsertDestination *output_destination_;
-  std::unique_ptr<Tuple> tuple_;
+  std::vector<std::unique_ptr<Tuple>> tuples_;
 
   DISALLOW_COPY_AND_ASSIGN(InsertWorkOrder);
 };
diff --git a/relational_operators/WorkOrder.proto b/relational_operators/WorkOrder.proto
index aaf7929..b84e758 100644
--- a/relational_operators/WorkOrder.proto
+++ b/relational_operators/WorkOrder.proto
@@ -201,7 +201,7 @@
   extend WorkOrder {
     // All required.
     optional int32 insert_destination_index = 176;
-    optional uint32 tuple_index = 177;
+    repeated uint32 tuple_indexes = 177;
   }
 }
 
diff --git a/relational_operators/WorkOrderFactory.cpp b/relational_operators/WorkOrderFactory.cpp
index 3a991bd..7f11e3e 100644
--- a/relational_operators/WorkOrderFactory.cpp
+++ b/relational_operators/WorkOrderFactory.cpp
@@ -395,12 +395,22 @@
     }
     case serialization::INSERT: {
       LOG(INFO) << "Creating InsertWorkOrder for Query " << query_id << " in Shiftboss " << shiftboss_index;
+
+      const int tuple_count = proto.ExtensionSize(serialization::InsertWorkOrder::tuple_indexes);
+      std::vector<std::unique_ptr<Tuple>> tuple_indexes;
+
+      for (int specific_tuple_index = 0; specific_tuple_index < tuple_count; specific_tuple_index++) {
+        const int tuple_index =
+            proto.GetExtension(serialization::InsertWorkOrder::tuple_indexes, specific_tuple_index);
+        tuple_indexes.emplace_back(
+            std::unique_ptr<Tuple>(query_context->releaseTuple(tuple_index)));
+      }
+
       return new InsertWorkOrder(
           query_id,
           query_context->getInsertDestination(
               proto.GetExtension(serialization::InsertWorkOrder::insert_destination_index)),
-          query_context->releaseTuple(
-              proto.GetExtension(serialization::InsertWorkOrder::tuple_index)));
+              std::move(tuple_indexes));
     }
     case serialization::NESTED_LOOP_JOIN: {
       const partition_id part_id =
@@ -852,12 +862,20 @@
              proto.HasExtension(serialization::InitializeAggregationWorkOrder::state_partition_id);
     }
     case serialization::INSERT: {
+      const int tuple_count = proto.ExtensionSize(serialization::InsertWorkOrder::tuple_indexes);
+      std::vector<QueryContext::tuple_id> tuple_indexes;
+
+      for (int specific_tuple_index = 0; specific_tuple_index < tuple_count; specific_tuple_index++) {
+        const int tuple_index =
+            proto.GetExtension(serialization::InsertWorkOrder::tuple_indexes, specific_tuple_index);
+        tuple_indexes.push_back(tuple_index);
+      }
+
       return proto.HasExtension(serialization::InsertWorkOrder::insert_destination_index) &&
              query_context.isValidInsertDestinationId(
                  proto.GetExtension(serialization::InsertWorkOrder::insert_destination_index)) &&
-             proto.HasExtension(serialization::InsertWorkOrder::tuple_index) &&
-             query_context.isValidTupleId(
-                 proto.GetExtension(serialization::InsertWorkOrder::tuple_index));
+             proto.HasExtension(serialization::InsertWorkOrder::tuple_indexes) &&
+             query_context.areValidTupleIds(tuple_indexes);
     }
     case serialization::NESTED_LOOP_JOIN: {
       if (!proto.HasExtension(serialization::NestedLoopsJoinWorkOrder::left_relation_id) ||