diff --git a/fs_mgr/libsnapshot/include/libsnapshot/snapshot.h b/fs_mgr/libsnapshot/include/libsnapshot/snapshot.h
index 660f60e..bb6ff4e 100644
--- a/fs_mgr/libsnapshot/include/libsnapshot/snapshot.h
+++ b/fs_mgr/libsnapshot/include/libsnapshot/snapshot.h
@@ -20,6 +20,7 @@
 #include <map>
 #include <memory>
 #include <string>
+#include <string_view>
 #include <vector>
 
 #include <android-base/unique_fd.h>
@@ -51,6 +52,9 @@
 struct AutoDeleteCowImage;
 struct AutoDeleteSnapshot;
 struct PartitionCowCreator;
+struct AutoDeviceList;
+
+static constexpr const std::string_view kCowGroupName = "cow";
 
 enum class UpdateState : unsigned int {
     // No update or merge is in progress.
@@ -266,9 +270,8 @@
     // be mapped with two table entries: a dm-snapshot range covering
     // snapshot_size, and a dm-linear range covering the remainder.
     //
-    // All sizes are specified in bytes, and the device, snapshot and COW partition sizes
-    // must be a multiple of the sector size (512 bytes). COW file size will be rounded up
-    // to the nearest sector.
+    // All sizes are specified in bytes, and the device, snapshot, COW partition and COW file sizes
+    // must be a multiple of the sector size (512 bytes).
     bool CreateSnapshot(LockedFile* lock, const std::string& name, SnapshotStatus status);
 
     // |name| should be the base partition name (e.g. "system_a"). Create the
@@ -287,8 +290,7 @@
                      std::string* dev_path);
 
     // Map a COW image that was previous created with CreateCowImage.
-    bool MapCowImage(const std::string& name, const std::chrono::milliseconds& timeout_ms,
-                     std::string* cow_image_device);
+    bool MapCowImage(const std::string& name, const std::chrono::milliseconds& timeout_ms);
 
     // Remove the backing copy-on-write image and snapshot states for the named snapshot. The
     // caller is responsible for ensuring that the snapshot is unmapped.
@@ -369,6 +371,21 @@
     bool MapPartitionWithSnapshot(LockedFile* lock, CreateLogicalPartitionParams params,
                                   std::string* path);
 
+    // Map the COW devices, including the partition in super and the images.
+    // |params|:
+    //    - |partition_name| should be the name of the top-level partition (e.g. system_b),
+    //            not system_b-cow-img
+    //    - |device_name| and |partition| is ignored
+    //    - |timeout_ms| and the rest is respected
+    // Return the path in |cow_device_path| (e.g. /dev/block/dm-1) and major:minor in
+    // |cow_device_string|
+    bool MapCowDevices(LockedFile* lock, const CreateLogicalPartitionParams& params,
+                       const SnapshotStatus& snapshot_status, AutoDeviceList* created_devices,
+                       std::string* cow_name);
+
+    // The reverse of MapCowDevices.
+    bool UnmapCowDevices(LockedFile* lock, const std::string& name);
+
     // The reverse of MapPartitionWithSnapshot.
     bool UnmapPartitionWithSnapshot(LockedFile* lock, const std::string& target_partition_name);
 
diff --git a/fs_mgr/libsnapshot/partition_cow_creator.cpp b/fs_mgr/libsnapshot/partition_cow_creator.cpp
index dd4116b..44d8860 100644
--- a/fs_mgr/libsnapshot/partition_cow_creator.cpp
+++ b/fs_mgr/libsnapshot/partition_cow_creator.cpp
@@ -20,7 +20,9 @@
 
 #include "utility.h"
 
+using android::dm::kSectorSize;
 using android::fs_mgr::Extent;
+using android::fs_mgr::Interval;
 using android::fs_mgr::kDefaultBlockSize;
 using android::fs_mgr::Partition;
 
@@ -44,8 +46,8 @@
     auto target_linear_extent = target_extent->AsLinearExtent();
     if (!target_linear_extent) return nullptr;
 
-    return android::fs_mgr::Interval::Intersect(target_linear_extent->AsInterval(),
-                                                existing_linear_extent->AsInterval())
+    return Interval::Intersect(target_linear_extent->AsInterval(),
+                               existing_linear_extent->AsInterval())
             .AsExtent();
 }
 
@@ -121,6 +123,10 @@
     CHECK(current_metadata->GetBlockDevicePartitionName(0) == LP_METADATA_DEFAULT_PARTITION_NAME &&
           target_metadata->GetBlockDevicePartitionName(0) == LP_METADATA_DEFAULT_PARTITION_NAME);
 
+    uint64_t logical_block_size = current_metadata->logical_block_size();
+    CHECK(logical_block_size != 0 && !(logical_block_size & (logical_block_size - 1)))
+            << "logical_block_size is not power of 2";
+
     Return ret;
     ret.snapshot_status.device_size = target_partition->size();
 
@@ -137,9 +143,31 @@
                 RoundUp(ret.snapshot_status.snapshot_size * kCowEstimateFactor, kDefaultBlockSize);
     }
 
-    // TODO: create COW partition in target_metadata to save space.
-    ret.snapshot_status.cow_partition_size = 0;
+    // Compute regions that are free in both current and target metadata. These are the regions
+    // we can use for COW partition.
+    auto target_free_regions = target_metadata->GetFreeRegions();
+    auto current_free_regions = current_metadata->GetFreeRegions();
+    auto free_regions = Interval::Intersect(target_free_regions, current_free_regions);
+    uint64_t free_region_length = 0;
+    for (const auto& interval : free_regions) {
+        free_region_length += interval.length() * kSectorSize;
+    }
+
+    LOG(INFO) << "Remaining free space for COW: " << free_region_length << " bytes";
+
+    // Compute the COW partition size.
+    ret.snapshot_status.cow_partition_size = std::min(*cow_size, free_region_length);
+    // Round it down to the nearest logical block. Logical partitions must be a multiple
+    // of logical blocks.
+    ret.snapshot_status.cow_partition_size &= ~(logical_block_size - 1);
+    // Assign cow_partition_usable_regions to indicate what regions should the COW partition uses.
+    ret.cow_partition_usable_regions = std::move(free_regions);
+
+    // The rest of the COW space is allocated on ImageManager.
     ret.snapshot_status.cow_file_size = (*cow_size) - ret.snapshot_status.cow_partition_size;
+    // Round it up to the nearest sector.
+    ret.snapshot_status.cow_file_size += kSectorSize - 1;
+    ret.snapshot_status.cow_file_size &= ~(kSectorSize - 1);
 
     return ret;
 }
diff --git a/fs_mgr/libsnapshot/partition_cow_creator.h b/fs_mgr/libsnapshot/partition_cow_creator.h
index d9ee490..a235914 100644
--- a/fs_mgr/libsnapshot/partition_cow_creator.h
+++ b/fs_mgr/libsnapshot/partition_cow_creator.h
@@ -29,6 +29,7 @@
 // Helper class that creates COW for a partition.
 struct PartitionCowCreator {
     using Extent = android::fs_mgr::Extent;
+    using Interval = android::fs_mgr::Interval;
     using MetadataBuilder = android::fs_mgr::MetadataBuilder;
     using Partition = android::fs_mgr::Partition;
 
@@ -48,6 +49,7 @@
 
     struct Return {
         SnapshotManager::SnapshotStatus snapshot_status;
+        std::vector<Interval> cow_partition_usable_regions;
     };
 
     std::optional<Return> Run();
diff --git a/fs_mgr/libsnapshot/snapshot.cpp b/fs_mgr/libsnapshot/snapshot.cpp
index 410074a..111b1da 100644
--- a/fs_mgr/libsnapshot/snapshot.cpp
+++ b/fs_mgr/libsnapshot/snapshot.cpp
@@ -55,6 +55,7 @@
 using android::fs_mgr::CreateDmTable;
 using android::fs_mgr::CreateLogicalPartition;
 using android::fs_mgr::CreateLogicalPartitionParams;
+using android::fs_mgr::GetPartitionGroupName;
 using android::fs_mgr::GetPartitionName;
 using android::fs_mgr::LpMetadata;
 using android::fs_mgr::MetadataBuilder;
@@ -105,7 +106,7 @@
     metadata_dir_ = device_->GetMetadataDir();
 }
 
-[[maybe_unused]] static std::string GetCowName(const std::string& snapshot_name) {
+static std::string GetCowName(const std::string& snapshot_name) {
     return snapshot_name + "-cow";
 }
 
@@ -208,7 +209,7 @@
     CHECK(lock->lock_mode() == LOCK_EX);
     // Sanity check these sizes. Like liblp, we guarantee the partition size
     // is respected, which means it has to be sector-aligned. (This guarantee
-    // is useful for locating avb footers correctly). The COW size, however,
+    // is useful for locating avb footers correctly). The COW file size, however,
     // can be arbitrarily larger than specified, so we can safely round it up.
     if (status.device_size % kSectorSize != 0) {
         LOG(ERROR) << "Snapshot " << name
@@ -220,10 +221,17 @@
                    << status.snapshot_size;
         return false;
     }
-
-    // Round the COW size up to the nearest sector.
-    status.cow_file_size += kSectorSize - 1;
-    status.cow_file_size &= ~(kSectorSize - 1);
+    if (status.cow_partition_size % kSectorSize != 0) {
+        LOG(ERROR) << "Snapshot " << name
+                   << " cow partition size is not a multiple of the sector size: "
+                   << status.cow_partition_size;
+        return false;
+    }
+    if (status.cow_file_size % kSectorSize != 0) {
+        LOG(ERROR) << "Snapshot " << name << " cow file size is not a multiple of the sector size: "
+                   << status.cow_partition_size;
+        return false;
+    }
 
     status.state = SnapshotState::Created;
     status.sectors_allocated = 0;
@@ -256,26 +264,7 @@
 
     std::string cow_image_name = GetCowImageDeviceName(name);
     int cow_flags = IImageManager::CREATE_IMAGE_DEFAULT;
-    if (!images_->CreateBackingImage(cow_image_name, status.cow_file_size, cow_flags)) {
-        return false;
-    }
-
-    // when the kernel creates a persistent dm-snapshot, it requires a CoW file
-    // to store the modifications. The kernel interface does not specify how
-    // the CoW is used, and there is no standard associated.
-    // By looking at the current implementation, the CoW file is treated as:
-    // - a _NEW_ snapshot if its first 32 bits are zero, so the newly created
-    // dm-snapshot device will look like a perfect copy of the origin device;
-    // - an _EXISTING_ snapshot if the first 32 bits are equal to a
-    // kernel-specified magic number and the CoW file metadata is set as valid,
-    // so it can be used to resume the last state of a snapshot device;
-    // - an _INVALID_ snapshot otherwise.
-    // To avoid zero-filling the whole CoW file when a new dm-snapshot is
-    // created, here we zero-fill only the first 32 bits. This is a temporary
-    // workaround that will be discussed again when the kernel API gets
-    // consolidated.
-    ssize_t dm_snap_magic_size = 4;  // 32 bit
-    return images_->ZeroFillNewImage(cow_image_name, dm_snap_magic_size);
+    return images_->CreateBackingImage(cow_image_name, status.cow_file_size, cow_flags);
 }
 
 bool SnapshotManager::MapSnapshot(LockedFile* lock, const std::string& name,
@@ -393,21 +382,24 @@
 }
 
 bool SnapshotManager::MapCowImage(const std::string& name,
-                                  const std::chrono::milliseconds& timeout_ms,
-                                  std::string* cow_dev) {
+                                  const std::chrono::milliseconds& timeout_ms) {
     if (!EnsureImageManager()) return false;
     auto cow_image_name = GetCowImageDeviceName(name);
 
     bool ok;
+    std::string cow_dev;
     if (has_local_image_manager_) {
         // If we forced a local image manager, it means we don't have binder,
         // which means first-stage init. We must use device-mapper.
         const auto& opener = device_->GetPartitionOpener();
-        ok = images_->MapImageWithDeviceMapper(opener, cow_image_name, cow_dev);
+        ok = images_->MapImageWithDeviceMapper(opener, cow_image_name, &cow_dev);
     } else {
-        ok = images_->MapImageDevice(cow_image_name, timeout_ms, cow_dev);
+        ok = images_->MapImageDevice(cow_image_name, timeout_ms, &cow_dev);
     }
-    if (!ok) {
+
+    if (ok) {
+        LOG(INFO) << "Mapped " << cow_image_name << " to " << cow_dev;
+    } else {
         LOG(ERROR) << "Could not map image device: " << cow_image_name;
     }
     return ok;
@@ -441,11 +433,12 @@
     CHECK(lock->lock_mode() == LOCK_EX);
     if (!EnsureImageManager()) return false;
 
+    if (!UnmapCowDevices(lock, name)) {
+        return false;
+    }
+
     auto cow_image_name = GetCowImageDeviceName(name);
     if (images_->BackingImageExists(cow_image_name)) {
-        if (!images_->UnmapImageIfExists(cow_image_name)) {
-            return false;
-        }
         if (!images_->DeleteBackingImage(cow_image_name)) {
             return false;
         }
@@ -1176,6 +1169,12 @@
     }
 
     for (const auto& partition : metadata->partitions) {
+        if (GetPartitionGroupName(metadata->groups[partition.group_index]) == kCowGroupName) {
+            LOG(INFO) << "Skip mapping partition " << GetPartitionName(partition) << " in group "
+                      << kCowGroupName;
+            continue;
+        }
+
         CreateLogicalPartitionParams params = {
                 .block_device = super_device,
                 .metadata = metadata.get(),
@@ -1298,22 +1297,20 @@
         return false;
     }
 
-    // If there is a timeout specified, compute the remaining time to call Map* functions.
-    // init calls CreateLogicalAndSnapshotPartitions, which has no timeout specified. Still call
-    // Map* functions in this case.
     auto remaining_time = GetRemainingTime(params.timeout_ms, begin);
     if (remaining_time.count() < 0) return false;
 
-    std::string cow_image_device;
-    if (!MapCowImage(params.GetPartitionName(), remaining_time, &cow_image_device)) {
-        LOG(ERROR) << "Could not map cow image for partition: " << params.GetPartitionName();
+    std::string cow_name;
+    CreateLogicalPartitionParams cow_params = params;
+    cow_params.timeout_ms = remaining_time;
+    if (!MapCowDevices(lock, cow_params, *live_snapshot_status, &created_devices, &cow_name)) {
         return false;
     }
-    created_devices.EmplaceBack<AutoUnmapImage>(images_.get(),
-                                                GetCowImageDeviceName(params.partition_name));
-
-    // TODO: map cow linear device here
-    std::string cow_device = cow_image_device;
+    std::string cow_device;
+    if (!dm.GetDeviceString(cow_name, &cow_device)) {
+        LOG(ERROR) << "Could not determine major/minor for: " << cow_name;
+        return false;
+    }
 
     remaining_time = GetRemainingTime(params.timeout_ms, begin);
     if (remaining_time.count() < 0) return false;
@@ -1340,12 +1337,11 @@
         return false;
     }
 
-    if (!UnmapCowImage(target_partition_name)) {
+    if (!UnmapCowDevices(lock, target_partition_name)) {
         return false;
     }
 
     auto& dm = DeviceMapper::Instance();
-
     std::string base_name = GetBaseDeviceName(target_partition_name);
     if (!dm.DeleteDeviceIfExists(base_name)) {
         LOG(ERROR) << "Cannot delete base device: " << base_name;
@@ -1357,6 +1353,97 @@
     return true;
 }
 
+bool SnapshotManager::MapCowDevices(LockedFile* lock, const CreateLogicalPartitionParams& params,
+                                    const SnapshotStatus& snapshot_status,
+                                    AutoDeviceList* created_devices, std::string* cow_name) {
+    CHECK(lock);
+    if (!EnsureImageManager()) return false;
+    CHECK(snapshot_status.cow_partition_size + snapshot_status.cow_file_size > 0);
+    auto begin = std::chrono::steady_clock::now();
+
+    std::string partition_name = params.GetPartitionName();
+    std::string cow_image_name = GetCowImageDeviceName(partition_name);
+    *cow_name = GetCowName(partition_name);
+
+    auto& dm = DeviceMapper::Instance();
+
+    // Map COW image if necessary.
+    if (snapshot_status.cow_file_size > 0) {
+        auto remaining_time = GetRemainingTime(params.timeout_ms, begin);
+        if (remaining_time.count() < 0) return false;
+
+        if (!MapCowImage(partition_name, remaining_time)) {
+            LOG(ERROR) << "Could not map cow image for partition: " << partition_name;
+            return false;
+        }
+        created_devices->EmplaceBack<AutoUnmapImage>(images_.get(), cow_image_name);
+
+        // If no COW partition exists, just return the image alone.
+        if (snapshot_status.cow_partition_size == 0) {
+            *cow_name = std::move(cow_image_name);
+            LOG(INFO) << "Mapped COW image for " << partition_name << " at " << *cow_name;
+            return true;
+        }
+    }
+
+    auto remaining_time = GetRemainingTime(params.timeout_ms, begin);
+    if (remaining_time.count() < 0) return false;
+
+    CHECK(snapshot_status.cow_partition_size > 0);
+
+    // Create the DmTable for the COW device. It is the DmTable of the COW partition plus
+    // COW image device as the last extent.
+    CreateLogicalPartitionParams cow_partition_params = params;
+    cow_partition_params.partition = nullptr;
+    cow_partition_params.partition_name = *cow_name;
+    cow_partition_params.device_name.clear();
+    DmTable table;
+    if (!CreateDmTable(cow_partition_params, &table)) {
+        return false;
+    }
+    // If the COW image exists, append it as the last extent.
+    if (snapshot_status.cow_file_size > 0) {
+        std::string cow_image_device;
+        if (!dm.GetDeviceString(cow_image_name, &cow_image_device)) {
+            LOG(ERROR) << "Cannot determine major/minor for: " << cow_image_name;
+            return false;
+        }
+        auto cow_partition_sectors = snapshot_status.cow_partition_size / kSectorSize;
+        auto cow_image_sectors = snapshot_status.cow_file_size / kSectorSize;
+        table.Emplace<DmTargetLinear>(cow_partition_sectors, cow_image_sectors, cow_image_device,
+                                      0);
+    }
+
+    // We have created the DmTable now. Map it.
+    std::string cow_path;
+    if (!dm.CreateDevice(*cow_name, table, &cow_path, remaining_time)) {
+        LOG(ERROR) << "Could not create COW device: " << *cow_name;
+        return false;
+    }
+    created_devices->EmplaceBack<AutoUnmapDevice>(&dm, *cow_name);
+    LOG(INFO) << "Mapped COW device for " << params.GetPartitionName() << " at " << cow_path;
+    return true;
+}
+
+bool SnapshotManager::UnmapCowDevices(LockedFile* lock, const std::string& name) {
+    CHECK(lock);
+    if (!EnsureImageManager()) return false;
+
+    auto& dm = DeviceMapper::Instance();
+    auto cow_name = GetCowName(name);
+    if (!dm.DeleteDeviceIfExists(cow_name)) {
+        LOG(ERROR) << "Cannot unmap " << cow_name;
+        return false;
+    }
+
+    std::string cow_image_name = GetCowImageDeviceName(name);
+    if (!images_->UnmapImageIfExists(cow_image_name)) {
+        LOG(ERROR) << "Cannot unmap image " << cow_image_name;
+        return false;
+    }
+    return true;
+}
+
 auto SnapshotManager::OpenFile(const std::string& file, int open_flags, int lock_flags)
         -> std::unique_ptr<LockedFile> {
     unique_fd fd(open(file.c_str(), open_flags | O_CLOEXEC | O_NOFOLLOW | O_SYNC, 0660));
@@ -1638,6 +1725,22 @@
         suffixed_cow_sizes[name + target_suffix] = size;
     }
 
+    target_metadata->RemoveGroupAndPartitions(kCowGroupName);
+    if (!target_metadata->AddGroup(kCowGroupName, 0)) {
+        LOG(ERROR) << "Cannot add group " << kCowGroupName;
+        return false;
+    }
+
+    // Check that all these metadata is not retrofit dynamic partitions. Snapshots on
+    // devices with retrofit dynamic partitions does not make sense.
+    // This ensures that current_metadata->GetFreeRegions() uses the same device
+    // indices as target_metadata (i.e. 0 -> "super").
+    // This is also assumed in MapCowDevices() call below.
+    CHECK(current_metadata->GetBlockDevicePartitionName(0) == LP_METADATA_DEFAULT_PARTITION_NAME &&
+          target_metadata->GetBlockDevicePartitionName(0) == LP_METADATA_DEFAULT_PARTITION_NAME);
+
+    std::map<std::string, SnapshotStatus> all_snapshot_status;
+
     // In case of error, automatically delete devices that are created along the way.
     // Note that "lock" is destroyed after "created_devices", so it is safe to use |lock| for
     // these devices.
@@ -1648,7 +1751,7 @@
         auto it = suffixed_cow_sizes.find(target_partition->name());
         if (it != suffixed_cow_sizes.end()) {
             cow_size = it->second;
-            LOG(INFO) << "Using provided COW size " << cow_size << " for partition "
+            LOG(INFO) << "Using provided COW size " << *cow_size << " for partition "
                       << target_partition->name();
         }
 
@@ -1694,6 +1797,30 @@
         }
         created_devices.EmplaceBack<AutoDeleteSnapshot>(this, lock.get(), target_partition->name());
 
+        // Create the COW partition. That is, use any remaining free space in super partition before
+        // creating the COW images.
+        if (cow_creator_ret->snapshot_status.cow_partition_size > 0) {
+            CHECK(cow_creator_ret->snapshot_status.cow_partition_size % kSectorSize == 0)
+                    << "cow_partition_size == "
+                    << cow_creator_ret->snapshot_status.cow_partition_size
+                    << " is not a multiple of sector size " << kSectorSize;
+            auto cow_partition = target_metadata->AddPartition(GetCowName(target_partition->name()),
+                                                               kCowGroupName, 0 /* flags */);
+            if (cow_partition == nullptr) {
+                return false;
+            }
+
+            if (!target_metadata->ResizePartition(
+                        cow_partition, cow_creator_ret->snapshot_status.cow_partition_size,
+                        cow_creator_ret->cow_partition_usable_regions)) {
+                LOG(ERROR) << "Cannot create COW partition on metadata with size "
+                           << cow_creator_ret->snapshot_status.cow_partition_size;
+                return false;
+            }
+            // Only the in-memory target_metadata is modified; nothing to clean up if there is an
+            // error in the future.
+        }
+
         // Create the backing COW image if necessary.
         if (cow_creator_ret->snapshot_status.cow_file_size > 0) {
             if (!CreateCowImage(lock.get(), target_partition->name())) {
@@ -1701,13 +1828,55 @@
             }
         }
 
+        all_snapshot_status[target_partition->name()] = std::move(cow_creator_ret->snapshot_status);
+
         LOG(INFO) << "Successfully created snapshot for " << target_partition->name();
     }
 
+    auto& dm = DeviceMapper::Instance();
+    auto exported_target_metadata = target_metadata->Export();
+    CreateLogicalPartitionParams cow_params{
+            .block_device = LP_METADATA_DEFAULT_PARTITION_NAME,
+            .metadata = exported_target_metadata.get(),
+            .timeout_ms = std::chrono::milliseconds::max(),
+            .partition_opener = &device_->GetPartitionOpener(),
+    };
+    for (auto* target_partition : ListPartitionsWithSuffix(target_metadata, target_suffix)) {
+        AutoDeviceList created_devices_for_cow;
+
+        if (!UnmapPartitionWithSnapshot(lock.get(), target_partition->name())) {
+            LOG(ERROR) << "Cannot unmap existing COW devices before re-mapping them for zero-fill: "
+                       << target_partition->name();
+            return false;
+        }
+
+        auto it = all_snapshot_status.find(target_partition->name());
+        CHECK(it != all_snapshot_status.end()) << target_partition->name();
+        cow_params.partition_name = target_partition->name();
+        std::string cow_name;
+        if (!MapCowDevices(lock.get(), cow_params, it->second, &created_devices_for_cow,
+                           &cow_name)) {
+            return false;
+        }
+
+        std::string cow_path;
+        if (!dm.GetDmDevicePathByName(cow_name, &cow_path)) {
+            LOG(ERROR) << "Cannot determine path for " << cow_name;
+            return false;
+        }
+
+        if (!InitializeCow(cow_path)) {
+            LOG(ERROR) << "Can't zero-fill COW device for " << target_partition->name() << ": "
+                       << cow_path;
+            return false;
+        }
+        // Let destructor of created_devices_for_cow to unmap the COW devices.
+    };
+
     created_devices.Release();
     LOG(INFO) << "Successfully created all snapshots for target slot " << target_suffix;
 
-    return res;
+    return true;
 }
 
 bool SnapshotManager::MapUpdateSnapshot(const CreateLogicalPartitionParams& params,
diff --git a/fs_mgr/libsnapshot/snapshot_test.cpp b/fs_mgr/libsnapshot/snapshot_test.cpp
index 429fd8e..086ee95 100644
--- a/fs_mgr/libsnapshot/snapshot_test.cpp
+++ b/fs_mgr/libsnapshot/snapshot_test.cpp
@@ -34,6 +34,7 @@
 #include <liblp/mock_property_fetcher.h>
 
 #include "test_helpers.h"
+#include "utility.h"
 
 namespace android {
 namespace snapshot {
@@ -214,7 +215,9 @@
     void DeleteSnapshotDevice(const std::string& snapshot) {
         DeleteDevice(snapshot);
         DeleteDevice(snapshot + "-inner");
+        DeleteDevice(snapshot + "-cow");
         ASSERT_TRUE(image_manager_->UnmapImageIfExists(snapshot + "-cow-img"));
+        DeleteDevice(snapshot + "-base");
     }
     void DeleteDevice(const std::string& device) {
         if (dm_.GetState(device) != DmDeviceState::INVALID) {
@@ -222,6 +225,35 @@
         }
     }
 
+    AssertionResult CreateCowImage(const std::string& name) {
+        if (!sm->CreateCowImage(lock_.get(), name)) {
+            return AssertionFailure() << "Cannot create COW image " << name;
+        }
+        std::string cow_device;
+        auto map_res = MapCowImage(name, 10s, &cow_device);
+        if (!map_res) {
+            return map_res;
+        }
+        if (!InitializeCow(cow_device)) {
+            return AssertionFailure() << "Cannot zero fill " << cow_device;
+        }
+        if (!sm->UnmapCowImage(name)) {
+            return AssertionFailure() << "Cannot unmap " << name << " after zero filling it";
+        }
+        return AssertionSuccess();
+    }
+
+    AssertionResult MapCowImage(const std::string& name,
+                                const std::chrono::milliseconds& timeout_ms, std::string* path) {
+        if (!sm->MapCowImage(name, timeout_ms)) {
+            return AssertionFailure() << "Cannot map cow image " << name;
+        }
+        if (!dm_.GetDmDevicePathByName(name + "-cow-img"s, path)) {
+            return AssertionFailure() << "No path for " << name << "-cow-img";
+        }
+        return AssertionSuccess();
+    }
+
     DeviceMapper& dm_;
     std::unique_ptr<SnapshotManager::LockedFile> lock_;
     android::fiemap::IImageManager* image_manager_ = nullptr;
@@ -236,7 +268,7 @@
                                    {.device_size = kDeviceSize,
                                     .snapshot_size = kDeviceSize,
                                     .cow_file_size = kDeviceSize}));
-    ASSERT_TRUE(sm->CreateCowImage(lock_.get(), "test-snapshot"));
+    ASSERT_TRUE(CreateCowImage("test-snapshot"));
 
     std::vector<std::string> snapshots;
     ASSERT_TRUE(sm->ListSnapshots(lock_.get(), &snapshots));
@@ -265,13 +297,13 @@
                                    {.device_size = kDeviceSize,
                                     .snapshot_size = kDeviceSize,
                                     .cow_file_size = kDeviceSize}));
-    ASSERT_TRUE(sm->CreateCowImage(lock_.get(), "test-snapshot"));
+    ASSERT_TRUE(CreateCowImage("test-snapshot"));
 
     std::string base_device;
     ASSERT_TRUE(CreatePartition("base-device", kDeviceSize, &base_device));
 
     std::string cow_device;
-    ASSERT_TRUE(sm->MapCowImage("test-snapshot", 10s, &cow_device));
+    ASSERT_TRUE(MapCowImage("test-snapshot", 10s, &cow_device));
 
     std::string snap_device;
     ASSERT_TRUE(sm->MapSnapshot(lock_.get(), "test-snapshot", base_device, cow_device, 10s,
@@ -288,13 +320,13 @@
                                    {.device_size = kDeviceSize,
                                     .snapshot_size = kSnapshotSize,
                                     .cow_file_size = kSnapshotSize}));
-    ASSERT_TRUE(sm->CreateCowImage(lock_.get(), "test-snapshot"));
+    ASSERT_TRUE(CreateCowImage("test-snapshot"));
 
     std::string base_device;
     ASSERT_TRUE(CreatePartition("base-device", kDeviceSize, &base_device));
 
     std::string cow_device;
-    ASSERT_TRUE(sm->MapCowImage("test-snapshot", 10s, &cow_device));
+    ASSERT_TRUE(MapCowImage("test-snapshot", 10s, &cow_device));
 
     std::string snap_device;
     ASSERT_TRUE(sm->MapSnapshot(lock_.get(), "test-snapshot", base_device, cow_device, 10s,
@@ -344,8 +376,8 @@
                                    {.device_size = kDeviceSize,
                                     .snapshot_size = kDeviceSize,
                                     .cow_file_size = kDeviceSize}));
-    ASSERT_TRUE(sm->CreateCowImage(lock_.get(), "test_partition_b"));
-    ASSERT_TRUE(sm->MapCowImage("test_partition_b", 10s, &cow_device));
+    ASSERT_TRUE(CreateCowImage("test_partition_b"));
+    ASSERT_TRUE(MapCowImage("test_partition_b", 10s, &cow_device));
     ASSERT_TRUE(sm->MapSnapshot(lock_.get(), "test_partition_b", base_device, cow_device, 10s,
                                 &snap_device));
 
@@ -403,11 +435,11 @@
                                    {.device_size = kDeviceSize,
                                     .snapshot_size = kDeviceSize,
                                     .cow_file_size = kDeviceSize}));
-    ASSERT_TRUE(sm->CreateCowImage(lock_.get(), "test-snapshot"));
+    ASSERT_TRUE(CreateCowImage("test-snapshot"));
 
     std::string base_device, cow_device, snap_device;
     ASSERT_TRUE(CreatePartition("base-device", kDeviceSize, &base_device));
-    ASSERT_TRUE(sm->MapCowImage("test-snapshot", 10s, &cow_device));
+    ASSERT_TRUE(MapCowImage("test-snapshot", 10s, &cow_device));
     ASSERT_TRUE(sm->MapSnapshot(lock_.get(), "test-snapshot", base_device, cow_device, 10s,
                                 &snap_device));
 
@@ -437,7 +469,7 @@
 
     // Map snapshot should fail now, because we're in a merge-complete state.
     ASSERT_TRUE(AcquireLock());
-    ASSERT_TRUE(sm->MapCowImage("test-snapshot", 10s, &cow_device));
+    ASSERT_TRUE(MapCowImage("test-snapshot", 10s, &cow_device));
     ASSERT_FALSE(sm->MapSnapshot(lock_.get(), "test-snapshot", base_device, cow_device, 10s,
                                  &snap_device));
 
@@ -462,7 +494,7 @@
                                    {.device_size = kDeviceSize,
                                     .snapshot_size = kDeviceSize,
                                     .cow_file_size = kDeviceSize}));
-    ASSERT_TRUE(sm->CreateCowImage(lock_.get(), "test_partition_b"));
+    ASSERT_TRUE(CreateCowImage("test_partition_b"));
 
     // Simulate a reboot into the new slot.
     lock_ = nullptr;
@@ -504,7 +536,7 @@
                                    {.device_size = kDeviceSize,
                                     .snapshot_size = kDeviceSize,
                                     .cow_file_size = kDeviceSize}));
-    ASSERT_TRUE(sm->CreateCowImage(lock_.get(), "test_partition_b"));
+    ASSERT_TRUE(CreateCowImage("test_partition_b"));
 
     // Simulate a reboot into the new slot.
     lock_ = nullptr;
@@ -552,7 +584,7 @@
                                    {.device_size = kDeviceSize,
                                     .snapshot_size = kDeviceSize,
                                     .cow_file_size = kDeviceSize}));
-    ASSERT_TRUE(sm->CreateCowImage(lock_.get(), "test_partition_b"));
+    ASSERT_TRUE(CreateCowImage("test_partition_b"));
 
     // Simulate a reboot into the new slot.
     lock_ = nullptr;
diff --git a/fs_mgr/libsnapshot/utility.cpp b/fs_mgr/libsnapshot/utility.cpp
index 19d58b7..66629e8 100644
--- a/fs_mgr/libsnapshot/utility.cpp
+++ b/fs_mgr/libsnapshot/utility.cpp
@@ -14,6 +14,7 @@
 
 #include "utility.h"
 
+#include <android-base/file.h>
 #include <android-base/logging.h>
 #include <android-base/strings.h>
 
@@ -75,5 +76,38 @@
     }
 }
 
+bool InitializeCow(const std::string& device) {
+    // When the kernel creates a persistent dm-snapshot, it requires a CoW file
+    // to store the modifications. The kernel interface does not specify how
+    // the CoW is used, and there is no standard associated.
+    // By looking at the current implementation, the CoW file is treated as:
+    // - a _NEW_ snapshot if its first 32 bits are zero, so the newly created
+    // dm-snapshot device will look like a perfect copy of the origin device;
+    // - an _EXISTING_ snapshot if the first 32 bits are equal to a
+    // kernel-specified magic number and the CoW file metadata is set as valid,
+    // so it can be used to resume the last state of a snapshot device;
+    // - an _INVALID_ snapshot otherwise.
+    // To avoid zero-filling the whole CoW file when a new dm-snapshot is
+    // created, here we zero-fill only the first 32 bits. This is a temporary
+    // workaround that will be discussed again when the kernel API gets
+    // consolidated.
+    // TODO(b/139202197): Remove this hack once the kernel API is consolidated.
+    constexpr ssize_t kDmSnapZeroFillSize = 4;  // 32-bit
+
+    char zeros[kDmSnapZeroFillSize] = {0};
+    android::base::unique_fd fd(open(device.c_str(), O_WRONLY | O_BINARY));
+    if (fd < 0) {
+        PLOG(ERROR) << "Can't open COW device: " << device;
+        return false;
+    }
+
+    LOG(INFO) << "Zero-filling COW device: " << device;
+    if (!android::base::WriteFully(fd, zeros, kDmSnapZeroFillSize)) {
+        PLOG(ERROR) << "Can't zero-fill COW device for " << device;
+        return false;
+    }
+    return true;
+}
+
 }  // namespace snapshot
 }  // namespace android
diff --git a/fs_mgr/libsnapshot/utility.h b/fs_mgr/libsnapshot/utility.h
index 7e7e24a..a700c46 100644
--- a/fs_mgr/libsnapshot/utility.h
+++ b/fs_mgr/libsnapshot/utility.h
@@ -103,5 +103,8 @@
 std::vector<android::fs_mgr::Partition*> ListPartitionsWithSuffix(
         android::fs_mgr::MetadataBuilder* builder, const std::string& suffix);
 
+// Initialize a device before using it as the COW device for a dm-snapshot device.
+bool InitializeCow(const std::string& device);
+
 }  // namespace snapshot
 }  // namespace android