diff --git a/Android.bp b/Android.bp
index d74e78f..a37a34f 100644
--- a/Android.bp
+++ b/Android.bp
@@ -232,6 +232,7 @@
         "payload_consumer/file_descriptor_utils.cc",
         "payload_consumer/file_writer.cc",
         "payload_consumer/filesystem_verifier_action.cc",
+        "payload_consumer/install_operation_executor.cc",
         "payload_consumer/install_plan.cc",
         "payload_consumer/mount_history.cc",
         "payload_consumer/payload_constants.cc",
@@ -805,6 +806,7 @@
         "payload_consumer/file_writer_unittest.cc",
         "payload_consumer/filesystem_verifier_action_unittest.cc",
         "payload_consumer/install_plan_unittest.cc",
+        "payload_consumer/install_operation_executor_unittest.cc",
         "payload_consumer/partition_update_generator_android_unittest.cc",
         "payload_consumer/postinstall_runner_action_unittest.cc",
         "payload_consumer/verity_writer_android_unittest.cc",
diff --git a/payload_consumer/install_operation_executor.cc b/payload_consumer/install_operation_executor.cc
new file mode 100644
index 0000000..3e477e5
--- /dev/null
+++ b/payload_consumer/install_operation_executor.cc
@@ -0,0 +1,316 @@
+//
+// Copyright (C) 2021 The Android Open Source Project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+
+#include "update_engine/payload_consumer/install_operation_executor.h"
+#include <memory>
+#include <utility>
+#include <vector>
+
+#include <fcntl.h>
+#include <glob.h>
+#include <linux/fs.h>
+
+#include <base/files/memory_mapped_file.h>
+#include <bsdiff/bspatch.h>
+#include <puffin/puffpatch.h>
+#include <sys/mman.h>
+
+#include "update_engine/common/utils.h"
+#include "update_engine/payload_consumer/bzip_extent_writer.h"
+#include "update_engine/payload_consumer/cached_file_descriptor.h"
+#include "update_engine/payload_consumer/extent_reader.h"
+#include "update_engine/payload_consumer/extent_writer.h"
+#include "update_engine/payload_consumer/file_descriptor.h"
+#include "update_engine/payload_consumer/file_descriptor_utils.h"
+#include "update_engine/payload_consumer/xz_extent_writer.h"
+#include "update_engine/payload_generator/delta_diff_generator.h"
+#include "update_engine/update_metadata.pb.h"
+
+namespace chromeos_update_engine {
+
+class BsdiffExtentFile : public bsdiff::FileInterface {
+ public:
+  BsdiffExtentFile(std::unique_ptr<ExtentReader> reader, size_t size)
+      : BsdiffExtentFile(std::move(reader), nullptr, size) {}
+  BsdiffExtentFile(std::unique_ptr<ExtentWriter> writer, size_t size)
+      : BsdiffExtentFile(nullptr, std::move(writer), size) {}
+
+  ~BsdiffExtentFile() override = default;
+
+  bool Read(void* buf, size_t count, size_t* bytes_read) override {
+    TEST_AND_RETURN_FALSE(reader_->Read(buf, count));
+    *bytes_read = count;
+    offset_ += count;
+    return true;
+  }
+
+  bool Write(const void* buf, size_t count, size_t* bytes_written) override {
+    TEST_AND_RETURN_FALSE(writer_->Write(buf, count));
+    *bytes_written = count;
+    offset_ += count;
+    return true;
+  }
+
+  bool Seek(off_t pos) override {
+    if (reader_ != nullptr) {
+      TEST_AND_RETURN_FALSE(reader_->Seek(pos));
+      offset_ = pos;
+    } else {
+      // For writes technically there should be no change of position, or it
+      // should be equivalent of current offset.
+      TEST_AND_RETURN_FALSE(offset_ == static_cast<uint64_t>(pos));
+    }
+    return true;
+  }
+
+  bool Close() override { return true; }
+
+  bool GetSize(uint64_t* size) override {
+    *size = size_;
+    return true;
+  }
+
+ private:
+  BsdiffExtentFile(std::unique_ptr<ExtentReader> reader,
+                   std::unique_ptr<ExtentWriter> writer,
+                   size_t size)
+      : reader_(std::move(reader)),
+        writer_(std::move(writer)),
+        size_(size),
+        offset_(0) {}
+
+  std::unique_ptr<ExtentReader> reader_;
+  std::unique_ptr<ExtentWriter> writer_;
+  uint64_t size_;
+  uint64_t offset_;
+
+  DISALLOW_COPY_AND_ASSIGN(BsdiffExtentFile);
+};
+// A class to be passed to |puffpatch| for reading from |source_fd_| and writing
+// into |target_fd_|.
+class PuffinExtentStream : public puffin::StreamInterface {
+ public:
+  // Constructor for creating a stream for reading from an |ExtentReader|.
+  PuffinExtentStream(std::unique_ptr<ExtentReader> reader, uint64_t size)
+      : PuffinExtentStream(std::move(reader), nullptr, size) {}
+
+  // Constructor for creating a stream for writing to an |ExtentWriter|.
+  PuffinExtentStream(std::unique_ptr<ExtentWriter> writer, uint64_t size)
+      : PuffinExtentStream(nullptr, std::move(writer), size) {}
+
+  ~PuffinExtentStream() override = default;
+
+  bool GetSize(uint64_t* size) const override {
+    *size = size_;
+    return true;
+  }
+
+  bool GetOffset(uint64_t* offset) const override {
+    *offset = offset_;
+    return true;
+  }
+
+  bool Seek(uint64_t offset) override {
+    if (is_read_) {
+      TEST_AND_RETURN_FALSE(reader_->Seek(offset));
+      offset_ = offset;
+    } else {
+      // For writes technically there should be no change of position, or it
+      // should equivalent of current offset.
+      TEST_AND_RETURN_FALSE(offset_ == offset);
+    }
+    return true;
+  }
+
+  bool Read(void* buffer, size_t count) override {
+    TEST_AND_RETURN_FALSE(is_read_);
+    TEST_AND_RETURN_FALSE(reader_->Read(buffer, count));
+    offset_ += count;
+    return true;
+  }
+
+  bool Write(const void* buffer, size_t count) override {
+    TEST_AND_RETURN_FALSE(!is_read_);
+    TEST_AND_RETURN_FALSE(writer_->Write(buffer, count));
+    offset_ += count;
+    return true;
+  }
+
+  bool Close() override { return true; }
+
+ private:
+  PuffinExtentStream(std::unique_ptr<ExtentReader> reader,
+                     std::unique_ptr<ExtentWriter> writer,
+                     uint64_t size)
+      : reader_(std::move(reader)),
+        writer_(std::move(writer)),
+        size_(size),
+        offset_(0),
+        is_read_(reader_ ? true : false) {}
+
+  std::unique_ptr<ExtentReader> reader_;
+  std::unique_ptr<ExtentWriter> writer_;
+  uint64_t size_;
+  uint64_t offset_;
+  bool is_read_;
+
+  DISALLOW_COPY_AND_ASSIGN(PuffinExtentStream);
+};
+
+bool InstallOperationExecutor::ExecuteInstallOp(
+    const InstallOperation& op,
+    std::unique_ptr<ExtentWriter> writer,
+    FileDescriptorPtr source_fd,
+    const void* data,
+    size_t size) {
+  switch (op.type()) {
+    case InstallOperation::REPLACE:
+    case InstallOperation::REPLACE_BZ:
+    case InstallOperation::REPLACE_XZ:
+      return ExecuteReplaceOperation(op, std::move(writer), data, size);
+    case InstallOperation::ZERO:
+    case InstallOperation::DISCARD:
+      return ExecuteZeroOrDiscardOperation(op, writer.get());
+    case InstallOperation::SOURCE_COPY:
+      return ExecuteSourceCopyOperation(op, writer.get(), source_fd);
+    case InstallOperation::SOURCE_BSDIFF:
+    case InstallOperation::BROTLI_BSDIFF:
+      return ExecuteSourceBsdiffOperation(
+          op, std::move(writer), source_fd, data, size);
+    case InstallOperation::PUFFDIFF:
+      return ExecutePuffDiffOperation(
+          op, std::move(writer), source_fd, data, size);
+      break;
+    default:
+      return false;
+  }
+  return false;
+}
+
+bool InstallOperationExecutor::ExecuteReplaceOperation(
+    const InstallOperation& operation,
+    std::unique_ptr<ExtentWriter> writer,
+    const void* data,
+    size_t count) {
+  TEST_AND_RETURN_FALSE(operation.type() == InstallOperation::REPLACE ||
+                        operation.type() == InstallOperation::REPLACE_BZ ||
+                        operation.type() == InstallOperation::REPLACE_XZ);
+  // Setup the ExtentWriter stack based on the operation type.
+  if (operation.type() == InstallOperation::REPLACE_BZ) {
+    writer.reset(new BzipExtentWriter(std::move(writer)));
+  } else if (operation.type() == InstallOperation::REPLACE_XZ) {
+    writer.reset(new XzExtentWriter(std::move(writer)));
+  }
+  TEST_AND_RETURN_FALSE(writer->Init(operation.dst_extents(), block_size_));
+  TEST_AND_RETURN_FALSE(writer->Write(data, operation.data_length()));
+
+  return true;
+}
+
+bool InstallOperationExecutor::ExecuteZeroOrDiscardOperation(
+    const InstallOperation& operation, ExtentWriter* writer) {
+  TEST_AND_RETURN_FALSE(operation.type() == InstallOperation::ZERO ||
+                        operation.type() == InstallOperation::DISCARD);
+  using base::MemoryMappedFile;
+  using Access = base::MemoryMappedFile::Access;
+  using Region = base::MemoryMappedFile::Region;
+  writer->Init(operation.dst_extents(), block_size_);
+  for (const auto& extent : operation.dst_extents()) {
+    // Mmap a region of /dev/zero, as we don't need any actual memory to store
+    // these 0s, so mmap a region of "free memory".
+    base::File dev_zero(base::FilePath("/dev/zero"),
+                        base::File::FLAG_OPEN | base::File::FLAG_READ);
+    MemoryMappedFile buffer;
+    TEST_AND_RETURN_FALSE_ERRNO(buffer.Initialize(
+        std::move(dev_zero),
+        Region{0, static_cast<size_t>(extent.num_blocks() * block_size_)},
+        Access::READ_ONLY));
+    writer->Write(buffer.data(), buffer.length());
+  }
+  return true;
+}
+
+bool InstallOperationExecutor::ExecuteSourceCopyOperation(
+    const InstallOperation& operation,
+    ExtentWriter* writer,
+    FileDescriptorPtr source_fd) {
+  TEST_AND_RETURN_FALSE(operation.type() == InstallOperation::SOURCE_COPY);
+  writer->Init(operation.dst_extents(), block_size_);
+  return fd_utils::CommonHashExtents(
+      source_fd, operation.src_extents(), writer, block_size_, nullptr);
+}
+
+bool InstallOperationExecutor::ExecuteSourceBsdiffOperation(
+    const InstallOperation& operation,
+    std::unique_ptr<ExtentWriter> writer,
+    FileDescriptorPtr source_fd,
+    const void* data,
+    size_t count) {
+  TEST_AND_RETURN_FALSE(operation.type() == InstallOperation::SOURCE_BSDIFF ||
+                        operation.type() == InstallOperation::BROTLI_BSDIFF ||
+                        operation.type() == InstallOperation::BSDIFF);
+  TEST_AND_RETURN_FALSE(source_fd != nullptr);
+
+  auto reader = std::make_unique<DirectExtentReader>();
+  TEST_AND_RETURN_FALSE(
+      reader->Init(source_fd, operation.src_extents(), block_size_));
+  auto src_file = std::make_unique<BsdiffExtentFile>(
+      std::move(reader),
+      utils::BlocksInExtents(operation.src_extents()) * block_size_);
+
+  TEST_AND_RETURN_FALSE(writer->Init(operation.dst_extents(), block_size_));
+  auto dst_file = std::make_unique<BsdiffExtentFile>(
+      std::move(writer),
+      utils::BlocksInExtents(operation.dst_extents()) * block_size_);
+
+  TEST_AND_RETURN_FALSE(bsdiff::bspatch(std::move(src_file),
+                                        std::move(dst_file),
+                                        reinterpret_cast<const uint8_t*>(data),
+                                        count) == 0);
+  return true;
+}
+
+bool InstallOperationExecutor::ExecutePuffDiffOperation(
+    const InstallOperation& operation,
+    std::unique_ptr<ExtentWriter> writer,
+    FileDescriptorPtr source_fd,
+    const void* data,
+    size_t count) {
+  TEST_AND_RETURN_FALSE(operation.type() == InstallOperation::PUFFDIFF);
+  TEST_AND_RETURN_FALSE(source_fd != nullptr);
+
+  auto reader = std::make_unique<DirectExtentReader>();
+  TEST_AND_RETURN_FALSE(
+      reader->Init(source_fd, operation.src_extents(), block_size_));
+  puffin::UniqueStreamPtr src_stream(new PuffinExtentStream(
+      std::move(reader),
+      utils::BlocksInExtents(operation.src_extents()) * block_size_));
+
+  TEST_AND_RETURN_FALSE(writer->Init(operation.dst_extents(), block_size_));
+  puffin::UniqueStreamPtr dst_stream(new PuffinExtentStream(
+      std::move(writer),
+      utils::BlocksInExtents(operation.dst_extents()) * block_size_));
+
+  constexpr size_t kMaxCacheSize = 5 * 1024 * 1024;  // Total 5MB cache.
+  TEST_AND_RETURN_FALSE(
+      puffin::PuffPatch(std::move(src_stream),
+                        std::move(dst_stream),
+                        reinterpret_cast<const uint8_t*>(data),
+                        count,
+                        kMaxCacheSize));
+  return true;
+}
+}  // namespace chromeos_update_engine
diff --git a/payload_consumer/install_operation_executor.h b/payload_consumer/install_operation_executor.h
new file mode 100644
index 0000000..a43139b
--- /dev/null
+++ b/payload_consumer/install_operation_executor.h
@@ -0,0 +1,64 @@
+//
+// Copyright (C) 2021 The Android Open Source Project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+
+#ifndef UPDATE_ENGINE_INSTALL_OPERATION_EXECUTOR_H
+#define UPDATE_ENGINE_INSTALL_OPERATION_EXECUTOR_H
+
+#include <memory>
+
+#include "update_engine/payload_consumer/extent_writer.h"
+#include "update_engine/payload_consumer/file_descriptor.h"
+#include "update_engine/update_metadata.pb.h"
+
+namespace chromeos_update_engine {
+
+class InstallOperationExecutor {
+ public:
+  explicit InstallOperationExecutor(size_t block_size)
+      : block_size_(block_size) {}
+
+  bool ExecuteInstallOp(const InstallOperation& op,
+                        std::unique_ptr<ExtentWriter> writer,
+                        FileDescriptorPtr source_fd,
+                        const void* data,
+                        size_t size);
+  bool ExecuteReplaceOperation(const InstallOperation& operation,
+                               std::unique_ptr<ExtentWriter> writer,
+                               const void* data,
+                               size_t count);
+  bool ExecuteZeroOrDiscardOperation(const InstallOperation& operation,
+                                     ExtentWriter* writer);
+  bool ExecuteSourceCopyOperation(const InstallOperation& operation,
+                                  ExtentWriter* writer,
+                                  FileDescriptorPtr source_fd);
+  bool ExecuteSourceBsdiffOperation(const InstallOperation& operation,
+                                    std::unique_ptr<ExtentWriter> writer,
+                                    FileDescriptorPtr source_fd,
+                                    const void* data,
+                                    size_t count);
+  bool ExecutePuffDiffOperation(const InstallOperation& operation,
+                                std::unique_ptr<ExtentWriter> writer,
+                                FileDescriptorPtr source_fd,
+                                const void* data,
+                                size_t count);
+
+ private:
+  size_t block_size_;
+};
+
+}  // namespace chromeos_update_engine
+
+#endif
diff --git a/payload_consumer/install_operation_executor_unittest.cc b/payload_consumer/install_operation_executor_unittest.cc
new file mode 100644
index 0000000..c90b28d
--- /dev/null
+++ b/payload_consumer/install_operation_executor_unittest.cc
@@ -0,0 +1,221 @@
+//
+// Copyright (C) 2021 The Android Open Source Project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+
+#include "update_engine/payload_consumer/install_operation_executor.h"
+
+#include <fcntl.h>
+#include <unistd.h>
+
+#include <algorithm>
+#include <array>
+#include <cstring>
+#include <limits>
+#include <memory>
+#include <ostream>
+#include <utility>
+#include <vector>
+
+#include <brillo/secure_blob.h>
+#include <gtest/gtest.h>
+#include <update_engine/update_metadata.pb.h>
+
+#include "update_engine/common/utils.h"
+#include "update_engine/payload_consumer/extent_writer.h"
+#include "update_engine/payload_consumer/file_descriptor.h"
+#include "update_engine/payload_consumer/payload_constants.h"
+#include "update_engine/payload_generator/extent_ranges.h"
+#include "update_engine/payload_generator/extent_utils.h"
+
+namespace chromeos_update_engine {
+
+std::ostream& operator<<(std::ostream& out,
+                         const chromeos_update_engine::InstallOperation& op) {
+  out << InstallOperationTypeName(op.type())
+      << " SRC: " << ExtentsToString(op.src_extents())
+      << " DST: " << ExtentsToString(op.dst_extents());
+  return out;
+}
+
+namespace {
+template <typename Container>
+size_t GetNthBlock(const Container& extents, const size_t n) {
+  size_t cur_block_count = 0;
+  for (const auto& extent : extents) {
+    if (cur_block_count + extent.num_blocks() >= n) {
+      return extent.start_block() + (n - cur_block_count);
+    }
+    cur_block_count += extent.num_blocks();
+  }
+  return std::numeric_limits<size_t>::max();
+}
+
+}  // namespace
+
+class InstallOperationExecutorTest : public ::testing::Test {
+ public:
+  static constexpr size_t NUM_BLOCKS = 10;
+  static constexpr size_t BLOCK_SIZE = 4096;
+  void SetUp() override {
+    // Fill source partition with arbitrary data.
+    std::array<uint8_t, BLOCK_SIZE> buffer{};
+    for (size_t i = 0; i < NUM_BLOCKS; i++) {
+      // Fill block with arbitrary data. We don't care about what data is being
+      // written to source partition, so as long as each block is slightly
+      // different.
+      std::fill(buffer.begin(), buffer.end(), i);
+      ASSERT_TRUE(utils::WriteAll(source_.fd(), buffer.data(), buffer.size()))
+          << "Failed to write to source partition file: " << strerror(errno);
+      std::fill(buffer.begin(), buffer.end(), NUM_BLOCKS + i);
+      ASSERT_TRUE(utils::WriteAll(target_.fd(), buffer.data(), buffer.size()))
+          << "Failed to write to target partition file: " << strerror(errno);
+    }
+    fsync(source_.fd());
+    fsync(target_.fd());
+
+    // set target partition to have same size as source partition.
+    // update_engine mostly assumes that target partition have the desired
+    // size, so we mock that.
+    ASSERT_GE(ftruncate64(target_.fd(), NUM_BLOCKS * BLOCK_SIZE), 0)
+        << strerror(errno) << " failed to set target partition size to "
+        << NUM_BLOCKS * BLOCK_SIZE;
+
+    source_fd_->Open(source_.path().c_str(), O_RDONLY);
+    target_fd_->Open(target_.path().c_str(), O_RDWR);
+  }
+
+  void VerityUntouchedExtents(const InstallOperation& op) {
+    ExtentRanges extent_set;
+    extent_set.AddExtent(ExtentForRange(0, 10));
+    extent_set.SubtractRepeatedExtents(op.dst_extents());
+    std::vector<Extent> untouched_extents{extent_set.extent_set().begin(),
+                                          extent_set.extent_set().end()};
+    brillo::Blob actual_data;
+    ASSERT_TRUE(utils::ReadExtents(target_.path(),
+                                   untouched_extents,
+                                   &actual_data,
+                                   extent_set.blocks() * BLOCK_SIZE,
+                                   BLOCK_SIZE));
+    const auto untouched_blocks = ExpandExtents(untouched_extents);
+    for (size_t i = 0; i < actual_data.size(); i++) {
+      const auto block_offset = i / BLOCK_SIZE;
+      const auto offset = i % BLOCK_SIZE;
+      ASSERT_EQ(
+          actual_data[i],
+          static_cast<uint8_t>(NUM_BLOCKS + untouched_blocks[block_offset]))
+          << "After performing op " << op << ", offset " << offset
+          << " in block " << GetNthBlock(untouched_extents, block_offset)
+          << " is modified but it shouldn't.";
+    }
+  }
+  ScopedTempFile source_{"source_partition.XXXXXXXX", true};
+  ScopedTempFile target_{"target_partition.XXXXXXXX", true};
+  FileDescriptorPtr source_fd_ = std::make_shared<EintrSafeFileDescriptor>();
+  FileDescriptorPtr target_fd_ = std::make_shared<EintrSafeFileDescriptor>();
+  InstallOperationExecutor executor_{BLOCK_SIZE};
+};
+
+TEST_F(InstallOperationExecutorTest, ReplaceOpTest) {
+  InstallOperation op;
+  op.set_type(InstallOperation::REPLACE);
+  *op.mutable_dst_extents()->Add() = ExtentForRange(2, 2);
+  *op.mutable_dst_extents()->Add() = ExtentForRange(6, 2);
+  op.set_data_length(BLOCK_SIZE * 4);
+  brillo::Blob expected_data;
+  expected_data.resize(BLOCK_SIZE * 4);
+  // Fill buffer with arbitrary data. Doesn't matter what it is. Each block
+  // needs to be different so that we can ensure the InstallOperationExecutor
+  // is reading data from the correct offset.
+  for (int i = 0; i < 4; i++) {
+    std::fill(&expected_data[i * BLOCK_SIZE],
+              &expected_data[(i + 1) * BLOCK_SIZE],
+              i + 99);
+  }
+  auto writer = std::make_unique<DirectExtentWriter>(target_fd_);
+  ASSERT_TRUE(executor_.ExecuteReplaceOperation(
+      op, std::move(writer), expected_data.data(), expected_data.size()));
+
+  brillo::Blob actual_data;
+  utils::ReadExtents(
+      target_.path(),
+      std::vector<Extent>{op.dst_extents().begin(), op.dst_extents().end()},
+      &actual_data,
+      BLOCK_SIZE * 4,
+      BLOCK_SIZE);
+  ASSERT_EQ(actual_data, expected_data);
+  VerityUntouchedExtents(op);
+}
+
+TEST_F(InstallOperationExecutorTest, ZeroOrDiscardeOpTest) {
+  InstallOperation op;
+  op.set_type(InstallOperation::ZERO);
+  *op.mutable_dst_extents()->Add() = ExtentForRange(2, 2);
+  *op.mutable_dst_extents()->Add() = ExtentForRange(6, 2);
+  auto writer = std::make_unique<DirectExtentWriter>(target_fd_);
+  ASSERT_TRUE(executor_.ExecuteZeroOrDiscardOperation(op, writer.get()));
+  brillo::Blob actual_data;
+  utils::ReadExtents(
+      target_.path(),
+      std::vector<Extent>{op.dst_extents().begin(), op.dst_extents().end()},
+      &actual_data,
+      BLOCK_SIZE * 4,
+      BLOCK_SIZE);
+  for (size_t i = 0; i < actual_data.size(); i++) {
+    ASSERT_EQ(actual_data[i], 0U) << "position " << i << " isn't zeroed!";
+  }
+  VerityUntouchedExtents(op);
+}
+
+TEST_F(InstallOperationExecutorTest, SourceCopyOpTest) {
+  InstallOperation op;
+  op.set_type(InstallOperation::SOURCE_COPY);
+  *op.mutable_src_extents()->Add() = ExtentForRange(1, 2);
+  *op.mutable_src_extents()->Add() = ExtentForRange(5, 1);
+  *op.mutable_src_extents()->Add() = ExtentForRange(7, 1);
+
+  *op.mutable_dst_extents()->Add() = ExtentForRange(2, 2);
+  *op.mutable_dst_extents()->Add() = ExtentForRange(6, 2);
+
+  auto writer = std::make_unique<DirectExtentWriter>(target_fd_);
+  ASSERT_TRUE(
+      executor_.ExecuteSourceCopyOperation(op, writer.get(), source_fd_));
+  brillo::Blob actual_data;
+  utils::ReadExtents(
+      target_.path(),
+      std::vector<Extent>{op.dst_extents().begin(), op.dst_extents().end()},
+      &actual_data,
+      BLOCK_SIZE * 4,
+      BLOCK_SIZE);
+  brillo::Blob expected_data;
+  utils::ReadExtents(
+      source_.path(),
+      std::vector<Extent>{op.src_extents().begin(), op.src_extents().end()},
+      &expected_data,
+      BLOCK_SIZE * 4,
+      BLOCK_SIZE);
+
+  ASSERT_EQ(expected_data.size(), actual_data.size());
+  for (size_t i = 0; i < actual_data.size(); i++) {
+    const auto block_offset = i / BLOCK_SIZE;
+    const auto offset = i % BLOCK_SIZE;
+    ASSERT_EQ(actual_data[i], expected_data[i])
+        << "After performing op " << op << ", offset " << offset << " in  ["
+        << GetNthBlock(op.src_extents(), block_offset) << " -> "
+        << GetNthBlock(op.dst_extents(), block_offset) << "]"
+        << " is not copied correctly";
+  }
+  VerityUntouchedExtents(op);
+}
+}  // namespace chromeos_update_engine
diff --git a/payload_consumer/partition_writer.cc b/payload_consumer/partition_writer.cc
index c58a802..4df0af6 100644
--- a/payload_consumer/partition_writer.cc
+++ b/payload_consumer/partition_writer.cc
@@ -25,12 +25,7 @@
 #include <utility>
 #include <vector>
 
-#include <base/files/memory_mapped_file.h>
 #include <base/strings/string_number_conversions.h>
-#include <bsdiff/bspatch.h>
-#include <puffin/puffpatch.h>
-#include <bsdiff/file_interface.h>
-#include <puffin/stream.h>
 
 #include "update_engine/common/terminator.h"
 #include "update_engine/common/utils.h"
@@ -113,135 +108,6 @@
   return fd;
 }
 
-class BsdiffExtentFile : public bsdiff::FileInterface {
- public:
-  BsdiffExtentFile(std::unique_ptr<ExtentReader> reader, size_t size)
-      : BsdiffExtentFile(std::move(reader), nullptr, size) {}
-  BsdiffExtentFile(std::unique_ptr<ExtentWriter> writer, size_t size)
-      : BsdiffExtentFile(nullptr, std::move(writer), size) {}
-
-  ~BsdiffExtentFile() override = default;
-
-  bool Read(void* buf, size_t count, size_t* bytes_read) override {
-    TEST_AND_RETURN_FALSE(reader_->Read(buf, count));
-    *bytes_read = count;
-    offset_ += count;
-    return true;
-  }
-
-  bool Write(const void* buf, size_t count, size_t* bytes_written) override {
-    TEST_AND_RETURN_FALSE(writer_->Write(buf, count));
-    *bytes_written = count;
-    offset_ += count;
-    return true;
-  }
-
-  bool Seek(off_t pos) override {
-    if (reader_ != nullptr) {
-      TEST_AND_RETURN_FALSE(reader_->Seek(pos));
-      offset_ = pos;
-    } else {
-      // For writes technically there should be no change of position, or it
-      // should be equivalent of current offset.
-      TEST_AND_RETURN_FALSE(offset_ == static_cast<uint64_t>(pos));
-    }
-    return true;
-  }
-
-  bool Close() override { return true; }
-
-  bool GetSize(uint64_t* size) override {
-    *size = size_;
-    return true;
-  }
-
- private:
-  BsdiffExtentFile(std::unique_ptr<ExtentReader> reader,
-                   std::unique_ptr<ExtentWriter> writer,
-                   size_t size)
-      : reader_(std::move(reader)),
-        writer_(std::move(writer)),
-        size_(size),
-        offset_(0) {}
-
-  std::unique_ptr<ExtentReader> reader_;
-  std::unique_ptr<ExtentWriter> writer_;
-  uint64_t size_;
-  uint64_t offset_;
-
-  DISALLOW_COPY_AND_ASSIGN(BsdiffExtentFile);
-};
-// A class to be passed to |puffpatch| for reading from |source_fd_| and writing
-// into |target_fd_|.
-class PuffinExtentStream : public puffin::StreamInterface {
- public:
-  // Constructor for creating a stream for reading from an |ExtentReader|.
-  PuffinExtentStream(std::unique_ptr<ExtentReader> reader, uint64_t size)
-      : PuffinExtentStream(std::move(reader), nullptr, size) {}
-
-  // Constructor for creating a stream for writing to an |ExtentWriter|.
-  PuffinExtentStream(std::unique_ptr<ExtentWriter> writer, uint64_t size)
-      : PuffinExtentStream(nullptr, std::move(writer), size) {}
-
-  ~PuffinExtentStream() override = default;
-
-  bool GetSize(uint64_t* size) const override {
-    *size = size_;
-    return true;
-  }
-
-  bool GetOffset(uint64_t* offset) const override {
-    *offset = offset_;
-    return true;
-  }
-
-  bool Seek(uint64_t offset) override {
-    if (is_read_) {
-      TEST_AND_RETURN_FALSE(reader_->Seek(offset));
-      offset_ = offset;
-    } else {
-      // For writes technically there should be no change of position, or it
-      // should equivalent of current offset.
-      TEST_AND_RETURN_FALSE(offset_ == offset);
-    }
-    return true;
-  }
-
-  bool Read(void* buffer, size_t count) override {
-    TEST_AND_RETURN_FALSE(is_read_);
-    TEST_AND_RETURN_FALSE(reader_->Read(buffer, count));
-    offset_ += count;
-    return true;
-  }
-
-  bool Write(const void* buffer, size_t count) override {
-    TEST_AND_RETURN_FALSE(!is_read_);
-    TEST_AND_RETURN_FALSE(writer_->Write(buffer, count));
-    offset_ += count;
-    return true;
-  }
-
-  bool Close() override { return true; }
-
- private:
-  PuffinExtentStream(std::unique_ptr<ExtentReader> reader,
-                     std::unique_ptr<ExtentWriter> writer,
-                     uint64_t size)
-      : reader_(std::move(reader)),
-        writer_(std::move(writer)),
-        size_(size),
-        offset_(0),
-        is_read_(reader_ ? true : false) {}
-
-  std::unique_ptr<ExtentReader> reader_;
-  std::unique_ptr<ExtentWriter> writer_;
-  uint64_t size_;
-  uint64_t offset_;
-  bool is_read_;
-
-  DISALLOW_COPY_AND_ASSIGN(PuffinExtentStream);
-};
-
 PartitionWriter::PartitionWriter(
     const PartitionUpdate& partition_update,
     const InstallPlan::Partition& install_part,
@@ -320,26 +186,6 @@
   return true;
 }
 
-bool InstallOperationExecutor::ExecuteReplaceOperation(
-    const InstallOperation& operation,
-    std::unique_ptr<ExtentWriter> writer,
-    const void* data,
-    size_t count) {
-  TEST_AND_RETURN_FALSE(operation.type() == InstallOperation::REPLACE ||
-                        operation.type() == InstallOperation::REPLACE_BZ ||
-                        operation.type() == InstallOperation::REPLACE_XZ);
-  // Setup the ExtentWriter stack based on the operation type.
-  if (operation.type() == InstallOperation::REPLACE_BZ) {
-    writer.reset(new BzipExtentWriter(std::move(writer)));
-  } else if (operation.type() == InstallOperation::REPLACE_XZ) {
-    writer.reset(new XzExtentWriter(std::move(writer)));
-  }
-  TEST_AND_RETURN_FALSE(writer->Init(operation.dst_extents(), block_size_));
-  TEST_AND_RETURN_FALSE(writer->Write(data, operation.data_length()));
-
-  return true;
-}
-
 bool PartitionWriter::PerformReplaceOperation(const InstallOperation& operation,
                                               const void* data,
                                               size_t count) {
@@ -350,27 +196,6 @@
       operation, std::move(writer), data, count);
 }
 
-bool InstallOperationExecutor::ExecuteZeroOrDiscardOperation(
-    const InstallOperation& operation, ExtentWriter* writer) {
-  TEST_AND_RETURN_FALSE(operation.type() == InstallOperation::ZERO ||
-                        operation.type() == InstallOperation::DISCARD);
-  for (const auto& extent : operation.dst_extents()) {
-    // Mmap a region of /dev/zero, as we don't need any actual memory to store
-    // these 0s, so mmap a region of "free memory".
-    base::File dev_zero(base::FilePath("/dev/zero"),
-                        base::File::FLAG_OPEN | base::File::FLAG_READ);
-    base::MemoryMappedFile buffer;
-    TEST_AND_RETURN_FALSE_ERRNO(buffer.Initialize(
-        std::move(dev_zero),
-        base::MemoryMappedFile::Region{
-            0, static_cast<size_t>(extent.num_blocks() * block_size_)},
-        base::MemoryMappedFile::Access::READ_ONLY));
-    TEST_AND_RETURN_FALSE(buffer.data() != nullptr);
-    writer->Write(buffer.data(), buffer.length());
-  }
-  return true;
-}
-
 bool PartitionWriter::PerformZeroOrDiscardOperation(
     const InstallOperation& operation) {
 #ifdef BLKZEROOUT
@@ -417,15 +242,6 @@
   return out;
 }
 
-bool InstallOperationExecutor::ExecuteSourceCopyOperation(
-    const InstallOperation& operation,
-    ExtentWriter* writer,
-    FileDescriptorPtr source_fd) {
-  TEST_AND_RETURN_FALSE(operation.type() == InstallOperation::SOURCE_COPY);
-  return fd_utils::CommonHashExtents(
-      source_fd, operation.src_extents(), writer, block_size_, nullptr);
-}
-
 bool PartitionWriter::PerformSourceCopyOperation(
     const InstallOperation& operation, ErrorCode* error) {
   TEST_AND_RETURN_FALSE(source_fd_ != nullptr);
@@ -457,36 +273,6 @@
       optimized, writer.get(), source_fd);
 }
 
-bool InstallOperationExecutor::ExecuteSourceBsdiffOperation(
-    const InstallOperation& operation,
-    std::unique_ptr<ExtentWriter> writer,
-    FileDescriptorPtr source_fd,
-    const void* data,
-    size_t count) {
-  TEST_AND_RETURN_FALSE(operation.type() == InstallOperation::SOURCE_BSDIFF ||
-                        operation.type() == InstallOperation::BROTLI_BSDIFF ||
-                        operation.type() == InstallOperation::BSDIFF);
-  TEST_AND_RETURN_FALSE(source_fd != nullptr);
-
-  auto reader = std::make_unique<DirectExtentReader>();
-  TEST_AND_RETURN_FALSE(
-      reader->Init(source_fd, operation.src_extents(), block_size_));
-  auto src_file = std::make_unique<BsdiffExtentFile>(
-      std::move(reader),
-      utils::BlocksInExtents(operation.src_extents()) * block_size_);
-
-  TEST_AND_RETURN_FALSE(writer->Init(operation.dst_extents(), block_size_));
-  auto dst_file = std::make_unique<BsdiffExtentFile>(
-      std::move(writer),
-      utils::BlocksInExtents(operation.dst_extents()) * block_size_);
-
-  TEST_AND_RETURN_FALSE(bsdiff::bspatch(std::move(src_file),
-                                        std::move(dst_file),
-                                        reinterpret_cast<const uint8_t*>(data),
-                                        count) == 0);
-  return true;
-}
-
 bool PartitionWriter::PerformSourceBsdiffOperation(
     const InstallOperation& operation,
     ErrorCode* error,
@@ -501,37 +287,6 @@
       operation, std::move(writer), source_fd, data, count);
 }
 
-bool InstallOperationExecutor::ExecutePuffDiffOperation(
-    const InstallOperation& operation,
-    std::unique_ptr<ExtentWriter> writer,
-    FileDescriptorPtr source_fd,
-    const void* data,
-    size_t count) {
-  TEST_AND_RETURN_FALSE(operation.type() == InstallOperation::PUFFDIFF);
-  TEST_AND_RETURN_FALSE(source_fd != nullptr);
-
-  auto reader = std::make_unique<DirectExtentReader>();
-  TEST_AND_RETURN_FALSE(
-      reader->Init(source_fd, operation.src_extents(), block_size_));
-  puffin::UniqueStreamPtr src_stream(new PuffinExtentStream(
-      std::move(reader),
-      utils::BlocksInExtents(operation.src_extents()) * block_size_));
-
-  TEST_AND_RETURN_FALSE(writer->Init(operation.dst_extents(), block_size_));
-  puffin::UniqueStreamPtr dst_stream(new PuffinExtentStream(
-      std::move(writer),
-      utils::BlocksInExtents(operation.dst_extents()) * block_size_));
-
-  constexpr size_t kMaxCacheSize = 5 * 1024 * 1024;  // Total 5MB cache.
-  TEST_AND_RETURN_FALSE(
-      puffin::PuffPatch(std::move(src_stream),
-                        std::move(dst_stream),
-                        reinterpret_cast<const uint8_t*>(data),
-                        count,
-                        kMaxCacheSize));
-  return true;
-}
-
 bool PartitionWriter::PerformPuffDiffOperation(
     const InstallOperation& operation,
     ErrorCode* error,
@@ -670,34 +425,4 @@
   return std::make_unique<DirectExtentWriter>(target_fd_);
 }
 
-bool InstallOperationExecutor::ExecuteInstallOp(
-    const InstallOperation& op,
-    std::unique_ptr<ExtentWriter> writer,
-    FileDescriptorPtr source_fd,
-    const void* data,
-    size_t size) {
-  switch (op.type()) {
-    case InstallOperation::REPLACE:
-    case InstallOperation::REPLACE_BZ:
-    case InstallOperation::REPLACE_XZ:
-      return ExecuteReplaceOperation(op, std::move(writer), data, size);
-    case InstallOperation::ZERO:
-    case InstallOperation::DISCARD:
-      return ExecuteZeroOrDiscardOperation(op, writer.get());
-    case InstallOperation::SOURCE_COPY:
-      return ExecuteSourceCopyOperation(op, writer.get(), source_fd);
-    case InstallOperation::SOURCE_BSDIFF:
-    case InstallOperation::BROTLI_BSDIFF:
-      return ExecuteSourceBsdiffOperation(
-          op, std::move(writer), source_fd, data, size);
-    case InstallOperation::PUFFDIFF:
-      return ExecutePuffDiffOperation(
-          op, std::move(writer), source_fd, data, size);
-      break;
-    default:
-      return false;
-  }
-  return false;
-}
-
 }  // namespace chromeos_update_engine
diff --git a/payload_consumer/partition_writer.h b/payload_consumer/partition_writer.h
index dee30e8..554e590 100644
--- a/payload_consumer/partition_writer.h
+++ b/payload_consumer/partition_writer.h
@@ -27,50 +27,12 @@
 #include "update_engine/common/dynamic_partition_control_interface.h"
 #include "update_engine/payload_consumer/extent_writer.h"
 #include "update_engine/payload_consumer/file_descriptor.h"
+#include "update_engine/payload_consumer/install_operation_executor.h"
 #include "update_engine/payload_consumer/install_plan.h"
 #include "update_engine/update_metadata.pb.h"
 
 namespace chromeos_update_engine {
 
-// A reference class for interpretation of different OTA ops.
-// Different partition writers(VABCPartitionWriter and the regular
-// PartitionWriter) will use this class via composition, and optionally optimize
-// for some specific operations. For example, in VABC, copy operations are
-// handled with special care, but other operations are defaulted to this class.
-class InstallOperationExecutor {
- public:
-  explicit InstallOperationExecutor(size_t block_size)
-      : block_size_(block_size) {}
-
-  bool ExecuteInstallOp(const InstallOperation& op,
-                        std::unique_ptr<ExtentWriter> writer,
-                        FileDescriptorPtr source_fd,
-                        const void* data,
-                        size_t size);
-  bool ExecuteReplaceOperation(const InstallOperation& operation,
-                               std::unique_ptr<ExtentWriter> writer,
-                               const void* data,
-                               size_t count);
-  bool ExecuteZeroOrDiscardOperation(const InstallOperation& operation,
-                                     ExtentWriter* writer);
-  bool ExecuteSourceCopyOperation(const InstallOperation& operation,
-                                  ExtentWriter* writer,
-                                  FileDescriptorPtr source_fd);
-  bool ExecuteSourceBsdiffOperation(const InstallOperation& operation,
-                                    std::unique_ptr<ExtentWriter> writer,
-                                    FileDescriptorPtr source_fd,
-                                    const void* data,
-                                    size_t count);
-  bool ExecutePuffDiffOperation(const InstallOperation& operation,
-                                std::unique_ptr<ExtentWriter> writer,
-                                FileDescriptorPtr source_fd,
-                                const void* data,
-                                size_t count);
-
- private:
-  size_t block_size_;
-};
-
 class PartitionWriter {
  public:
   PartitionWriter(const PartitionUpdate& partition_update,
diff --git a/payload_generator/extent_utils.cc b/payload_generator/extent_utils.cc
index 2efef12..f4a9ff0 100644
--- a/payload_generator/extent_utils.cc
+++ b/payload_generator/extent_utils.cc
@@ -87,7 +87,8 @@
   }
 }
 
-string ExtentsToString(const vector<Extent>& extents) {
+template <typename Container>
+string ExtentsToStringTemplate(const Container& extents) {
   string ext_str;
   for (const Extent& e : extents)
     ext_str += base::StringPrintf("[%" PRIu64 ", %" PRIu64 "] ",
@@ -96,6 +97,15 @@
   return ext_str;
 }
 
+std::string ExtentsToString(const std::vector<Extent>& extents) {
+  return ExtentsToStringTemplate(extents);
+}
+
+std::string ExtentsToString(
+    const google::protobuf::RepeatedPtrField<Extent>& extents) {
+  return ExtentsToStringTemplate(extents);
+}
+
 void NormalizeExtents(vector<Extent>* extents) {
   vector<Extent> new_extents;
   for (const Extent& curr_ext : *extents) {
diff --git a/payload_generator/extent_utils.h b/payload_generator/extent_utils.h
index 7aa614a..e9afa98 100644
--- a/payload_generator/extent_utils.h
+++ b/payload_generator/extent_utils.h
@@ -22,6 +22,7 @@
 
 #include <base/logging.h>
 
+#include "google/protobuf/repeated_field.h"
 #include "update_engine/payload_consumer/payload_constants.h"
 #include "update_engine/update_metadata.pb.h"
 
@@ -63,6 +64,8 @@
 
 // Returns a string representing all extents in |extents|.
 std::string ExtentsToString(const std::vector<Extent>& extents);
+std::string ExtentsToString(
+    const google::protobuf::RepeatedPtrField<Extent>& extents);
 
 // Takes a pointer to extents |extents| and extents |extents_to_add|, and
 // merges them by adding |extents_to_add| to |extents| and normalizing.