diff --git a/relational_operators/tests/HashJoinOperator_unittest.cpp b/relational_operators/tests/HashJoinOperator_unittest.cpp
index 1fc84fc..cfd4314 100644
--- a/relational_operators/tests/HashJoinOperator_unittest.cpp
+++ b/relational_operators/tests/HashJoinOperator_unittest.cpp
@@ -485,7 +485,7 @@
   InsertDestination *insert_destination = query_context_->getInsertDestination(prober->getInsertDestinationID());
   DCHECK(insert_destination);
 
-  const std::vector<block_id> &result_blocks = insert_destination->getTouchedBlocks();
+  const std::vector<block_id> result_blocks = insert_destination->getTouchedBlocks();
   for (std::size_t bid = 0; bid < result_blocks.size(); ++bid) {
     BlockReference result_block = storage_manager_->getBlock(result_blocks[bid],
                                                              insert_destination->getRelation());
@@ -640,7 +640,7 @@
   InsertDestination *insert_destination = query_context_->getInsertDestination(prober->getInsertDestinationID());
   DCHECK(insert_destination);
 
-  const std::vector<block_id> &result_blocks = insert_destination->getTouchedBlocks();
+  const std::vector<block_id> result_blocks = insert_destination->getTouchedBlocks();
   for (std::size_t bid = 0; bid < result_blocks.size(); ++bid) {
     BlockReference result_block = storage_manager_->getBlock(result_blocks[bid],
                                                              insert_destination->getRelation());
@@ -795,7 +795,7 @@
   InsertDestination *insert_destination = query_context_->getInsertDestination(prober->getInsertDestinationID());
   DCHECK(insert_destination);
 
-  const std::vector<block_id> &result_blocks = insert_destination->getTouchedBlocks();
+  const std::vector<block_id> result_blocks = insert_destination->getTouchedBlocks();
   for (std::size_t bid = 0; bid < result_blocks.size(); ++bid) {
     BlockReference result_block = storage_manager_->getBlock(result_blocks[bid],
                                                              insert_destination->getRelation());
@@ -944,7 +944,7 @@
   InsertDestination *insert_destination = query_context_->getInsertDestination(prober->getInsertDestinationID());
   DCHECK(insert_destination);
 
-  const std::vector<block_id> &result_blocks = insert_destination->getTouchedBlocks();
+  const std::vector<block_id> result_blocks = insert_destination->getTouchedBlocks();
   for (std::size_t bid = 0; bid < result_blocks.size(); ++bid) {
     BlockReference result_block = storage_manager_->getBlock(result_blocks[bid],
                                                              insert_destination->getRelation());
@@ -1123,7 +1123,7 @@
   InsertDestination *insert_destination = query_context_->getInsertDestination(prober->getInsertDestinationID());
   DCHECK(insert_destination);
 
-  const std::vector<block_id> &result_blocks = insert_destination->getTouchedBlocks();
+  const std::vector<block_id> result_blocks = insert_destination->getTouchedBlocks();
   for (std::size_t bid = 0; bid < result_blocks.size(); ++bid) {
     BlockReference result_block = storage_manager_->getBlock(result_blocks[bid],
                                                              insert_destination->getRelation());
@@ -1313,7 +1313,7 @@
   InsertDestination *insert_destination = query_context_->getInsertDestination(prober->getInsertDestinationID());
   DCHECK(insert_destination);
 
-  const std::vector<block_id> &result_blocks = insert_destination->getTouchedBlocks();
+  const std::vector<block_id> result_blocks = insert_destination->getTouchedBlocks();
   for (std::size_t bid = 0; bid < result_blocks.size(); ++bid) {
     BlockReference result_block = storage_manager_->getBlock(result_blocks[bid],
                                                              insert_destination->getRelation());
@@ -1370,7 +1370,7 @@
 }
 
 // Hash join tests with single attribute partitions.
-TEST_P(HashJoinOperatorTest, SinlgeAttributePartitionedLongKeyHashJoinTest) {
+TEST_P(HashJoinOperatorTest, SingleAttributePartitionedLongKeyHashJoinTest) {
   insertTuplesWithSingleAttributePartitions();
 
   // Setup the hash table proto in the query context proto.
@@ -1480,7 +1480,7 @@
   InsertDestination *insert_destination = query_context_->getInsertDestination(prober->getInsertDestinationID());
   DCHECK(insert_destination);
 
-  const std::vector<block_id> &result_blocks = insert_destination->getTouchedBlocks();
+  const std::vector<block_id> result_blocks = insert_destination->getTouchedBlocks();
   for (std::size_t bid = 0; bid < result_blocks.size(); ++bid) {
     BlockReference result_block = storage_manager_->getBlock(result_blocks[bid],
                                                              insert_destination->getRelation());
@@ -1514,7 +1514,7 @@
   db_->dropRelationById(output_relation_id);
 }
 
-TEST_P(HashJoinOperatorTest, SinlgeAttributePartitionedCompositeKeyHashJoinTest) {
+TEST_P(HashJoinOperatorTest, SingleAttributePartitionedCompositeKeyHashJoinTest) {
   insertTuplesWithSingleAttributePartitions();
 
   // Setup the hash table proto in the query context proto.
@@ -1629,7 +1629,7 @@
   InsertDestination *insert_destination = query_context_->getInsertDestination(prober->getInsertDestinationID());
   DCHECK(insert_destination);
 
-  const std::vector<block_id> &result_blocks = insert_destination->getTouchedBlocks();
+  const std::vector<block_id> result_blocks = insert_destination->getTouchedBlocks();
   for (std::size_t bid = 0; bid < result_blocks.size(); ++bid) {
     BlockReference result_block = storage_manager_->getBlock(result_blocks[bid],
                                                              insert_destination->getRelation());
@@ -1685,7 +1685,7 @@
   db_->dropRelationById(output_relation_id);
 }
 
-TEST_P(HashJoinOperatorTest, SinlgeAttributePartitionedCompositeKeyHashJoinWithResidualPredicateTest) {
+TEST_P(HashJoinOperatorTest, SingleAttributePartitionedCompositeKeyHashJoinWithResidualPredicateTest) {
   insertTuplesWithSingleAttributePartitions();
 
   // Setup the hash table proto in the query context proto.
@@ -1810,7 +1810,7 @@
   InsertDestination *insert_destination = query_context_->getInsertDestination(prober->getInsertDestinationID());
   DCHECK(insert_destination);
 
-  const std::vector<block_id> &result_blocks = insert_destination->getTouchedBlocks();
+  const std::vector<block_id> result_blocks = insert_destination->getTouchedBlocks();
   for (std::size_t bid = 0; bid < result_blocks.size(); ++bid) {
     BlockReference result_block = storage_manager_->getBlock(result_blocks[bid],
                                                              insert_destination->getRelation());
diff --git a/storage/InsertDestination.cpp b/storage/InsertDestination.cpp
index ff39c55..8821019 100644
--- a/storage/InsertDestination.cpp
+++ b/storage/InsertDestination.cpp
@@ -475,7 +475,7 @@
   sendBlockFilledMessage(block->getID());
 }
 
-const std::vector<block_id>& BlockPoolInsertDestination::getTouchedBlocksInternal() {
+std::vector<block_id> BlockPoolInsertDestination::getTouchedBlocksInternal() {
   for (std::vector<MutableBlockReference>::size_type i = 0; i < available_block_refs_.size(); ++i) {
     done_block_ids_.push_back(available_block_refs_[i]->getID());
   }
@@ -546,27 +546,22 @@
   return storage_manager_->getBlockMutable(new_id, relation_);
 }
 
-const std::vector<block_id>& PartitionAwareInsertDestination::getTouchedBlocksInternal() {
+std::vector<block_id> PartitionAwareInsertDestination::getTouchedBlocksInternal() {
+  std::vector<block_id> all_partitions_done_block_ids;
   // Iterate through each partition and get all the touched blocks.
   for (std::size_t part_id = 0;
        part_id < partition_scheme_header_->getNumPartitions();
        ++part_id) {
-    done_block_ids_[part_id] = getTouchedBlocksInternalInPartition(part_id);
-    all_partitions_done_block_ids_.insert(
-        all_partitions_done_block_ids_.end(), done_block_ids_[part_id].begin(), done_block_ids_[part_id].end());
+    for (std::size_t i = 0; i < available_block_refs_[part_id].size(); ++i) {
+      done_block_ids_[part_id].push_back(available_block_refs_[part_id][i]->getID());
+    }
+    available_block_refs_[part_id].clear();
+
+    all_partitions_done_block_ids.insert(
+        all_partitions_done_block_ids.end(), done_block_ids_[part_id].begin(), done_block_ids_[part_id].end());
     done_block_ids_[part_id].clear();
   }
-  return all_partitions_done_block_ids_;
-}
-
-const std::vector<block_id>& PartitionAwareInsertDestination::getTouchedBlocksInternalInPartition(
-    partition_id part_id) {
-  for (std::vector<MutableBlockReference>::size_type i = 0; i < available_block_refs_[part_id].size(); ++i) {
-    done_block_ids_[part_id].push_back(available_block_refs_[part_id][i]->getID());
-  }
-  available_block_refs_[part_id].clear();
-
-  return done_block_ids_[part_id];
+  return all_partitions_done_block_ids;
 }
 
 PartitionSchemeHeader::PartitionAttributeIds PartitionAwareInsertDestination::getPartitioningAttributes() const {
diff --git a/storage/InsertDestination.hpp b/storage/InsertDestination.hpp
index c3d5641..a0a7bc2 100644
--- a/storage/InsertDestination.hpp
+++ b/storage/InsertDestination.hpp
@@ -58,6 +58,8 @@
 
 namespace merge_run_operator {
 class RunCreator;
+class RunMergerTest;
+class RunTest;
 }  // namespace merge_run_operator
 
 namespace serialization { class InsertDestination; }
@@ -173,21 +175,6 @@
   }
 
   /**
-   * @brief Get the set of blocks that were used by clients of this
-   *        InsertDestination for insertion.
-   * @warning Should only be called AFTER this InsertDestination will no longer
-   *          be used, and all blocks have been returned to it via
-   *          returnBlock().
-   *
-   * @return A reference to a vector of block_ids of blocks that were used for
-   *         insertion.
-   **/
-  const std::vector<block_id>& getTouchedBlocks() {
-    SpinMutexLock lock(mutex_);
-    return getTouchedBlocksInternal();
-  }
-
-  /**
    * @brief Get the set of blocks that were partially filled by clients of this
    *        InsertDestination for insertion.
    * @warning Should only be called AFTER this InsertDestination will no longer
@@ -233,8 +220,6 @@
   // this without holding the mutex.
   virtual MutableBlockReference createNewBlock() = 0;
 
-  virtual const std::vector<block_id>& getTouchedBlocksInternal() = 0;
-
   /**
    * @brief When a StorageBlock becomes full, pipeline the block id to the
    *        scheduler.
@@ -311,12 +296,44 @@
   SpinMutex mutex_;
 
  private:
+  /**
+   * @brief Get the set of blocks that were used by clients of this
+   *        InsertDestination for insertion.
+   * @warning Should only be called AFTER this InsertDestination will no longer
+   *          be used, and all blocks have been returned to it via
+   *          returnBlock().
+   *
+   * @return A vector of block_ids of blocks that were used for insertion.
+   **/
+  std::vector<block_id> getTouchedBlocks() {
+    SpinMutexLock lock(mutex_);
+    return getTouchedBlocksInternal();
+  }
+
+  virtual std::vector<block_id> getTouchedBlocksInternal() = 0;
+
   // TODO(shoban): Workaround to support sort. Sort needs finegrained control of
   // blocks being used to insert, since inserting in an arbitrary block could
   // lead to unsorted results. InsertDestination API changed while sort was
   // being implemented.
   friend class merge_run_operator::RunCreator;
 
+  // NOTE(zuyu): Access getTouchedBlocks.
+  friend class AggregationOperatorTest;
+  friend class merge_run_operator::RunTest;
+  friend class merge_run_operator::RunMergerTest;
+
+  FRIEND_TEST(HashJoinOperatorTest, LongKeyHashJoinTest);
+  FRIEND_TEST(HashJoinOperatorTest, IntDuplicateKeyHashJoinTest);
+  FRIEND_TEST(HashJoinOperatorTest, CharKeyCartesianProductHashJoinTest);
+  FRIEND_TEST(HashJoinOperatorTest, VarCharDuplicateKeyHashJoinTest);
+  FRIEND_TEST(HashJoinOperatorTest, CompositeKeyHashJoinTest);
+  FRIEND_TEST(HashJoinOperatorTest, CompositeKeyHashJoinWithResidualPredicateTest);
+  FRIEND_TEST(HashJoinOperatorTest, SingleAttributePartitionedLongKeyHashJoinTest);
+  FRIEND_TEST(HashJoinOperatorTest, SingleAttributePartitionedCompositeKeyHashJoinTest);
+  FRIEND_TEST(HashJoinOperatorTest,
+              SingleAttributePartitionedCompositeKeyHashJoinWithResidualPredicateTest);
+
   DISALLOW_COPY_AND_ASSIGN(InsertDestination);
 };
 
@@ -358,15 +375,15 @@
 
   MutableBlockReference createNewBlock() override;
 
-  const std::vector<block_id>& getTouchedBlocksInternal() override {
-    return returned_block_ids_;
-  }
-
   void getPartiallyFilledBlocks(std::vector<MutableBlockReference> *partial_blocks,
                                 std::vector<partition_id> *part_ids) override {
   }
 
  private:
+  std::vector<block_id> getTouchedBlocksInternal() override {
+    return returned_block_ids_;
+  }
+
   std::vector<block_id> returned_block_ids_;
 
   DISALLOW_COPY_AND_ASSIGN(AlwaysCreateBlockInsertDestination);
@@ -454,11 +471,11 @@
   void getPartiallyFilledBlocks(std::vector<MutableBlockReference> *partial_blocks,
                                 std::vector<partition_id> *part_ids) override;
 
-  const std::vector<block_id>& getTouchedBlocksInternal() override;
-
   MutableBlockReference createNewBlock() override;
 
  private:
+  std::vector<block_id> getTouchedBlocksInternal() override;
+
   FRIEND_TEST(QueryManagerTest, TwoNodesDAGPartiallyFilledBlocksTest);
 
   // A vector of references to blocks which are loaded in memory.
@@ -585,10 +602,9 @@
   MutableBlockReference createNewBlock() override;
   MutableBlockReference createNewBlockInPartition(const partition_id part_id);
 
-  const std::vector<block_id>& getTouchedBlocksInternal() override;
-  const std::vector<block_id>& getTouchedBlocksInternalInPartition(partition_id part_id);
-
  private:
+  std::vector<block_id> getTouchedBlocksInternal() override;
+
   /**
    * @brief Get the set of blocks that were partially filled by clients of this
    *        InsertDestination for insertion.
@@ -652,8 +668,6 @@
   std::vector< std::vector<block_id> > available_block_ids_;
   // A vector of done block ids for each partition.
   std::vector< std::vector<block_id> > done_block_ids_;
-  // Done block ids across all partitions.
-  std::vector<block_id> all_partitions_done_block_ids_;
   // Mutex for locking each partition separately.
   SpinMutex *mutexes_for_partition_;