diff --git a/Android.bp b/Android.bp
index 27ba172..52019b6 100644
--- a/Android.bp
+++ b/Android.bp
@@ -203,6 +203,8 @@
         "libgsi",
         "libpayload_consumer",
         "libsnapshot",
+        "libsnapshot_cow",
+        "libz",
         "update_metadata-protos",
     ],
     shared_libs: [
diff --git a/common/dynamic_partition_control_interface.h b/common/dynamic_partition_control_interface.h
index 22f6db8..530b0af 100644
--- a/common/dynamic_partition_control_interface.h
+++ b/common/dynamic_partition_control_interface.h
@@ -26,8 +26,14 @@
 #include "update_engine/common/action.h"
 #include "update_engine/common/cleanup_previous_update_action_delegate.h"
 #include "update_engine/common/error_code.h"
+#include "update_engine/payload_consumer/file_descriptor.h"
 #include "update_engine/update_metadata.pb.h"
 
+// Forware declare for libsnapshot/snapshot_writer.h
+namespace android::snapshot {
+class ISnapshotWriter;
+}
+
 namespace chromeos_update_engine {
 
 struct FeatureFlag {
@@ -139,6 +145,15 @@
       uint32_t source_slot,
       uint32_t target_slot,
       const std::vector<std::string>& partitions) = 0;
+  // Partition name is expected to be unsuffixed. e.g. system, vendor
+  // Return an interface to write to a snapshoted partition.
+  // If `is_append` is false, then existing COW data will be overwritten.
+  // Otherwise the cow writer will be opened on APPEND mode, existing COW data
+  // is preserved.
+  virtual std::unique_ptr<android::snapshot::ISnapshotWriter> OpenCowWriter(
+      const std::string& unsuffixed_partition_name,
+      const std::optional<std::string>&,
+      bool is_append = false) = 0;
 };
 
 }  // namespace chromeos_update_engine
diff --git a/common/dynamic_partition_control_stub.cc b/common/dynamic_partition_control_stub.cc
index c63a8ff..64ab201 100644
--- a/common/dynamic_partition_control_stub.cc
+++ b/common/dynamic_partition_control_stub.cc
@@ -20,6 +20,7 @@
 #include <string>
 
 #include <base/logging.h>
+#include <libsnapshot/cow_writer.h>
 
 #include "update_engine/common/dynamic_partition_control_stub.h"
 
@@ -87,4 +88,12 @@
   return true;
 }
 
+std::unique_ptr<android::snapshot::ISnapshotWriter>
+DynamicPartitionControlStub::OpenCowWriter(
+    const std::string& /*unsuffixed_partition_name*/,
+    const std::optional<std::string>& /*source_path*/,
+    bool /*is_append*/) {
+  return nullptr;
+}
+
 }  // namespace chromeos_update_engine
diff --git a/common/dynamic_partition_control_stub.h b/common/dynamic_partition_control_stub.h
index 8bff474..a939cfb 100644
--- a/common/dynamic_partition_control_stub.h
+++ b/common/dynamic_partition_control_stub.h
@@ -57,8 +57,12 @@
       uint32_t source_slot,
       uint32_t target_slot,
       const std::vector<std::string>& partitions) override;
-};
 
+  std::unique_ptr<android::snapshot::ISnapshotWriter> OpenCowWriter(
+      const std::string& unsuffixed_partition_name,
+      const std::optional<std::string>&,
+      bool is_append) override;
+};
 }  // namespace chromeos_update_engine
 
 #endif  // UPDATE_ENGINE_COMMON_DYNAMIC_PARTITION_CONTROL_STUB_H_
diff --git a/dynamic_partition_control_android.cc b/dynamic_partition_control_android.cc
index 1fb2c25..06e5745 100644
--- a/dynamic_partition_control_android.cc
+++ b/dynamic_partition_control_android.cc
@@ -17,6 +17,7 @@
 #include "update_engine/dynamic_partition_control_android.h"
 
 #include <chrono>  // NOLINT(build/c++11) - using libsnapshot / liblp API
+#include <cstdint>
 #include <map>
 #include <memory>
 #include <set>
@@ -36,6 +37,8 @@
 #include <fs_mgr_overlayfs.h>
 #include <libavb/libavb.h>
 #include <libdm/dm.h>
+#include <liblp/liblp.h>
+#include <libsnapshot/cow_writer.h>
 #include <libsnapshot/snapshot.h>
 #include <libsnapshot/snapshot_stub.h>
 
@@ -1230,4 +1233,30 @@
   return metadata_device_ != nullptr;
 }
 
+std::unique_ptr<android::snapshot::ISnapshotWriter>
+DynamicPartitionControlAndroid::OpenCowWriter(
+    const std::string& partition_name,
+    const std::optional<std::string>& source_path,
+    bool is_append) {
+  auto suffix = SlotSuffixForSlotNumber(target_slot_);
+
+  std::string device_dir_str;
+  if (!GetDeviceDir(&device_dir_str)) {
+    LOG(ERROR) << "Failed to get device dir!";
+    return nullptr;
+  }
+  base::FilePath device_dir(device_dir_str);
+  auto super_device =
+      device_dir.Append(GetSuperPartitionName(target_slot_)).value();
+  CreateLogicalPartitionParams params = {
+      .block_device = super_device,
+      .metadata_slot = target_slot_,
+      .partition_name = partition_name + suffix,
+      .force_writable = true,
+  };
+  // TODO(zhangkelvin) Open an APPEND mode CowWriter once there's an API to do
+  // it.
+  return snapshot_->OpenSnapshotWriter(params, std::move(source_path));
+}
+
 }  // namespace chromeos_update_engine
diff --git a/dynamic_partition_control_android.h b/dynamic_partition_control_android.h
index f3805f0..9bffb59 100644
--- a/dynamic_partition_control_android.h
+++ b/dynamic_partition_control_android.h
@@ -25,6 +25,7 @@
 #include <base/files/file_util.h>
 #include <libsnapshot/auto_device.h>
 #include <libsnapshot/snapshot.h>
+#include <libsnapshot/snapshot_writer.h>
 
 #include "update_engine/common/dynamic_partition_control_interface.h"
 
@@ -82,6 +83,13 @@
                           uint32_t current_slot,
                           std::string* device);
 
+  // Partition name is expected to be unsuffixed. e.g. system, vendor
+  // Return an interface to write to a snapshoted partition.
+  std::unique_ptr<android::snapshot::ISnapshotWriter> OpenCowWriter(
+      const std::string& unsuffixed_partition_name,
+      const std::optional<std::string>& source_path,
+      bool is_append) override;
+
  protected:
   // These functions are exposed for testing.
 
diff --git a/mock_dynamic_partition_control.h b/mock_dynamic_partition_control.h
index e85df32..5144cbb 100644
--- a/mock_dynamic_partition_control.h
+++ b/mock_dynamic_partition_control.h
@@ -22,6 +22,9 @@
 
 #include <gmock/gmock.h>
 
+#include <libsnapshot/cow_writer.h>
+
+#include "libsnapshot/snapshot_writer.h"
 #include "update_engine/common/boot_control_interface.h"
 #include "update_engine/common/dynamic_partition_control_interface.h"
 #include "update_engine/dynamic_partition_control_android.h"
@@ -81,6 +84,12 @@
               PrepareDynamicPartitionsForUpdate,
               (uint32_t, uint32_t, const DeltaArchiveManifest&, bool),
               (override));
+  MOCK_METHOD(std::unique_ptr<android::snapshot::ISnapshotWriter>,
+              OpenCowWriter,
+              (const std::string& unsuffixed_partition_name,
+               const std::optional<std::string>& source_path,
+               bool is_append),
+              (override));
 
   void set_fake_mapped_devices(const std::set<std::string>& fake) override {
     DynamicPartitionControlAndroid::set_fake_mapped_devices(fake);
diff --git a/payload_consumer/delta_performer.cc b/payload_consumer/delta_performer.cc
index 9c13546..ea00b01 100644
--- a/payload_consumer/delta_performer.cc
+++ b/payload_consumer/delta_performer.cc
@@ -1129,28 +1129,35 @@
   auto&& timestamp_valid = [this](const PartitionUpdate& partition,
                                   bool allow_empty_version,
                                   bool* downgrade_detected) -> ErrorCode {
+    const auto& partition_name = partition.partition_name();
     if (!partition.has_version()) {
+      if (hardware_->GetVersionForLogging(partition_name).empty()) {
+        LOG(INFO) << partition_name << " does't have version, skipping "
+                  << "downgrade check.";
+        return ErrorCode::kSuccess;
+      }
+
       if (allow_empty_version) {
         return ErrorCode::kSuccess;
       }
       LOG(ERROR)
-          << "PartitionUpdate " << partition.partition_name()
-          << " does ot have a version field. Not allowed in partial updates.";
+          << "PartitionUpdate " << partition_name
+          << " doesn't have a version field. Not allowed in partial updates.";
       return ErrorCode::kDownloadManifestParseError;
     }
 
-    auto error_code = hardware_->IsPartitionUpdateValid(
-        partition.partition_name(), partition.version());
+    auto error_code =
+        hardware_->IsPartitionUpdateValid(partition_name, partition.version());
     switch (error_code) {
       case ErrorCode::kSuccess:
         break;
       case ErrorCode::kPayloadTimestampError:
         *downgrade_detected = true;
-        LOG(WARNING) << "PartitionUpdate " << partition.partition_name()
+        LOG(WARNING) << "PartitionUpdate " << partition_name
                      << " has an older version than partition on device.";
         break;
       default:
-        LOG(ERROR) << "IsPartitionUpdateValid(" << partition.partition_name()
+        LOG(ERROR) << "IsPartitionUpdateValid(" << partition_name
                    << ") returned" << utils::ErrorCodeToString(error_code);
         break;
     }