diff --git a/aptos-move/aptos-aggregator/src/aggregator_extension.rs b/aptos-move/aptos-aggregator/src/aggregator_extension.rs
index 3495706859cd8..c5c1a49d6eda5 100644
--- a/aptos-move/aptos-aggregator/src/aggregator_extension.rs
+++ b/aptos-move/aptos-aggregator/src/aggregator_extension.rs
@@ -60,6 +60,7 @@ pub fn aggregator_id_for_test(key: u128) -> AggregatorID {
 /// This graph shows how delta of aggregator changed during a single transaction
 /// execution:
 ///
+/// ```text
 /// +A ===========================================>
 ///            ||
 ///          ||||                               +X
@@ -70,6 +71,7 @@ pub fn aggregator_id_for_test(key: u128) -> AggregatorID {
 ///                         ||
 ///                         ||
 /// -B ===========================================>
+/// ```
 ///
 /// Clearly, +X succeeds if +A and -B succeed. Therefore each delta
 /// validation consists of:
diff --git a/aptos-move/aptos-aggregator/src/delta_change_set.rs b/aptos-move/aptos-aggregator/src/delta_change_set.rs
index cff1d402ff8a7..4b2c67753a807 100644
--- a/aptos-move/aptos-aggregator/src/delta_change_set.rs
+++ b/aptos-move/aptos-aggregator/src/delta_change_set.rs
@@ -10,10 +10,9 @@ use aptos_state_view::StateView;
 use aptos_types::{
     state_store::state_key::StateKey,
     vm_status::{StatusCode, VMStatus},
-    write_set::{WriteOp, WriteSet, WriteSetMut},
+    write_set::WriteOp,
 };
 use move_binary_format::errors::{Location, PartialVMError, PartialVMResult};
-use std::collections::{btree_map::Entry, BTreeMap};
 
 /// When `Addition` operation overflows the `limit`.
 const EADD_OVERFLOW: u64 = 0x02_0001;
@@ -288,115 +287,6 @@ pub fn delta_add(v: u128, limit: u128) -> DeltaOp {
     DeltaOp::new(DeltaUpdate::Plus(v), limit, v, 0)
 }
 
-/// `DeltaChangeSet` contains all access paths that one transaction wants to update with deltas.
-#[derive(Clone, Debug, Eq, Hash, PartialEq)]
-pub struct DeltaChangeSet {
-    delta_change_set: BTreeMap<StateKey, DeltaOp>,
-}
-
-impl DeltaChangeSet {
-    pub fn empty() -> Self {
-        DeltaChangeSet {
-            delta_change_set: BTreeMap::new(),
-        }
-    }
-
-    pub fn len(&self) -> usize {
-        self.delta_change_set.len()
-    }
-
-    pub fn new(delta_change_set: impl IntoIterator<Item = (StateKey, DeltaOp)>) -> Self {
-        DeltaChangeSet {
-            delta_change_set: delta_change_set.into_iter().collect(),
-        }
-    }
-
-    pub fn get(&self, key: &StateKey) -> Option<&DeltaOp> {
-        self.delta_change_set.get(key)
-    }
-
-    pub fn insert(&mut self, delta: (StateKey, DeltaOp)) {
-        self.delta_change_set.insert(delta.0, delta.1);
-    }
-
-    pub fn remove(&mut self, key: &StateKey) -> Option<DeltaOp> {
-        self.delta_change_set.remove(key)
-    }
-
-    #[inline]
-    pub fn iter(&self) -> ::std::collections::btree_map::Iter<'_, StateKey, DeltaOp> {
-        self.into_iter()
-    }
-
-    #[inline]
-    pub fn entry(&mut self, key: StateKey) -> Entry<StateKey, DeltaOp> {
-        self.delta_change_set.entry(key)
-    }
-
-    #[inline]
-    pub fn is_empty(&self) -> bool {
-        self.delta_change_set.is_empty()
-    }
-
-    pub fn as_inner_mut(&mut self) -> &mut BTreeMap<StateKey, DeltaOp> {
-        &mut self.delta_change_set
-    }
-
-    /// Converts deltas to a vector of write ops. In case conversion to a write op
-    /// failed, the error is propagated to the caller.
-    pub fn try_materialize(
-        self,
-        state_view: &dyn StateView,
-    ) -> anyhow::Result<Vec<(StateKey, WriteOp)>, VMStatus> {
-        // Converts every item of DeltaChangeSet into an item of a WriteSet. If
-        // conversion fails, error is returned.
-        let into_write_set_item =
-            |item: (StateKey, DeltaOp)| -> anyhow::Result<(StateKey, WriteOp), VMStatus> {
-                let write_op = item.1.try_into_write_op(state_view, &item.0)?;
-                Ok((item.0, write_op))
-            };
-
-        self.delta_change_set
-            .into_iter()
-            .map(into_write_set_item)
-            .collect()
-    }
-
-    /// Consumes the delta change set and tries to materialize it into a write set.
-    pub fn try_into_write_set(
-        self,
-        state_view: &dyn StateView,
-    ) -> anyhow::Result<WriteSet, VMStatus> {
-        let materialized_write_set = self.try_materialize(state_view)?;
-        WriteSetMut::new(materialized_write_set)
-            .freeze()
-            .map_err(|_err| {
-                VMStatus::error(
-                    StatusCode::UNKNOWN_INVARIANT_VIOLATION_ERROR,
-                    Some("Error when freezing materialized deltas.".to_string()),
-                )
-            })
-    }
-}
-
-impl<'a> IntoIterator for &'a DeltaChangeSet {
-    type IntoIter = ::std::collections::btree_map::Iter<'a, StateKey, DeltaOp>;
-    type Item = (&'a StateKey, &'a DeltaOp);
-
-    fn into_iter(self) -> Self::IntoIter {
-        self.delta_change_set.iter()
-    }
-}
-
-impl ::std::iter::IntoIterator for DeltaChangeSet {
-    type IntoIter = ::std::collections::btree_map::IntoIter<StateKey, DeltaOp>;
-    type Item = (StateKey, DeltaOp);
-
-    fn into_iter(self) -> Self::IntoIter {
-        self.delta_change_set.into_iter()
-    }
-}
-
 #[cfg(test)]
 mod test {
     use super::*;
@@ -665,14 +555,6 @@ mod test {
         );
     }
 
-    #[test]
-    fn test_empty_storage_error_propagated() {
-        let state_view = FakeDataStore::default();
-        let deltas = vec![(KEY.clone(), delta_add(10, 100))];
-        let delta_change_set = DeltaChangeSet::new(deltas);
-        assert_err!(delta_change_set.try_into_write_set(&state_view));
-    }
-
     struct BadStorage;
 
     impl TStateView for BadStorage {
@@ -708,34 +590,6 @@ mod test {
         );
     }
 
-    #[test]
-    fn test_storage_error_propagated() {
-        let state_view = BadStorage;
-        let deltas = vec![(KEY.clone(), delta_add(10, 100))];
-        let delta_change_set = DeltaChangeSet::new(deltas);
-        assert_matches!(
-            delta_change_set.try_into_write_set(&state_view),
-            Err(VMStatus::Error {
-                status_code: StatusCode::STORAGE_ERROR,
-                message: Some(_),
-                sub_status: None
-            })
-        );
-    }
-
-    #[test]
-    fn test_delta_materialization_failure() {
-        let mut state_view = FakeDataStore::default();
-        state_view.set_legacy(KEY.clone(), serialize(&99));
-
-        let deltas = vec![(KEY.clone(), delta_add(10, 100))];
-        let delta_change_set = DeltaChangeSet::new(deltas);
-        assert_matches!(
-            delta_change_set.try_into_write_set(&state_view),
-            Err(VMStatus::MoveAbort(_, EADD_OVERFLOW))
-        );
-    }
-
     #[test]
     fn test_successful_write_op_conversion() {
         let mut state_view = FakeDataStore::default();
diff --git a/aptos-move/aptos-vm-types/src/change_set.rs b/aptos-move/aptos-vm-types/src/change_set.rs
index 31e6749d65515..f7c1eab09f5c2 100644
--- a/aptos-move/aptos-vm-types/src/change_set.rs
+++ b/aptos-move/aptos-vm-types/src/change_set.rs
@@ -2,129 +2,236 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use crate::check_change_set::CheckChangeSet;
-use aptos_aggregator::delta_change_set::{deserialize, serialize, DeltaChangeSet};
+use aptos_aggregator::delta_change_set::{deserialize, serialize, DeltaOp};
 use aptos_state_view::StateView;
 use aptos_types::{
     contract_event::ContractEvent,
-    write_set::{WriteOp, WriteSet},
+    state_store::state_key::{StateKey, StateKeyInner},
+    transaction::ChangeSet as StorageChangeSet,
+    write_set::{WriteOp, WriteSetMut},
 };
 use move_binary_format::errors::Location;
 use move_core_types::vm_status::{err_msg, StatusCode, VMStatus};
-use std::collections::btree_map::Entry::{Occupied, Vacant};
+use std::collections::{
+    btree_map::Entry::{Occupied, Vacant},
+    BTreeMap,
+};
 
-/// A change set produced by the VM. Just like VMOutput, this type should
-/// be used inside the VM. For storage backends, use ChangeSet.
-#[derive(Debug, Clone)]
+/// A change set produced by the VM.
+///
+/// **WARNING**: Just like VMOutput, this type should only be used inside the
+/// VM. For storage backends, use `ChangeSet`.
+#[derive(Debug, Clone, Eq, PartialEq)]
 pub struct VMChangeSet {
-    write_set: WriteSet,
-    delta_change_set: DeltaChangeSet,
+    resource_write_set: BTreeMap<StateKey, WriteOp>,
+    module_write_set: BTreeMap<StateKey, WriteOp>,
+    aggregator_write_set: BTreeMap<StateKey, WriteOp>,
+    aggregator_delta_set: BTreeMap<StateKey, DeltaOp>,
     events: Vec<ContractEvent>,
 }
 
+macro_rules! squash_writes_pair {
+    ($write_entry:ident, $additional_write:ident) => {
+        // Squashing creation and deletion is a no-op. In that case, we
+        // have to remove the old write op from the write set.
+        let noop = !WriteOp::squash($write_entry.get_mut(), $additional_write).map_err(|e| {
+            VMStatus::error(
+                StatusCode::UNKNOWN_INVARIANT_VIOLATION_ERROR,
+                err_msg(format!("Error while squashing two write ops: {}.", e)),
+            )
+        })?;
+        if noop {
+            $write_entry.remove();
+        }
+    };
+}
+
 impl VMChangeSet {
-    /// Returns an empty change set.
     pub fn empty() -> Self {
         Self {
-            write_set: WriteSet::default(),
-            delta_change_set: DeltaChangeSet::empty(),
+            resource_write_set: BTreeMap::new(),
+            module_write_set: BTreeMap::new(),
+            aggregator_write_set: BTreeMap::new(),
+            aggregator_delta_set: BTreeMap::new(),
             events: vec![],
         }
     }
 
-    /// Returns a new change set, and checks that it is well-formed.
     pub fn new(
-        write_set: WriteSet,
-        delta_change_set: DeltaChangeSet,
+        resource_write_set: BTreeMap<StateKey, WriteOp>,
+        module_write_set: BTreeMap<StateKey, WriteOp>,
+        aggregator_write_set: BTreeMap<StateKey, WriteOp>,
+        aggregator_delta_set: BTreeMap<StateKey, DeltaOp>,
         events: Vec<ContractEvent>,
         checker: &dyn CheckChangeSet,
     ) -> anyhow::Result<Self, VMStatus> {
-        // Check that writes and deltas have disjoint key set.
-        let disjoint = delta_change_set
-            .iter()
-            .all(|(k, _)| write_set.get(k).is_none());
-        if !disjoint {
-            return Err(VMStatus::error(
-                StatusCode::DATA_FORMAT_ERROR,
-                err_msg("DeltaChangeSet and WriteSet are not disjoint."),
-            ));
+        let change_set = Self {
+            resource_write_set,
+            module_write_set,
+            aggregator_write_set,
+            aggregator_delta_set,
+            events,
+        };
+
+        // Returns an error if structure of the change set is not valid,
+        // e.g. the size in bytes is too large.
+        checker.check_change_set(&change_set)?;
+        Ok(change_set)
+    }
+
+    /// Builds a new change set from the storage representation.
+    ///
+    /// **WARNING**: Has complexity O(#write_ops) because we need to iterate
+    /// over blobs and split them into resources or modules. Only used to
+    /// support transactions with write-set payload.
+    pub fn try_from_storage_change_set(
+        change_set: StorageChangeSet,
+        checker: &dyn CheckChangeSet,
+    ) -> anyhow::Result<Self, VMStatus> {
+        let (write_set, events) = change_set.into_inner();
+
+        // There should be no aggregator writes if we have a change set from
+        // storage.
+        let mut resource_write_set = BTreeMap::new();
+        let mut module_write_set = BTreeMap::new();
+
+        for (state_key, write_op) in write_set {
+            if matches!(state_key.inner(), StateKeyInner::AccessPath(ap) if ap.is_code()) {
+                module_write_set.insert(state_key, write_op);
+            } else {
+                // TODO(aggregator) While everything else must be a resource, first
+                // version of aggregators is implemented as a table item. Revisit when
+                // we split MVHashMap into data and aggregators.
+                resource_write_set.insert(state_key, write_op);
+            }
         }
 
         let change_set = Self {
-            write_set,
-            delta_change_set,
+            resource_write_set,
+            module_write_set,
+            aggregator_write_set: BTreeMap::new(),
+            aggregator_delta_set: BTreeMap::new(),
             events,
         };
-
-        // Check the newly-formed change set.
         checker.check_change_set(&change_set)?;
         Ok(change_set)
     }
 
-    pub fn write_set(&self) -> &WriteSet {
-        &self.write_set
+    pub(crate) fn into_storage_change_set_unchecked(self) -> StorageChangeSet {
+        let Self {
+            resource_write_set,
+            module_write_set,
+            aggregator_write_set,
+            aggregator_delta_set: _,
+            events,
+        } = self;
+
+        let mut write_set_mut = WriteSetMut::default();
+        write_set_mut.extend(resource_write_set);
+        write_set_mut.extend(module_write_set);
+        write_set_mut.extend(aggregator_write_set);
+
+        let write_set = write_set_mut
+            .freeze()
+            .expect("Freezing a WriteSet does not fail.");
+        StorageChangeSet::new(write_set, events)
     }
 
-    pub fn delta_change_set(&self) -> &DeltaChangeSet {
-        &self.delta_change_set
+    /// Converts VM-native change set into its storage representation with fully
+    /// serialized changes.
+    /// If deltas are not materialized the conversion fails.
+    pub fn try_into_storage_change_set(self) -> anyhow::Result<StorageChangeSet, VMStatus> {
+        if !self.aggregator_delta_set.is_empty() {
+            return Err(VMStatus::error(
+                StatusCode::DATA_FORMAT_ERROR,
+                err_msg(
+                    "Cannot convert from VMChangeSet with non-materialized deltas to ChangeSet.",
+                ),
+            ));
+        }
+        Ok(self.into_storage_change_set_unchecked())
     }
 
-    pub fn events(&self) -> &[ContractEvent] {
-        &self.events
+    pub fn write_set_iter(&self) -> impl Iterator<Item = (&StateKey, &WriteOp)> {
+        self.resource_write_set
+            .iter()
+            .chain(self.module_write_set.iter())
+            .chain(self.aggregator_write_set.iter())
     }
 
-    pub fn unpack(self) -> (WriteSet, DeltaChangeSet, Vec<ContractEvent>) {
-        (self.write_set, self.delta_change_set, self.events)
+    pub fn resource_write_set(&self) -> &BTreeMap<StateKey, WriteOp> {
+        &self.resource_write_set
+    }
+
+    pub fn module_write_set(&self) -> &BTreeMap<StateKey, WriteOp> {
+        &self.module_write_set
+    }
+
+    // Called by `try_into_transaction_output_with_materialized_writes` only.
+    pub(crate) fn extend_aggregator_write_set(
+        &mut self,
+        additional_aggregator_writes: impl Iterator<Item = (StateKey, WriteOp)>,
+    ) {
+        self.aggregator_write_set
+            .extend(additional_aggregator_writes)
+    }
+
+    pub fn aggregator_write_set(&self) -> &BTreeMap<StateKey, WriteOp> {
+        &self.aggregator_write_set
+    }
+
+    pub fn aggregator_delta_set(&self) -> &BTreeMap<StateKey, DeltaOp> {
+        &self.aggregator_delta_set
+    }
+
+    pub fn events(&self) -> &[ContractEvent] {
+        &self.events
     }
 
     /// Materializes this change set: all deltas are converted into writes and
-    /// are combined with existing write set. In case of materialization fails,
-    /// an error is returned.
+    /// are combined with existing aggregator writes.
     pub fn try_materialize(self, state_view: &impl StateView) -> anyhow::Result<Self, VMStatus> {
-        let (write_set, delta_change_set, events) = self.unpack();
+        let Self {
+            resource_write_set,
+            module_write_set,
+            mut aggregator_write_set,
+            aggregator_delta_set,
+            events,
+        } = self;
 
-        // Try to materialize deltas and add them to the write set.
-        let mut write_set_mut = write_set.into_mut();
-        let delta_writes = delta_change_set.try_materialize(state_view)?;
-        delta_writes
-            .into_iter()
-            .for_each(|item| write_set_mut.insert(item));
+        let into_write =
+            |(state_key, delta): (StateKey, DeltaOp)| -> anyhow::Result<(StateKey, WriteOp), VMStatus> {
+                let write = delta.try_into_write_op(state_view, &state_key)?;
+                Ok((state_key, write))
+            };
 
-        let write_set = write_set_mut.freeze().map_err(|_| {
-            VMStatus::error(
-                StatusCode::DATA_FORMAT_ERROR,
-                err_msg("Failed to freeze write when materializing VMChangeSet"),
-            )
-        })?;
+        let materialized_aggregator_delta_set =
+            aggregator_delta_set
+                .into_iter()
+                .map(into_write)
+                .collect::<anyhow::Result<BTreeMap<StateKey, WriteOp>, VMStatus>>()?;
+        aggregator_write_set.extend(materialized_aggregator_delta_set.into_iter());
 
         Ok(Self {
-            write_set,
-            delta_change_set: DeltaChangeSet::empty(),
+            resource_write_set,
+            module_write_set,
+            aggregator_write_set,
+            aggregator_delta_set: BTreeMap::new(),
             events,
         })
     }
 
-    /// Squashes `next` change set on top of this change set. The squashed
-    /// change set is then checked using the `checker`.
-    pub fn squash(
-        self,
-        next: Self,
-        checker: &dyn CheckChangeSet,
-    ) -> anyhow::Result<Self, VMStatus> {
+    fn squash_additional_aggregator_changes(
+        aggregator_write_set: &mut BTreeMap<StateKey, WriteOp>,
+        aggregator_delta_set: &mut BTreeMap<StateKey, DeltaOp>,
+        additional_aggregator_write_set: BTreeMap<StateKey, WriteOp>,
+        additional_aggregator_delta_set: BTreeMap<StateKey, DeltaOp>,
+    ) -> anyhow::Result<(), VMStatus> {
         use WriteOp::*;
 
-        // First, obtain write sets, delta change sets and events of this and other
-        // change sets.
-        let (next_write_set, next_delta_change_set, next_events) = next.unpack();
-        let (write_set, mut delta_change_set, mut events) = self.unpack();
-        let mut write_set_mut = write_set.into_mut();
-
-        // We are modifying current sets, so grab a mutable reference for them.
-        let delta_ops = delta_change_set.as_inner_mut();
-        let write_ops = write_set_mut.as_inner_mut();
-
-        // First, squash incoming deltas.
-        for (key, next_delta_op) in next_delta_change_set.into_iter() {
-            if let Some(write_op) = write_ops.get_mut(&key) {
+        // First, squash deltas.
+        for (key, additional_delta_op) in additional_aggregator_delta_set {
+            if let Some(write_op) = aggregator_write_set.get_mut(&key) {
                 // In this case, delta follows a write op.
                 match write_op {
                     Creation(data)
@@ -133,7 +240,7 @@ impl VMChangeSet {
                     | ModificationWithMetadata { data, .. } => {
                         // Apply delta on top of creation or modification.
                         let base: u128 = deserialize(data);
-                        let value = next_delta_op
+                        let value = additional_delta_op
                             .apply_to(base)
                             .map_err(|e| e.finish(Location::Undefined).into_vm_status())?;
                         *data = serialize(&value);
@@ -151,69 +258,95 @@ impl VMChangeSet {
             } else {
                 // Otherwise, this is a either a new delta or an additional delta
                 // for the same state key.
-                match delta_ops.entry(key) {
+                match aggregator_delta_set.entry(key) {
                     Occupied(entry) => {
                         // In this case, we need to merge the new incoming delta
                         // to the existing delta, ensuring the strict ordering.
                         entry
                             .into_mut()
-                            .merge_with_next_delta(next_delta_op)
+                            .merge_with_next_delta(additional_delta_op)
                             .map_err(|e| e.finish(Location::Undefined).into_vm_status())?;
                     },
                     Vacant(entry) => {
                         // We see this delta for the first time, so simply add it
                         // to the set.
-                        entry.insert(next_delta_op);
+                        entry.insert(additional_delta_op);
                     },
                 }
             }
         }
 
         // Next, squash write ops.
-        for (key, next_write_op) in next_write_set.into_iter() {
-            match write_ops.entry(key) {
+        for (key, additional_write_op) in additional_aggregator_write_set {
+            match aggregator_write_set.entry(key) {
                 Occupied(mut entry) => {
-                    // Squashing creation and deletion is a no-op. In that case, we
-                    // have to remove the old write op from the write set.
-                    let noop = !WriteOp::squash(entry.get_mut(), next_write_op).map_err(|e| {
-                        VMStatus::error(
-                            StatusCode::UNKNOWN_INVARIANT_VIOLATION_ERROR,
-                            err_msg(format!("Error while squashing two write ops: {}.", e)),
-                        )
-                    })?;
-                    if noop {
-                        entry.remove();
-                    }
+                    squash_writes_pair!(entry, additional_write_op);
                 },
                 Vacant(entry) => {
                     // This is a new write op. It can overwrite a delta so we
                     // have to make sure we remove such a delta from the set in
                     // this case.
-                    let removed_delta = delta_change_set.remove(entry.key());
+                    let removed_delta = aggregator_delta_set.remove(entry.key());
 
                     // We cannot create after modification with a delta!
-                    if removed_delta.is_some() && next_write_op.is_creation() {
+                    if removed_delta.is_some() && additional_write_op.is_creation() {
                         return Err(VMStatus::error(
                             StatusCode::UNKNOWN_INVARIANT_VIOLATION_ERROR,
                             err_msg("Cannot create a resource after modification with a delta."),
                         ));
                     }
 
-                    entry.insert(next_write_op);
+                    entry.insert(additional_write_op);
                 },
             }
         }
 
-        let write_set = write_set_mut.freeze().map_err(|_| {
-            VMStatus::error(
-                StatusCode::UNKNOWN_INVARIANT_VIOLATION_ERROR,
-                err_msg("Error when freezing squashed write sets."),
-            )
-        })?;
+        Ok(())
+    }
+
+    fn squash_additional_writes(
+        write_set: &mut BTreeMap<StateKey, WriteOp>,
+        additional_write_set: BTreeMap<StateKey, WriteOp>,
+    ) -> anyhow::Result<(), VMStatus> {
+        for (key, additional_write_op) in additional_write_set.into_iter() {
+            match write_set.entry(key) {
+                Occupied(mut entry) => {
+                    squash_writes_pair!(entry, additional_write_op);
+                },
+                Vacant(entry) => {
+                    entry.insert(additional_write_op);
+                },
+            }
+        }
+        Ok(())
+    }
+
+    pub fn squash_additional_change_set(
+        &mut self,
+        additional_change_set: Self,
+        checker: &dyn CheckChangeSet,
+    ) -> anyhow::Result<(), VMStatus> {
+        let Self {
+            resource_write_set: additional_resource_write_set,
+            module_write_set: additional_module_write_set,
+            aggregator_write_set: additional_aggregator_write_set,
+            aggregator_delta_set: additional_aggregator_delta_set,
+            events: additional_events,
+        } = additional_change_set;
 
-        // Squash events.
-        events.extend(next_events);
+        Self::squash_additional_aggregator_changes(
+            &mut self.aggregator_write_set,
+            &mut self.aggregator_delta_set,
+            additional_aggregator_write_set,
+            additional_aggregator_delta_set,
+        )?;
+        Self::squash_additional_writes(
+            &mut self.resource_write_set,
+            additional_resource_write_set,
+        )?;
+        Self::squash_additional_writes(&mut self.module_write_set, additional_module_write_set)?;
+        self.events.extend(additional_events);
 
-        Self::new(write_set, delta_change_set, events, checker)
+        checker.check_change_set(self)
     }
 }
diff --git a/aptos-move/aptos-vm-types/src/check_change_set.rs b/aptos-move/aptos-vm-types/src/check_change_set.rs
index d6aab81c37018..f6a71476d753a 100644
--- a/aptos-move/aptos-vm-types/src/check_change_set.rs
+++ b/aptos-move/aptos-vm-types/src/check_change_set.rs
@@ -4,9 +4,8 @@
 use crate::change_set::VMChangeSet;
 use move_core_types::vm_status::VMStatus;
 
-/// Useful trait for checking the contents of a change set. For example, the
-/// total number of bytes ber write op or event can be checked.
+/// Trait to check the contents of a change set, e.g. the total number of
+/// bytes per write op or event.
 pub trait CheckChangeSet {
-    /// Returns an error if the change set does not pass the check.
     fn check_change_set(&self, change_set: &VMChangeSet) -> anyhow::Result<(), VMStatus>;
 }
diff --git a/aptos-move/aptos-vm-types/src/output.rs b/aptos-move/aptos-vm-types/src/output.rs
index 5e40abf94dc67..fd9a2f45ff307 100644
--- a/aptos-move/aptos-vm-types/src/output.rs
+++ b/aptos-move/aptos-vm-types/src/output.rs
@@ -2,20 +2,20 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use crate::change_set::VMChangeSet;
-use aptos_aggregator::delta_change_set::DeltaChangeSet;
 use aptos_state_view::StateView;
 use aptos_types::{
-    contract_event::ContractEvent,
     fee_statement::FeeStatement,
     state_store::state_key::StateKey,
     transaction::{TransactionOutput, TransactionStatus},
-    write_set::{WriteOp, WriteSet},
+    write_set::WriteOp,
 };
 use move_core_types::vm_status::VMStatus;
 
-/// Output produced by the VM. Before VMOutput is passed to storage backends,
-/// it must be converted to TransactionOutput.
-#[derive(Debug, Clone)]
+/// Output produced by the VM after executing a transaction.
+///
+/// **WARNING**: This type should only be used inside the VM. For storage backends,
+/// use `TransactionOutput`.
+#[derive(Debug, Clone, Eq, PartialEq)]
 pub struct VMOutput {
     change_set: VMChangeSet,
     fee_statement: FeeStatement,
@@ -35,8 +35,6 @@ impl VMOutput {
         }
     }
 
-    /// Returns a new empty transaction output. Useful for handling discards or
-    /// system transactions.
     pub fn empty_with_status(status: TransactionStatus) -> Self {
         Self {
             change_set: VMChangeSet::empty(),
@@ -53,16 +51,8 @@ impl VMOutput {
         (self.change_set, self.fee_statement, self.status)
     }
 
-    pub fn write_set(&self) -> &WriteSet {
-        self.change_set.write_set()
-    }
-
-    pub fn delta_change_set(&self) -> &DeltaChangeSet {
-        self.change_set.delta_change_set()
-    }
-
-    pub fn events(&self) -> &[ContractEvent] {
-        self.change_set.events()
+    pub fn change_set(&self) -> &VMChangeSet {
+        &self.change_set
     }
 
     pub fn gas_used(&self) -> u64 {
@@ -77,20 +67,17 @@ impl VMOutput {
         &self.status
     }
 
-    /// Materializes this transaction output by materializing the change set it
-    /// carries. Materialization can fail due to delta applications, in which
-    /// case an error is returned.
-    /// If the call succeeds (returns `Ok(..)`), the output is guaranteed to have
-    /// an empty delta change set.
+    /// Materializes delta sets.
+    /// Guarantees that if deltas are materialized successfully, the output
+    /// has an empty delta set.
     pub fn try_materialize(self, state_view: &impl StateView) -> anyhow::Result<Self, VMStatus> {
         // First, check if output of transaction should be discarded or delta
         // change set is empty. In both cases, we do not need to apply any
         // deltas and can return immediately.
-        if self.status().is_discarded() || self.delta_change_set().is_empty() {
+        if self.status().is_discarded() || self.change_set().aggregator_delta_set().is_empty() {
             return Ok(self);
         }
 
-        // Try to materialize deltas and add them to the write set.
         let (change_set, fee_statement, status) = self.unpack_with_fee_statement();
         let materialized_change_set = change_set.try_materialize(state_view)?;
         Ok(VMOutput::new(
@@ -100,50 +87,48 @@ impl VMOutput {
         ))
     }
 
-    /// Converts VMOutput into TransactionOutput which can be used by storage
-    /// backends. During this conversion delta materialization can fail, in
-    /// which case an error is returned.
-    pub fn into_transaction_output(
+    /// Same as `try_materialize` but also constructs `TransactionOutput`.
+    pub fn try_into_transaction_output(
         self,
         state_view: &impl StateView,
     ) -> anyhow::Result<TransactionOutput, VMStatus> {
         let materialized_output = self.try_materialize(state_view)?;
-        let (change_set, gas_used, status) = materialized_output.unpack();
-        let (write_set, delta_change_set, events) = change_set.unpack();
-
         debug_assert!(
-            delta_change_set.is_empty(),
-            "DeltaChangeSet must be empty after materialization."
+            materialized_output
+                .change_set()
+                .aggregator_delta_set()
+                .is_empty(),
+            "Aggregator deltas must be empty after materialization."
         );
-
+        let (vm_change_set, gas_used, status) = materialized_output.unpack();
+        let (write_set, events) = vm_change_set.try_into_storage_change_set()?.into_inner();
         Ok(TransactionOutput::new(write_set, events, gas_used, status))
     }
 
-    /// Converts VM output into transaction output which storage or state sync
-    /// can understand. Extends writes with values from materialized deltas.
-    pub fn output_with_delta_writes(
-        self,
-        delta_writes: Vec<(StateKey, WriteOp)>,
+    /// Similar to `try_into_transaction_output` but deltas are materialized
+    /// externally by the caller beforehand.
+    pub fn into_transaction_output_with_materialized_deltas(
+        mut self,
+        materialized_deltas: Vec<(StateKey, WriteOp)>,
     ) -> TransactionOutput {
-        let (change_set, gas_used, status) = self.unpack();
-        let (write_set, mut delta_change_set, events) = change_set.unpack();
-        let mut write_set_mut = write_set.into_mut();
-
         // We should have a materialized delta for every delta in the output.
-        assert_eq!(delta_writes.len(), delta_change_set.len());
-
-        // Add the delta writes to the write set of the transaction.
-        delta_writes.into_iter().for_each(|item| {
-            debug_assert!(
-                delta_change_set.remove(&item.0).is_some(),
-                "Delta writes contain a key which does not exist in DeltaChangeSet."
-            );
-            write_set_mut.insert(item)
-        });
+        assert_eq!(
+            materialized_deltas.len(),
+            self.change_set().aggregator_delta_set().len()
+        );
+        debug_assert!(
+            materialized_deltas
+                .iter()
+                .all(|(k, _)| self.change_set().aggregator_delta_set().contains_key(k)),
+            "Materialized aggregator writes contain a key which does not exist in delta set."
+        );
+        self.change_set
+            .extend_aggregator_write_set(materialized_deltas.into_iter());
 
-        let write_set = write_set_mut
-            .freeze()
-            .expect("Freezing of WriteSet should succeed.");
+        let (vm_change_set, gas_used, status) = self.unpack();
+        let (write_set, events) = vm_change_set
+            .into_storage_change_set_unchecked()
+            .into_inner();
         TransactionOutput::new(write_set, events, gas_used, status)
     }
 }
diff --git a/aptos-move/aptos-vm-types/src/storage.rs b/aptos-move/aptos-vm-types/src/storage.rs
index dfc646051f942..cf0bdade7e105 100644
--- a/aptos-move/aptos-vm-types/src/storage.rs
+++ b/aptos-move/aptos-vm-types/src/storage.rs
@@ -374,7 +374,7 @@ impl CheckChangeSet for ChangeSetConfigs {
         const ERR: StatusCode = StatusCode::STORAGE_WRITE_LIMIT_REACHED;
 
         let mut write_set_size = 0;
-        for (key, op) in change_set.write_set() {
+        for (key, op) in change_set.write_set_iter() {
             if let Some(bytes) = op.bytes() {
                 let write_op_size = (bytes.len() + key.size()) as u64;
                 if write_op_size > self.max_bytes_per_write_op {
diff --git a/aptos-move/aptos-vm-types/src/tests/test_change_set.rs b/aptos-move/aptos-vm-types/src/tests/test_change_set.rs
index e5d102c9fc98c..13217c62bcb5e 100644
--- a/aptos-move/aptos-vm-types/src/tests/test_change_set.rs
+++ b/aptos-move/aptos-vm-types/src/tests/test_change_set.rs
@@ -4,23 +4,25 @@
 use crate::{
     change_set::VMChangeSet,
     tests::utils::{
-        build_change_set, contains_delta_op, contains_write_op, create, delete, get_delta_op,
-        get_write_op, key, modify, NoOpChangeSetChecker,
+        build_change_set, mock_add, mock_create, mock_delete, mock_modify, MockChangeSetChecker,
     },
 };
-use aptos_aggregator::delta_change_set::{delta_add, DeltaChangeSet};
-use aptos_types::write_set::WriteSetMut;
+use aptos_types::{
+    access_path::AccessPath,
+    state_store::state_key::StateKey,
+    transaction::ChangeSet as StorageChangeSet,
+    write_set::{WriteOp, WriteSetMut},
+};
 use claims::{assert_matches, assert_ok};
-use move_core_types::vm_status::{StatusCode, VMStatus};
-
-macro_rules! add {
-    ($v:expr) => {
-        // Limit doesn't matter here, so set it to be relatively high.
-        delta_add($v, 100000)
-    };
-}
+use move_core_types::{
+    account_address::AccountAddress,
+    ident_str,
+    language_storage::{ModuleId, StructTag},
+    vm_status::{StatusCode, VMStatus},
+};
+use std::collections::BTreeMap;
 
-/// Returns two change sets according tow specification:
+/// Testcases:
 /// ```text
 /// *--------------*----------------*----------------*-----------------*
 /// |   state key  |  change set 1  |  change set 2  |    squashed     |
@@ -54,213 +56,204 @@ macro_rules! add {
 /// |      23      |    +23         |    delete      |    delete       |
 /// *--------------*----------------*----------------*-----------------*
 /// ```
-fn build_change_sets_for_test() -> (VMChangeSet, VMChangeSet) {
-    // Create write sets and delta change sets.
-    let mut write_set_1 = WriteSetMut::default();
-    let mut write_set_2 = WriteSetMut::default();
-    let mut delta_change_set_1 = DeltaChangeSet::empty();
-    let mut delta_change_set_2 = DeltaChangeSet::empty();
 
-    // Populate sets according to the spec. Skip keys which lead to
-    // errors because we test them separately.
-    write_set_1.insert((key(0), create(0)));
-    write_set_1.insert((key(1), modify(1)));
-    write_set_1.insert((key(2), delete()));
-    write_set_2.insert((key(3), create(103)));
-    write_set_2.insert((key(4), modify(104)));
-    write_set_2.insert((key(5), delete()));
+macro_rules! write_set_1 {
+    ($d:ident) => {
+        vec![
+            mock_create(format!("0{}", $d), 0),
+            mock_modify(format!("1{}", $d), 1),
+            mock_delete(format!("2{}", $d)),
+            mock_create(format!("7{}", $d), 7),
+            mock_create(format!("8{}", $d), 8),
+            mock_modify(format!("10{}", $d), 10),
+            mock_modify(format!("11{}", $d), 11),
+            mock_delete(format!("12{}", $d)),
+        ]
+    };
+}
 
-    write_set_1.insert((key(7), create(7)));
-    write_set_2.insert((key(7), modify(107)));
-    write_set_1.insert((key(8), create(8)));
-    write_set_2.insert((key(8), delete()));
+macro_rules! write_set_2 {
+    ($d:ident) => {
+        vec![
+            mock_create(format!("3{}", $d), 103),
+            mock_modify(format!("4{}", $d), 104),
+            mock_delete(format!("5{}", $d)),
+            mock_modify(format!("7{}", $d), 107),
+            mock_delete(format!("8{}", $d)),
+            mock_modify(format!("10{}", $d), 110),
+            mock_delete(format!("11{}", $d)),
+            mock_create(format!("12{}", $d), 112),
+        ]
+    };
+}
 
-    write_set_1.insert((key(10), modify(10)));
-    write_set_2.insert((key(10), modify(110)));
-    write_set_1.insert((key(11), modify(111)));
-    write_set_2.insert((key(11), delete()));
-    write_set_1.insert((key(12), delete()));
-    write_set_2.insert((key(12), create(112)));
+macro_rules! expected_write_set {
+    ($d:ident) => {
+        BTreeMap::from([
+            mock_create(format!("0{}", $d), 0),
+            mock_modify(format!("1{}", $d), 1),
+            mock_delete(format!("2{}", $d)),
+            mock_create(format!("3{}", $d), 103),
+            mock_modify(format!("4{}", $d), 104),
+            mock_delete(format!("5{}", $d)),
+            mock_create(format!("7{}", $d), 107),
+            mock_modify(format!("10{}", $d), 110),
+            mock_delete(format!("11{}", $d)),
+            mock_modify(format!("12{}", $d), 112),
+        ])
+    };
+}
 
-    delta_change_set_1.insert((key(15), add!(15)));
-    delta_change_set_2.insert((key(16), add!(116)));
-    delta_change_set_1.insert((key(17), add!(17)));
-    delta_change_set_2.insert((key(17), add!(117)));
-    write_set_1.insert((key(18), create(18)));
-    delta_change_set_2.insert((key(18), add!(118)));
-    write_set_1.insert((key(19), modify(19)));
-    delta_change_set_2.insert((key(19), add!(119)));
+// Populate sets according to the spec. Skip keys which lead to
+// errors because we test them separately.
+fn build_change_sets_for_test() -> (VMChangeSet, VMChangeSet) {
+    let mut descriptor = "r";
+    let resource_write_set_1 = write_set_1!(descriptor);
+    descriptor = "m";
+    let module_write_set_1 = write_set_1!(descriptor);
+    let aggregator_write_set_1 = vec![mock_create("18a", 18), mock_modify("19a", 19)];
+    let aggregator_delta_set_1 = vec![
+        mock_add("15a", 15),
+        mock_add("17a", 17),
+        mock_add("22a", 22),
+        mock_add("23a", 23),
+    ];
+    let change_set_1 = build_change_set(
+        resource_write_set_1,
+        module_write_set_1,
+        aggregator_write_set_1,
+        aggregator_delta_set_1,
+    );
 
-    delta_change_set_1.insert((key(22), add!(22)));
-    write_set_2.insert((key(22), modify(122)));
-    delta_change_set_1.insert((key(23), add!(23)));
-    write_set_2.insert((key(23), delete()));
+    descriptor = "r";
+    let resource_write_set_2 = write_set_2!(descriptor);
+    descriptor = "m";
+    let module_write_set_2 = write_set_2!(descriptor);
+    let aggregator_write_set_2 = vec![mock_modify("22a", 122), mock_delete("23a")];
+    let aggregator_delta_set_2 = vec![
+        mock_add("16a", 116),
+        mock_add("17a", 117),
+        mock_add("18a", 118),
+        mock_add("19a", 119),
+    ];
+    let change_set_2 = build_change_set(
+        resource_write_set_2,
+        module_write_set_2,
+        aggregator_write_set_2,
+        aggregator_delta_set_2,
+    );
 
-    (
-        build_change_set(write_set_1, delta_change_set_1),
-        build_change_set(write_set_2, delta_change_set_2),
-    )
+    (change_set_1, change_set_2)
 }
 
 #[test]
 fn test_successful_squash() {
-    let (change_set_1, change_set_2) = build_change_sets_for_test();
-
-    // Check squash is indeed successful.
-    let res = change_set_1.squash(change_set_2, &NoOpChangeSetChecker);
-    let change_set = assert_ok!(res);
-
-    // create 0 + ___ = create 0
-    assert_eq!(get_write_op(&change_set, 0), create(0));
-    assert!(!contains_delta_op(&change_set, 0));
-
-    // modify 1 + ___ = modify 1
-    assert_eq!(get_write_op(&change_set, 1), modify(1));
-    assert!(!contains_delta_op(&change_set, 1));
-
-    // delete + ___ = delete
-    assert_eq!(get_write_op(&change_set, 2), delete());
-    assert!(!contains_delta_op(&change_set, 2));
-
-    // ___ + create 103 = create 103
-    assert_eq!(get_write_op(&change_set, 3), create(103));
-    assert!(!contains_delta_op(&change_set, 3));
-
-    // ___ + modify 103 = modify 103
-    assert_eq!(get_write_op(&change_set, 4), modify(104));
-    assert!(!contains_delta_op(&change_set, 4));
-
-    // ___ + delete = delete
-    assert_eq!(get_write_op(&change_set, 5), delete());
-    assert!(!contains_delta_op(&change_set, 5));
-
-    // create 7 + modify 107 = create 107
-    assert_eq!(get_write_op(&change_set, 7), create(107));
-    assert!(!contains_delta_op(&change_set, 7));
-
-    // create 8 + delete = ___
-    assert!(!contains_write_op(&change_set, 8));
-    assert!(!contains_delta_op(&change_set, 8));
-
-    // modify 10 + modify 110 = modify 110
-    assert_eq!(get_write_op(&change_set, 10), modify(110));
-    assert!(!contains_delta_op(&change_set, 10));
-
-    // modify 10 + delete = delete
-    assert_eq!(get_write_op(&change_set, 11), delete());
-    assert!(!contains_delta_op(&change_set, 11));
-
-    // delete + create 112 = create 112
-    assert_eq!(get_write_op(&change_set, 12), modify(112));
-    assert!(!contains_delta_op(&change_set, 12));
-
-    // +15 + ___ = +15
-    assert!(!contains_write_op(&change_set, 15));
-    assert_eq!(get_delta_op(&change_set, 15), add!(15));
-
-    // ___ + +116 = +116
-    assert!(!contains_write_op(&change_set, 16));
-    assert_eq!(get_delta_op(&change_set, 16), add!(116));
-
-    // +17 + +117 = +134
-    assert!(!contains_write_op(&change_set, 17));
-    assert_eq!(get_delta_op(&change_set, 17), add!(134));
-
-    // create 18 + +118 = create 136
-    assert_eq!(get_write_op(&change_set, 18), create(136));
-    assert!(!contains_delta_op(&change_set, 18));
+    let (mut change_set, additional_change_set) = build_change_sets_for_test();
+    assert_ok!(
+        change_set.squash_additional_change_set(additional_change_set, &MockChangeSetChecker)
+    );
 
-    // modify 19 + +119 = modify 138
-    assert_eq!(get_write_op(&change_set, 19), modify(138));
-    assert!(!contains_delta_op(&change_set, 19));
+    let mut descriptor = "r";
+    assert_eq!(
+        change_set.resource_write_set(),
+        &expected_write_set!(descriptor)
+    );
+    descriptor = "m";
+    assert_eq!(
+        change_set.module_write_set(),
+        &expected_write_set!(descriptor)
+    );
 
-    // +22 + modify 122 = modify 122
-    assert_eq!(get_write_op(&change_set, 22), modify(122));
-    assert!(!contains_delta_op(&change_set, 22));
+    let expected_aggregator_write_set = BTreeMap::from([
+        mock_create("18a", 136),
+        mock_modify("19a", 138),
+        mock_modify("22a", 122),
+        mock_delete("23a"),
+    ]);
+    let expected_aggregator_delta_set = BTreeMap::from([
+        mock_add("15a", 15),
+        mock_add("16a", 116),
+        mock_add("17a", 134),
+    ]);
+    assert_eq!(
+        change_set.aggregator_write_set(),
+        &expected_aggregator_write_set
+    );
+    assert_eq!(
+        change_set.aggregator_delta_set(),
+        &expected_aggregator_delta_set
+    );
+}
 
-    // +23 + delete = delete
-    assert_eq!(get_write_op(&change_set, 23), delete());
-    assert!(!contains_delta_op(&change_set, 23));
+macro_rules! assert_invariant_violation {
+    ($w1:ident, $w2:ident) => {
+        let check = |res: anyhow::Result<(), VMStatus>| {
+            assert_matches!(
+                res,
+                Err(VMStatus::Error {
+                    status_code: StatusCode::UNKNOWN_INVARIANT_VIOLATION_ERROR,
+                    sub_status: None,
+                    message: Some(_),
+                })
+            );
+        };
+
+        let mut cs1 = build_change_set($w1.clone(), vec![], vec![], vec![]);
+        let cs2 = build_change_set($w2.clone(), vec![], vec![], vec![]);
+        let res = cs1.squash_additional_change_set(cs2, &MockChangeSetChecker);
+        check(res);
+        let mut cs1 = build_change_set(vec![], $w1.clone(), vec![], vec![]);
+        let cs2 = build_change_set(vec![], $w2.clone(), vec![], vec![]);
+        let res = cs1.squash_additional_change_set(cs2, &MockChangeSetChecker);
+        check(res);
+        let mut cs1 = build_change_set(vec![], vec![], $w1.clone(), vec![]);
+        let cs2 = build_change_set(vec![], vec![], $w2.clone(), vec![]);
+        let res = cs1.squash_additional_change_set(cs2, &MockChangeSetChecker);
+        check(res);
+    };
 }
 
 #[test]
-fn test_unsuccessful_squash_1() {
-    let mut write_set_1 = WriteSetMut::default();
-    let mut write_set_2 = WriteSetMut::default();
-
+fn test_unsuccessful_squash_create_create() {
     // create 6 + create 106 throws an error
-    write_set_1.insert((key(6), create(6)));
-    write_set_2.insert((key(6), create(106)));
-
-    let change_set_1 = build_change_set(write_set_1, DeltaChangeSet::empty());
-    let change_set_2 = build_change_set(write_set_2, DeltaChangeSet::empty());
-    let res = change_set_1.squash(change_set_2, &NoOpChangeSetChecker);
-    assert_matches!(
-        res,
-        Err(VMStatus::Error {
-            status_code: StatusCode::UNKNOWN_INVARIANT_VIOLATION_ERROR,
-            sub_status: None,
-            message: Some(_),
-        })
-    );
+    let write_set_1 = vec![mock_create("6", 6)];
+    let write_set_2 = vec![mock_create("6", 106)];
+    assert_invariant_violation!(write_set_1, write_set_2);
 }
 
 #[test]
 fn test_unsuccessful_squash_modify_create() {
-    let mut write_set_1 = WriteSetMut::default();
-    let mut write_set_2 = WriteSetMut::default();
-
     // modify 9 + create 109 throws an error
-    write_set_1.insert((key(9), modify(9)));
-    write_set_2.insert((key(9), create(109)));
-
-    let change_set_1 = build_change_set(write_set_1, DeltaChangeSet::empty());
-    let change_set_2 = build_change_set(write_set_2, DeltaChangeSet::empty());
-    let res = change_set_1.squash(change_set_2, &NoOpChangeSetChecker);
-    assert_matches!(
-        res,
-        Err(VMStatus::Error {
-            status_code: StatusCode::UNKNOWN_INVARIANT_VIOLATION_ERROR,
-            sub_status: None,
-            message: Some(_),
-        })
-    );
+    let write_set_1 = vec![mock_modify("9", 9)];
+    let write_set_2 = vec![mock_create("9", 109)];
+    assert_invariant_violation!(write_set_1, write_set_2);
 }
 
 #[test]
 fn test_unsuccessful_squash_delete_modify() {
-    let mut write_set_1 = WriteSetMut::default();
-    let mut write_set_2 = WriteSetMut::default();
-
     // delete + modify 113 throws an error
-    write_set_1.insert((key(13), delete()));
-    write_set_2.insert((key(13), modify(113)));
-
-    let change_set_1 = build_change_set(write_set_1, DeltaChangeSet::empty());
-    let change_set_2 = build_change_set(write_set_2, DeltaChangeSet::empty());
-    let res = change_set_1.squash(change_set_2, &NoOpChangeSetChecker);
-    assert_matches!(
-        res,
-        Err(VMStatus::Error {
-            status_code: StatusCode::UNKNOWN_INVARIANT_VIOLATION_ERROR,
-            sub_status: None,
-            message: Some(_),
-        })
-    );
+    let write_set_1 = vec![mock_delete("13")];
+    let write_set_2 = vec![mock_modify("13", 113)];
+    assert_invariant_violation!(write_set_1, write_set_2);
 }
 
 #[test]
 fn test_unsuccessful_squash_delete_delete() {
-    let mut write_set_1 = WriteSetMut::default();
-    let mut write_set_2 = WriteSetMut::default();
-
     // delete + delete throws an error
-    write_set_1.insert((key(14), delete()));
-    write_set_2.insert((key(14), delete()));
+    let write_set_1 = vec![mock_delete("14")];
+    let write_set_2 = vec![mock_delete("14")];
+    assert_invariant_violation!(write_set_1, write_set_2);
+}
 
-    let change_set_1 = build_change_set(write_set_1, DeltaChangeSet::empty());
-    let change_set_2 = build_change_set(write_set_2, DeltaChangeSet::empty());
-    let res = change_set_1.squash(change_set_2, &NoOpChangeSetChecker);
+#[test]
+fn test_unsuccessful_squash_delete_delta() {
+    // delete + +120 throws an error
+    let aggregator_write_set_1 = vec![mock_delete("20")];
+    let aggregator_delta_set_2 = vec![mock_add("20", 120)];
+
+    let mut change_set = build_change_set(vec![], vec![], aggregator_write_set_1, vec![]);
+    let additional_change_set = build_change_set(vec![], vec![], vec![], aggregator_delta_set_2);
+    let res = change_set.squash_additional_change_set(additional_change_set, &MockChangeSetChecker);
     assert_matches!(
         res,
         Err(VMStatus::Error {
@@ -272,17 +265,14 @@ fn test_unsuccessful_squash_delete_delete() {
 }
 
 #[test]
-fn test_unsuccessful_squash_delete_delta() {
-    let mut write_set_1 = WriteSetMut::default();
-    let mut delta_change_set_2 = DeltaChangeSet::empty();
-
-    // delete + +120 throws an error
-    write_set_1.insert((key(20), delete()));
-    delta_change_set_2.insert((key(20), add!(120)));
+fn test_unsuccessful_squash_delta_create() {
+    // +21 + create 122 throws an error
+    let aggregator_delta_set_1 = vec![mock_add("21", 21)];
+    let aggregator_write_set_2 = vec![mock_create("21", 121)];
 
-    let change_set_1 = build_change_set(write_set_1, DeltaChangeSet::empty());
-    let change_set_2 = build_change_set(WriteSetMut::default(), delta_change_set_2);
-    let res = change_set_1.squash(change_set_2, &NoOpChangeSetChecker);
+    let mut change_set = build_change_set(vec![], vec![], vec![], aggregator_delta_set_1);
+    let additional_change_set = build_change_set(vec![], vec![], aggregator_write_set_2, vec![]);
+    let res = change_set.squash_additional_change_set(additional_change_set, &MockChangeSetChecker);
     assert_matches!(
         res,
         Err(VMStatus::Error {
@@ -294,21 +284,50 @@ fn test_unsuccessful_squash_delete_delta() {
 }
 
 #[test]
-fn test_unsuccessful_squash_delta_create() {
-    let mut write_set_2 = WriteSetMut::default();
-    let mut delta_change_set_1 = DeltaChangeSet::empty();
+fn test_roundtrip_to_storage_change_set() {
+    let test_struct_tag = StructTag {
+        address: AccountAddress::ONE,
+        module: ident_str!("foo").into(),
+        name: ident_str!("Foo").into(),
+        type_params: vec![],
+    };
+    let test_module_id = ModuleId::new(AccountAddress::ONE, ident_str!("bar").into());
 
-    // +21 + create 122 throws an error
-    delta_change_set_1.insert((key(21), add!(21)));
-    write_set_2.insert((key(21), create(121)));
+    let resource_key = StateKey::access_path(
+        AccessPath::resource_access_path(AccountAddress::ONE, test_struct_tag).unwrap(),
+    );
+    let module_key = StateKey::access_path(AccessPath::code_access_path(test_module_id));
+    let write_set = WriteSetMut::new(vec![
+        (resource_key, WriteOp::Deletion),
+        (module_key, WriteOp::Deletion),
+    ])
+    .freeze()
+    .unwrap();
+
+    let storage_change_set_before = StorageChangeSet::new(write_set, vec![]);
+    let change_set = assert_ok!(VMChangeSet::try_from_storage_change_set(
+        storage_change_set_before.clone(),
+        &MockChangeSetChecker
+    ));
+    let storage_change_set_after = assert_ok!(change_set.try_into_storage_change_set());
+    assert_eq!(storage_change_set_before, storage_change_set_after)
+}
 
-    let change_set_1 = build_change_set(WriteSetMut::default(), delta_change_set_1);
-    let change_set_2 = build_change_set(write_set_2, DeltaChangeSet::empty());
-    let res = change_set_1.squash(change_set_2, &NoOpChangeSetChecker);
+#[test]
+fn test_failed_conversion_to_change_set() {
+    let resource_write_set = vec![mock_delete("a")];
+    let aggregator_delta_set = vec![mock_add("b", 100)];
+    let change_set = build_change_set(resource_write_set, vec![], vec![], aggregator_delta_set);
+
+    // Unchecked conversion ignores deltas.
+    let storage_change_set = change_set.clone().into_storage_change_set_unchecked();
+    assert_eq!(storage_change_set.write_set().clone().into_mut().len(), 1);
+
+    let vm_status = change_set.try_into_storage_change_set();
     assert_matches!(
-        res,
+        vm_status,
         Err(VMStatus::Error {
-            status_code: StatusCode::UNKNOWN_INVARIANT_VIOLATION_ERROR,
+            status_code: StatusCode::DATA_FORMAT_ERROR,
             sub_status: None,
             message: Some(_),
         })
diff --git a/aptos-move/aptos-vm-types/src/tests/test_output.rs b/aptos-move/aptos-vm-types/src/tests/test_output.rs
index be78c3c903581..ef994b0341d2a 100644
--- a/aptos-move/aptos-vm-types/src/tests/test_output.rs
+++ b/aptos-move/aptos-vm-types/src/tests/test_output.rs
@@ -1,131 +1,119 @@
 // Copyright © Aptos Foundation
 // SPDX-License-Identifier: Apache-2.0
 
-use crate::tests::utils::{build_vm_output, create, key, modify};
-use aptos_aggregator::delta_change_set::{delta_add, serialize, DeltaChangeSet};
+use crate::{
+    output::VMOutput,
+    tests::utils::{as_state_key, build_vm_output, mock_add, mock_create, mock_modify},
+};
+use aptos_aggregator::delta_change_set::serialize;
 use aptos_language_e2e_tests::data_store::FakeDataStore;
-use aptos_types::write_set::WriteSetMut;
+use aptos_types::{
+    state_store::state_key::StateKey, transaction::TransactionOutput, write_set::WriteOp,
+};
 use claims::{assert_err, assert_matches, assert_ok};
 use move_core_types::vm_status::{AbortLocation, VMStatus};
+use std::collections::BTreeMap;
+
+fn assert_eq_outputs(vm_output: &VMOutput, txn_output: TransactionOutput) {
+    let vm_output_writes = &vm_output
+        .change_set()
+        .write_set_iter()
+        .map(|(k, v)| (k.clone(), v.clone()))
+        .collect::<BTreeMap<StateKey, WriteOp>>();
+
+    // A way to obtain a reference to the map inside a WriteSet.
+    let mut write_set_mut = txn_output.write_set().clone().into_mut();
+    let txn_output_writes = write_set_mut.as_inner_mut();
+
+    assert_eq!(vm_output_writes, txn_output_writes);
+    assert_eq!(vm_output.gas_used(), txn_output.gas_used());
+    assert_eq!(vm_output.status(), txn_output.status());
+}
 
 #[test]
 fn test_ok_output_equality_no_deltas() {
     let state_view = FakeDataStore::default();
-
-    // Suppose transaction has the following write set:
-    //   create 0
-    //   modify 1
-    // and has no deltas.
-    let mut write_set = WriteSetMut::default();
-    write_set.insert((key(0), create(0)));
-    write_set.insert((key(1), modify(1)));
-
-    // Construct the VMOutput.
-    let output = build_vm_output(write_set, DeltaChangeSet::empty());
+    let vm_output = build_vm_output(
+        vec![mock_create("0", 0)],
+        vec![mock_modify("1", 1)],
+        vec![mock_modify("2", 2)],
+        vec![],
+    );
 
     // Different ways to materialize deltas:
     //   1. `try_materialize` preserves the type and returns a result.
-    //   2. `into_transaction_output` changes the type and returns a result.
-    //   3. `output_with_delta_writes` changes the type and simply merges delta sets.
+    //   2. `try_into_transaction_output` changes the type and returns a result.
+    //   3. `into_transaction_output_with_materialized_deltas` changes the type and
+    //       simply merges materialized deltas.
+    let materialized_vm_output = assert_ok!(vm_output.clone().try_materialize(&state_view));
+    let txn_output_1 = assert_ok!(vm_output.clone().try_into_transaction_output(&state_view));
+    let txn_output_2 = vm_output
+        .clone()
+        .into_transaction_output_with_materialized_deltas(vec![]);
+
     // Because there are no deltas, we should not see any difference in write sets and
     // also all calls must succeed.
-    let vm_output = assert_ok!(output.clone().try_materialize(&state_view));
-    let txn_output_1 = assert_ok!(output.clone().into_transaction_output(&state_view));
-    let txn_output_2 = output.clone().output_with_delta_writes(vec![]);
-
-    // Check the output of `try_materialize`.
-    assert!(vm_output.delta_change_set().is_empty());
-    assert_eq!(vm_output.write_set(), output.write_set());
-    assert_eq!(vm_output.gas_used(), output.gas_used());
-    assert_eq!(vm_output.status(), output.status());
-
-    // Check the output of `into_transaction_output`.
-    assert_eq!(txn_output_1.write_set(), output.write_set());
-    assert_eq!(txn_output_1.gas_used(), output.gas_used());
-    assert_eq!(txn_output_1.status(), output.status());
-
-    // Check the output of `output_with_delta_writes`.
-    assert_eq!(txn_output_2.write_set(), output.write_set());
-    assert_eq!(txn_output_2.gas_used(), output.gas_used());
-    assert_eq!(txn_output_2.status(), output.status());
+    assert_eq!(&vm_output, &materialized_vm_output);
+    assert_eq_outputs(&vm_output, txn_output_1);
+    assert_eq_outputs(&vm_output, txn_output_2);
 }
 
 #[test]
 fn test_ok_output_equality_with_deltas() {
-    // Ensure that we have something (30 to be precise) stored at key 1.
+    let delta_key = "3";
     let mut state_view = FakeDataStore::default();
-    state_view.set_legacy(key(1), serialize(&30));
-
-    // This transaction has the following write set:
-    //   create 0
-    // and the following delta set:
-    //   add 20
-    let mut write_set = WriteSetMut::default();
-    let mut delta_change_set = DeltaChangeSet::empty();
-    write_set.insert((key(0), create(0)));
-    delta_change_set.insert((key(1), delta_add(20, 100)));
-
-    // Construct the VMOutput.
-    let output = build_vm_output(write_set, delta_change_set);
-
-    // Again, we test three different ways to materialize deltas. Here, we
-    // has a single delta which when materialized turns into 30 + 20 = 50.
-    let vm_output = assert_ok!(output.clone().try_materialize(&state_view));
-    let txn_output_1 = assert_ok!(output.clone().into_transaction_output(&state_view));
-    let txn_output_2 = output
+    state_view.set_legacy(as_state_key!(delta_key), serialize(&100));
+
+    let vm_output = build_vm_output(
+        vec![mock_create("0", 0)],
+        vec![mock_modify("1", 1)],
+        vec![mock_modify("2", 2)],
+        vec![mock_add(delta_key, 300)],
+    );
+
+    let materialized_vm_output = assert_ok!(vm_output.clone().try_materialize(&state_view));
+    let txn_output_1 = assert_ok!(vm_output.clone().try_into_transaction_output(&state_view));
+    let txn_output_2 = vm_output
         .clone()
-        .output_with_delta_writes(vec![(key(1), modify(50))]);
-
-    // Due to materialization, the write set should become:
-    // This transaction has the following write set:
-    //   create 0
-    //   modify 50
-    let expected_write_set = WriteSetMut::new(vec![(key(0), create(0)), (key(1), modify(50))])
-        .freeze()
-        .unwrap();
-
-    // Check the output of `try_materialize`. Note that all deltas have to
-    // be removed.
-    assert!(vm_output.delta_change_set().is_empty());
-    assert_eq!(vm_output.write_set(), &expected_write_set);
-    assert_eq!(vm_output.gas_used(), output.gas_used());
-    assert_eq!(vm_output.status(), output.status());
-
-    // Check the output of `into_transaction_output`.
-    assert_eq!(txn_output_1.write_set(), &expected_write_set);
-    assert_eq!(txn_output_1.gas_used(), output.gas_used());
-    assert_eq!(txn_output_1.status(), output.status());
-
-    // Check the output of `output_with_delta_writes`.
-    assert_eq!(txn_output_2.write_set(), &expected_write_set);
-    assert_eq!(txn_output_2.gas_used(), output.gas_used());
-    assert_eq!(txn_output_2.status(), output.status());
+        .into_transaction_output_with_materialized_deltas(vec![mock_modify("3", 400)]);
+
+    let expected_aggregator_write_set =
+        BTreeMap::from([mock_modify("2", 2), mock_modify("3", 400)]);
+    assert_eq!(
+        materialized_vm_output.change_set().resource_write_set(),
+        vm_output.change_set().resource_write_set()
+    );
+    assert_eq!(
+        materialized_vm_output.change_set().module_write_set(),
+        vm_output.change_set().module_write_set()
+    );
+    assert_eq!(
+        materialized_vm_output.change_set().aggregator_write_set(),
+        &expected_aggregator_write_set
+    );
+    assert!(materialized_vm_output
+        .change_set()
+        .aggregator_delta_set()
+        .is_empty());
+    assert_eq!(
+        vm_output.fee_statement(),
+        materialized_vm_output.fee_statement()
+    );
+    assert_eq!(vm_output.status(), materialized_vm_output.status());
+    assert_eq_outputs(&materialized_vm_output, txn_output_1);
+    assert_eq_outputs(&materialized_vm_output, txn_output_2);
 }
 
 #[test]
 fn test_err_output_equality_with_deltas() {
-    // Make sure that state view has a large enough value which overflows
-    // on delta materialization. Here we use the value of 90.
+    let delta_key = "3";
     let mut state_view = FakeDataStore::default();
-    state_view.set_legacy(key(1), serialize(&90));
-
-    // This transaction has the following write set:
-    //   create 0
-    // and the following delta set:
-    //   add 20
-    // Note that the last delta overflows when added to 90.
-    let mut write_set = WriteSetMut::default();
-    let mut delta_change_set = DeltaChangeSet::empty();
-    write_set.insert((key(0), create(0)));
-    delta_change_set.insert((key(1), delta_add(20, 100)));
-
-    // Construct the VMOutput.
-    let output = build_vm_output(write_set, delta_change_set);
-
-    // Testing `output_with_delta_writes` doesn't make sense here because
-    // when delta writes are constructed the error is caught.
-    let vm_status_1 = assert_err!(output.clone().try_materialize(&state_view));
-    let vm_status_2 = assert_err!(output.into_transaction_output(&state_view));
+    state_view.set_legacy(as_state_key!(delta_key), serialize(&900));
+
+    let vm_output = build_vm_output(vec![], vec![], vec![], vec![mock_add(delta_key, 300)]);
+
+    let vm_status_1 = assert_err!(vm_output.clone().try_materialize(&state_view));
+    let vm_status_2 = assert_err!(vm_output.try_into_transaction_output(&state_view));
 
     // Error should be consistent.
     assert_eq!(vm_status_1, vm_status_2);
diff --git a/aptos-move/aptos-vm-types/src/tests/utils.rs b/aptos-move/aptos-vm-types/src/tests/utils.rs
index a98da919f083c..a6dfd66e960be 100644
--- a/aptos-move/aptos-vm-types/src/tests/utils.rs
+++ b/aptos-move/aptos-vm-types/src/tests/utils.rs
@@ -2,94 +2,87 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use crate::{change_set::VMChangeSet, check_change_set::CheckChangeSet, output::VMOutput};
-use aptos_aggregator::delta_change_set::{serialize, DeltaChangeSet, DeltaOp};
+use aptos_aggregator::delta_change_set::{delta_add, serialize, DeltaOp};
 use aptos_types::{
     fee_statement::FeeStatement,
     state_store::state_key::StateKey,
     transaction::{ExecutionStatus, TransactionStatus},
-    write_set::{WriteOp, WriteSetMut},
+    write_set::WriteOp,
 };
 use move_core_types::vm_status::VMStatus;
+use std::collections::BTreeMap;
 
-/// A mock for testing. Always succeeds on checking a change set.
-pub(crate) struct NoOpChangeSetChecker;
+pub(crate) struct MockChangeSetChecker;
 
-impl CheckChangeSet for NoOpChangeSetChecker {
+impl CheckChangeSet for MockChangeSetChecker {
     fn check_change_set(&self, _change_set: &VMChangeSet) -> anyhow::Result<(), VMStatus> {
         Ok(())
     }
 }
 
-/// Returns a new state key for the given index. Allows to associate indices
-/// with state keys.
-pub(crate) fn key(id: u128) -> StateKey {
-    StateKey::raw(format!("key-{}", id).into_bytes())
+macro_rules! as_state_key {
+    ($k:ident) => {
+        StateKey::raw($k.to_string().into_bytes())
+    };
 }
+pub(crate) use as_state_key;
 
-/// Returns a new write op which creates an integer value.
-pub(crate) fn create(value: u128) -> WriteOp {
-    WriteOp::Creation(serialize(&value))
+macro_rules! as_bytes {
+    ($v:ident) => {
+        serialize(&$v)
+    };
 }
 
-/// Returns a new write op which modifies an integer value.
-pub(crate) fn modify(value: u128) -> WriteOp {
-    WriteOp::Modification(serialize(&value))
+pub(crate) fn mock_create(k: impl ToString, v: u128) -> (StateKey, WriteOp) {
+    (as_state_key!(k), WriteOp::Creation(as_bytes!(v)))
 }
 
-/// Returns a new write op which deletes a value.
-pub(crate) fn delete() -> WriteOp {
-    WriteOp::Deletion
+pub(crate) fn mock_modify(k: impl ToString, v: u128) -> (StateKey, WriteOp) {
+    (as_state_key!(k), WriteOp::Modification(as_bytes!(v)))
 }
 
-/// Returns a write op from the change set stored at state key corresponding
-/// to an id.
-pub(crate) fn get_write_op(change_set: &VMChangeSet, id: u128) -> WriteOp {
-    change_set.write_set().get(&key(id)).unwrap().clone()
+pub(crate) fn mock_delete(k: impl ToString) -> (StateKey, WriteOp) {
+    (as_state_key!(k), WriteOp::Deletion)
 }
 
-/// Returns true if there is a write op in the change set for the state key
-/// corresponding to an id.
-pub(crate) fn contains_write_op(change_set: &VMChangeSet, id: u128) -> bool {
-    change_set.write_set().get(&key(id)).is_some()
+pub(crate) fn mock_add(k: impl ToString, v: u128) -> (StateKey, DeltaOp) {
+    const DUMMY_LIMIT: u128 = 1000;
+    (as_state_key!(k), delta_add(v, DUMMY_LIMIT))
 }
 
-/// Returns a delta op from the change set stored at state key corresponding
-/// to an id.
-pub(crate) fn get_delta_op(change_set: &VMChangeSet, id: u128) -> DeltaOp {
-    *change_set.delta_change_set().get(&key(id)).unwrap()
-}
-
-/// Returns true if there is a delta op in the change set for the state key
-/// corresponding to an id.
-pub(crate) fn contains_delta_op(change_set: &VMChangeSet, id: u128) -> bool {
-    change_set.delta_change_set().get(&key(id)).is_some()
-}
-
-/// Returns a new change set built from writes and deltas.
 pub(crate) fn build_change_set(
-    write_set: WriteSetMut,
-    delta_change_set: DeltaChangeSet,
+    resource_write_set: impl IntoIterator<Item = (StateKey, WriteOp)>,
+    module_write_set: impl IntoIterator<Item = (StateKey, WriteOp)>,
+    aggregator_write_set: impl IntoIterator<Item = (StateKey, WriteOp)>,
+    aggregator_delta_set: impl IntoIterator<Item = (StateKey, DeltaOp)>,
 ) -> VMChangeSet {
     VMChangeSet::new(
-        write_set.freeze().unwrap(),
-        delta_change_set,
+        BTreeMap::from_iter(resource_write_set),
+        BTreeMap::from_iter(module_write_set),
+        BTreeMap::from_iter(aggregator_write_set),
+        BTreeMap::from_iter(aggregator_delta_set),
         vec![],
-        &NoOpChangeSetChecker,
+        &MockChangeSetChecker,
     )
     .unwrap()
 }
 
-/// Returns a new VMOutput built from writes and deltas. The output has always a
-/// success execution status and uses 100 gas units (values are not significant
-/// for testing purposes).
+// For testing, output has always a success execution status and uses 100 gas units.
 pub(crate) fn build_vm_output(
-    write_set: WriteSetMut,
-    delta_change_set: DeltaChangeSet,
+    resource_write_set: impl IntoIterator<Item = (StateKey, WriteOp)>,
+    module_write_set: impl IntoIterator<Item = (StateKey, WriteOp)>,
+    aggregator_write_set: impl IntoIterator<Item = (StateKey, WriteOp)>,
+    aggregator_delta_set: impl IntoIterator<Item = (StateKey, DeltaOp)>,
 ) -> VMOutput {
     const GAS_USED: u64 = 100;
     const STATUS: TransactionStatus = TransactionStatus::Keep(ExecutionStatus::Success);
     VMOutput::new(
-        build_change_set(write_set, delta_change_set),
+        build_change_set(
+            resource_write_set,
+            module_write_set,
+            aggregator_write_set,
+            aggregator_delta_set,
+        ),
         FeeStatement::new(GAS_USED, GAS_USED, 0, 0, 0),
         STATUS,
     )
diff --git a/aptos-move/aptos-vm/src/aptos_vm.rs b/aptos-move/aptos-vm/src/aptos_vm.rs
index 29c595cf7bb37..1863eea5a3915 100644
--- a/aptos-move/aptos-vm/src/aptos_vm.rs
+++ b/aptos-move/aptos-vm/src/aptos_vm.rs
@@ -18,7 +18,6 @@ use crate::{
     verifier, VMExecutor, VMValidator,
 };
 use anyhow::{anyhow, Result};
-use aptos_aggregator::delta_change_set::DeltaChangeSet;
 use aptos_block_executor::txn_commit_hook::NoOpTransactionCommitHook;
 use aptos_crypto::HashValue;
 use aptos_framework::natives::code::PublishRequest;
@@ -35,6 +34,7 @@ use aptos_types::{
     block_metadata::BlockMetadata,
     fee_statement::FeeStatement,
     on_chain_config::{new_epoch_event_key, FeatureFlag, TimedFeatureOverride},
+    state_store::state_key::StateKey,
     transaction::{
         analyzed_transaction::AnalyzedTransaction, EntryFunction, ExecutionError, ExecutionStatus,
         ModuleBundle, Multisig, MultisigTransactionPayload, SignatureCheckedTransaction,
@@ -42,7 +42,7 @@ use aptos_types::{
         VMValidatorResult, WriteSetPayload,
     },
     vm_status::{AbortLocation, StatusCode, VMStatus},
-    write_set::WriteSet,
+    write_set::WriteOp,
 };
 use aptos_utils::{aptos_try, return_on_failure};
 use aptos_vm_logging::{log_schema::AdapterLogSchema, speculative_error, speculative_log};
@@ -493,12 +493,12 @@ impl AptosVM {
     ) -> Result<RespawnedSession<'r, 'l>, VMStatus> {
         let change_set = session.finish(&mut (), change_set_configs)?;
 
-        for (key, op) in change_set.write_set() {
+        for (key, op) in change_set.write_set_iter() {
             gas_meter.charge_io_gas_for_write(key, op)?;
         }
 
         gas_meter.charge_storage_fee_for_all(
-            change_set.write_set().iter(),
+            change_set.write_set_iter(),
             change_set.events(),
             txn_data.transaction_size,
             txn_data.gas_unit_price,
@@ -1199,14 +1199,7 @@ impl AptosVM {
 
         match writeset_payload {
             WriteSetPayload::Direct(change_set) => {
-                let write_set = change_set.write_set().clone();
-                let events = change_set.events().to_vec();
-                VMChangeSet::new(
-                    write_set,
-                    DeltaChangeSet::empty(),
-                    events,
-                    &change_set_configs,
-                )
+                VMChangeSet::try_from_storage_change_set(change_set.clone(), &change_set_configs)
             },
             WriteSetPayload::Script { script, execute_as } => {
                 let mut tmp_session = self.0.new_session(resolver, session_id);
@@ -1239,14 +1232,14 @@ impl AptosVM {
         }
     }
 
-    fn read_writeset(
+    fn read_writeset<'a>(
         &self,
         state_view: &impl StateView,
-        write_set: &WriteSet,
+        write_set: impl IntoIterator<Item = (&'a StateKey, &'a WriteOp)>,
     ) -> Result<(), VMStatus> {
         // All Move executions satisfy the read-before-write property. Thus we need to read each
         // access path that the write set is going to update.
-        for (state_key, _) in write_set.iter() {
+        for (state_key, _) in write_set.into_iter() {
             state_view
                 .get_state_value_bytes(state_key)
                 .map_err(|_| VMStatus::error(StatusCode::STORAGE_ERROR, None))?;
@@ -1293,7 +1286,12 @@ impl AptosVM {
         )?;
 
         Self::validate_waypoint_change_set(&change_set, log_context)?;
-        self.read_writeset(resolver, change_set.write_set())?;
+        self.read_writeset(resolver, change_set.write_set_iter())?;
+        assert!(
+            change_set.aggregator_write_set().is_empty(),
+            "Waypoint change set should not have any aggregator writes."
+        );
+
         SYSTEM_TRANSACTIONS_EXECUTED.inc();
 
         let output = VMOutput::new(change_set, FeeStatement::zero(), VMStatus::Executed.into());
@@ -1368,7 +1366,7 @@ impl AptosVM {
         (
             vm_status,
             vm_output
-                .into_transaction_output(state_view)
+                .try_into_transaction_output(state_view)
                 .expect("Simulation cannot fail"),
         )
     }
@@ -1643,6 +1641,7 @@ impl VMAdapter for AptosVM {
     fn should_restart_execution(vm_output: &VMOutput) -> bool {
         let new_epoch_event_key = aptos_types::on_chain_config::new_epoch_event_key();
         vm_output
+            .change_set()
             .events()
             .iter()
             .any(|event| *event.key() == new_epoch_event_key)
diff --git a/aptos-move/aptos-vm/src/block_executor/mod.rs b/aptos-move/aptos-vm/src/block_executor/mod.rs
index 36676c1d4c604..f4a938dedd9ac 100644
--- a/aptos-move/aptos-vm/src/block_executor/mod.rs
+++ b/aptos-move/aptos-vm/src/block_executor/mod.rs
@@ -71,7 +71,7 @@ impl AptosTransactionOutput {
                 .lock()
                 .take()
                 .expect("Output must be set")
-                .output_with_delta_writes(vec![]),
+                .into_transaction_output_with_materialized_deltas(vec![]),
         }
     }
 }
@@ -91,8 +91,8 @@ impl BlockExecutorTransactionOutput for AptosTransactionOutput {
             .lock()
             .as_ref()
             .expect("Output to be set to get writes")
-            .write_set()
-            .iter()
+            .change_set()
+            .write_set_iter()
             .map(|(key, op)| (key.clone(), op.clone()))
             .collect()
     }
@@ -104,7 +104,8 @@ impl BlockExecutorTransactionOutput for AptosTransactionOutput {
             .lock()
             .as_ref()
             .expect("Output to be set to get deltas")
-            .delta_change_set()
+            .change_set()
+            .aggregator_delta_set()
             .iter()
             .map(|(key, op)| (key.clone(), *op))
             .collect()
@@ -120,7 +121,7 @@ impl BlockExecutorTransactionOutput for AptosTransactionOutput {
                         .lock()
                         .take()
                         .expect("Output must be set to combine with deltas")
-                        .output_with_delta_writes(delta_writes),
+                        .into_transaction_output_with_materialized_deltas(delta_writes),
                 )
                 .is_ok(),
             "Could not combine VMOutput with deltas"
diff --git a/aptos-move/aptos-vm/src/move_vm_ext/respawned_session.rs b/aptos-move/aptos-vm/src/move_vm_ext/respawned_session.rs
index ead93729f027d..72950c39bab40 100644
--- a/aptos-move/aptos-vm/src/move_vm_ext/respawned_session.rs
+++ b/aptos-move/aptos-vm/src/move_vm_ext/respawned_session.rs
@@ -63,18 +63,19 @@ impl<'r, 'l> RespawnedSession<'r, 'l> {
         mut self,
         change_set_configs: &ChangeSetConfigs,
     ) -> Result<VMChangeSet, VMStatus> {
-        let new_change_set = self.with_session_mut(|session| {
+        let additional_change_set = self.with_session_mut(|session| {
             session.take().unwrap().finish(&mut (), change_set_configs)
         })?;
-        let change_set = self.into_heads().state_view.change_set;
+        let mut change_set = self.into_heads().state_view.change_set;
         change_set
-            .squash(new_change_set, change_set_configs)
+            .squash_additional_change_set(additional_change_set, change_set_configs)
             .map_err(|_err| {
                 VMStatus::error(
                     StatusCode::UNKNOWN_INVARIANT_VIOLATION_ERROR,
                     err_msg("Failed to squash VMChangeSet"),
                 )
-            })
+            })?;
+        Ok(change_set)
     }
 }
 
@@ -98,13 +99,21 @@ impl<'r> TStateView for ChangeSetStateView<'r> {
     }
 
     fn get_state_value(&self, state_key: &Self::Key) -> Result<Option<StateValue>> {
-        match self.change_set.delta_change_set().get(state_key) {
+        // TODO: `get_state_value` should differentiate between different write types.
+        match self.change_set.aggregator_delta_set().get(state_key) {
             Some(delta_op) => Ok(delta_op
                 .try_into_write_op(self.base, state_key)?
                 .as_state_value()),
-            None => match self.change_set.write_set().get(state_key) {
-                Some(write_op) => Ok(write_op.as_state_value()),
-                None => self.base.get_state_value(state_key),
+            None => {
+                let cached_value = self
+                    .change_set
+                    .write_set_iter()
+                    .find(|(k, _)| *k == state_key)
+                    .map(|(_, v)| v);
+                match cached_value {
+                    Some(write_op) => Ok(write_op.as_state_value()),
+                    None => self.base.get_state_value(state_key),
+                }
             },
         }
     }
@@ -121,14 +130,11 @@ impl<'r> TStateView for ChangeSetStateView<'r> {
 #[cfg(test)]
 mod test {
     use super::*;
-    use aptos_aggregator::delta_change_set::{delta_add, deserialize, serialize, DeltaChangeSet};
+    use aptos_aggregator::delta_change_set::{delta_add, deserialize, serialize};
     use aptos_language_e2e_tests::data_store::FakeDataStore;
-    use aptos_types::{
-        state_store::table::TableHandle,
-        write_set::{WriteOp, WriteSetMut},
-    };
+    use aptos_types::write_set::WriteOp;
     use aptos_vm_types::check_change_set::CheckChangeSet;
-    use move_core_types::account_address::AccountAddress;
+    use std::collections::BTreeMap;
 
     /// A mock for testing. Always succeeds on checking a change set.
     struct NoOpChangeSetChecker;
@@ -139,64 +145,72 @@ mod test {
         }
     }
 
+    fn key(s: impl ToString) -> StateKey {
+        StateKey::raw(s.to_string().into_bytes())
+    }
+
+    fn write(v: u128) -> WriteOp {
+        WriteOp::Modification(serialize(&v))
+    }
+
+    fn read(view: &ChangeSetStateView, s: impl ToString) -> u128 {
+        let bytes = view.get_state_value(&key(s)).unwrap().unwrap().into_bytes();
+        deserialize(&bytes)
+    }
+
     #[test]
     fn test_change_set_state_view() {
-        let key1 = StateKey::table_item(
-            TableHandle(AccountAddress::ZERO),
-            String::from("test-key1").into_bytes(),
-        );
-        let key2 = StateKey::table_item(
-            TableHandle(AccountAddress::ZERO),
-            String::from("test-key2").into_bytes(),
-        );
-        let key3 = StateKey::raw(String::from("test-key3").into_bytes());
-
         let mut base_view = FakeDataStore::default();
-        base_view.set_legacy(key1.clone(), serialize(&150));
-        base_view.set_legacy(key2.clone(), serialize(&300));
-        base_view.set_legacy(key3.clone(), serialize(&500));
-
-        let delta_op = delta_add(5, 500);
-        let mut delta_change_set = DeltaChangeSet::empty();
-        delta_change_set.insert((key1.clone(), delta_op));
-
-        let write_set_ops = [(key2.clone(), WriteOp::Modification(serialize(&400)))];
-        let write_set = WriteSetMut::new(write_set_ops.into_iter())
-            .freeze()
-            .unwrap();
-        let change_set =
-            VMChangeSet::new(write_set, delta_change_set, vec![], &NoOpChangeSetChecker).unwrap();
-        let change_set_state_view = ChangeSetStateView::new(&base_view, change_set).unwrap();
-
-        assert_eq!(
-            deserialize(
-                change_set_state_view
-                    .get_state_value(&key1)
-                    .unwrap()
-                    .unwrap()
-                    .bytes()
-            ),
-            155
-        );
-        assert_eq!(
-            deserialize(
-                change_set_state_view
-                    .get_state_value(&key2)
-                    .unwrap()
-                    .unwrap()
-                    .bytes()
-            ),
-            400
-        );
-        assert_eq!(
-            deserialize(
-                change_set_state_view
-                    .get_state_value(&key3)
-                    .unwrap()
-                    .unwrap()
-                    .bytes()
-            ),
-            500
-        );
+        base_view.set_legacy(key("module_base"), serialize(&10));
+        base_view.set_legacy(key("module_both"), serialize(&20));
+
+        base_view.set_legacy(key("resource_base"), serialize(&30));
+        base_view.set_legacy(key("resource_both"), serialize(&40));
+
+        base_view.set_legacy(key("aggregator_base"), serialize(&50));
+        base_view.set_legacy(key("aggregator_both"), serialize(&60));
+        base_view.set_legacy(key("aggregator_delta_set"), serialize(&70));
+
+        let resource_write_set = BTreeMap::from([
+            (key("resource_both"), write(80)),
+            (key("resource_write_set"), write(90)),
+        ]);
+
+        let module_write_set = BTreeMap::from([
+            (key("module_both"), write(100)),
+            (key("module_write_set"), write(110)),
+        ]);
+
+        let aggregator_write_set = BTreeMap::from([
+            (key("aggregator_both"), write(120)),
+            (key("aggregator_write_set"), write(130)),
+        ]);
+
+        let aggregator_delta_set =
+            BTreeMap::from([(key("aggregator_delta_set"), delta_add(1, 1000))]);
+
+        let change_set = VMChangeSet::new(
+            resource_write_set,
+            module_write_set,
+            aggregator_write_set,
+            aggregator_delta_set,
+            vec![],
+            &NoOpChangeSetChecker,
+        )
+        .unwrap();
+        let view = ChangeSetStateView::new(&base_view, change_set).unwrap();
+
+        assert_eq!(read(&view, "module_base"), 10);
+        assert_eq!(read(&view, "module_both"), 100);
+        assert_eq!(read(&view, "module_write_set"), 110);
+
+        assert_eq!(read(&view, "resource_base"), 30);
+        assert_eq!(read(&view, "resource_both"), 80);
+        assert_eq!(read(&view, "resource_write_set"), 90);
+
+        assert_eq!(read(&view, "aggregator_base"), 50);
+        assert_eq!(read(&view, "aggregator_both"), 120);
+        assert_eq!(read(&view, "aggregator_write_set"), 130);
+        assert_eq!(read(&view, "aggregator_delta_set"), 71);
     }
 }
diff --git a/aptos-move/aptos-vm/src/move_vm_ext/session.rs b/aptos-move/aptos-vm/src/move_vm_ext/session.rs
index 62264d8174999..66684c634a1ef 100644
--- a/aptos-move/aptos-vm/src/move_vm_ext/session.rs
+++ b/aptos-move/aptos-vm/src/move_vm_ext/session.rs
@@ -5,10 +5,7 @@ use crate::{
     access_path_cache::AccessPathCache, data_cache::get_resource_group_from_metadata,
     move_vm_ext::MoveResolverExt, transaction_metadata::TransactionMetadata,
 };
-use aptos_aggregator::{
-    aggregator_extension::AggregatorID,
-    delta_change_set::{serialize, DeltaChangeSet},
-};
+use aptos_aggregator::{aggregator_extension::AggregatorID, delta_change_set::serialize};
 use aptos_crypto::{hash::CryptoHash, HashValue};
 use aptos_crypto_derive::{BCSCryptoHash, CryptoHasher};
 use aptos_framework::natives::{
@@ -22,7 +19,7 @@ use aptos_types::{
     on_chain_config::{CurrentTimeMicroseconds, Features, OnChainConfig},
     state_store::{state_key::StateKey, state_value::StateValueMetadata, table::TableHandle},
     transaction::SignatureCheckedTransaction,
-    write_set::{WriteOp, WriteSetMut},
+    write_set::WriteOp,
 };
 use aptos_vm_types::{change_set::VMChangeSet, storage::ChangeSetConfigs};
 use move_binary_format::errors::{Location, PartialVMError, VMResult};
@@ -320,8 +317,6 @@ impl<'r, 'l> SessionExt<'r, 'l> {
         ap_cache: &mut C,
         configs: &ChangeSetConfigs,
     ) -> Result<VMChangeSet, VMStatus> {
-        let mut write_set_mut = WriteSetMut::new(Vec::new());
-        let mut delta_change_set = DeltaChangeSet::empty();
         let mut new_slot_metadata: Option<StateValueMetadata> = None;
         if is_storage_slot_metadata_enabled {
             if let Some(payer) = new_slot_payer {
@@ -335,6 +330,11 @@ impl<'r, 'l> SessionExt<'r, 'l> {
             new_slot_metadata,
         };
 
+        let mut resource_write_set = BTreeMap::new();
+        let mut module_write_set = BTreeMap::new();
+        let mut aggregator_write_set = BTreeMap::new();
+        let mut aggregator_delta_set = BTreeMap::new();
+
         for (addr, account_changeset) in change_set.into_inner() {
             let (modules, resources) = account_changeset.into_inner();
             for (struct_tag, blob_op) in resources {
@@ -345,14 +345,14 @@ impl<'r, 'l> SessionExt<'r, 'l> {
                     configs.legacy_resource_creation_as_modification(),
                 )?;
 
-                write_set_mut.insert((state_key, op))
+                resource_write_set.insert(state_key, op);
             }
 
             for (name, blob_op) in modules {
                 let state_key =
                     StateKey::access_path(ap_cache.get_module_path(ModuleId::new(addr, name)));
                 let op = woc.convert(&state_key, blob_op, false)?;
-                write_set_mut.insert((state_key, op))
+                module_write_set.insert(state_key, op);
             }
         }
 
@@ -362,7 +362,7 @@ impl<'r, 'l> SessionExt<'r, 'l> {
                 let state_key =
                     StateKey::access_path(ap_cache.get_resource_group_path(addr, struct_tag));
                 let op = woc.convert(&state_key, blob_op, false)?;
-                write_set_mut.insert((state_key, op))
+                resource_write_set.insert(state_key, op);
             }
         }
 
@@ -370,7 +370,7 @@ impl<'r, 'l> SessionExt<'r, 'l> {
             for (key, value_op) in change.entries {
                 let state_key = StateKey::table_item(handle.into(), key);
                 let op = woc.convert(&state_key, value_op, false)?;
-                write_set_mut.insert((state_key, op))
+                resource_write_set.insert(state_key, op);
             }
         }
 
@@ -382,20 +382,18 @@ impl<'r, 'l> SessionExt<'r, 'l> {
             match change {
                 AggregatorChange::Write(value) => {
                     let write_op = woc.convert_aggregator_mod(&state_key, value)?;
-                    write_set_mut.insert((state_key, write_op));
+                    aggregator_write_set.insert(state_key, write_op);
+                },
+                AggregatorChange::Merge(delta_op) => {
+                    aggregator_delta_set.insert(state_key, delta_op);
                 },
-                AggregatorChange::Merge(delta_op) => delta_change_set.insert((state_key, delta_op)),
                 AggregatorChange::Delete => {
                     let write_op = woc.convert(&state_key, MoveStorageOp::Delete, false)?;
-                    write_set_mut.insert((state_key, write_op));
+                    aggregator_write_set.insert(state_key, write_op);
                 },
             }
         }
 
-        let write_set = write_set_mut
-            .freeze()
-            .map_err(|_| VMStatus::error(StatusCode::DATA_FORMAT_ERROR, None))?;
-
         let events = events
             .into_iter()
             .map(|(guid, seq_num, ty_tag, blob)| {
@@ -404,7 +402,14 @@ impl<'r, 'l> SessionExt<'r, 'l> {
                 Ok(ContractEvent::new(key, seq_num, ty_tag, blob))
             })
             .collect::<Result<Vec<_>, VMStatus>>()?;
-        VMChangeSet::new(write_set, delta_change_set, events, configs)
+        VMChangeSet::new(
+            resource_write_set,
+            module_write_set,
+            aggregator_write_set,
+            aggregator_delta_set,
+            events,
+            configs,
+        )
     }
 }
 
diff --git a/aptos-move/block-executor/src/proptest_types/baseline.rs b/aptos-move/block-executor/src/proptest_types/baseline.rs
index e357ba4d516d3..807b8fc7d5e8b 100644
--- a/aptos-move/block-executor/src/proptest_types/baseline.rs
+++ b/aptos-move/block-executor/src/proptest_types/baseline.rs
@@ -227,7 +227,7 @@ impl<V: Debug + Clone + PartialEq + Eq + TransactionWrite> BaselineOutput<V> {
                         .as_ref()
                         .expect("Aggregator failures not yet tested")
                         .iter()
-                        .zip(output.2.iter())
+                        .zip(output.read_results.iter())
                         .for_each(|(baseline_read, result_read)| {
                             baseline_read.assert_read_result(result_read)
                         });
@@ -236,7 +236,13 @@ impl<V: Debug + Clone + PartialEq + Eq + TransactionWrite> BaselineOutput<V> {
                         .as_ref()
                         .expect("Aggregator failures not yet tested")
                         .iter()
-                        .zip(output.3.get().expect("Delta writes must be set").iter())
+                        .zip(
+                            output
+                                .materialized_delta_writes
+                                .get()
+                                .expect("Delta writes must be set")
+                                .iter(),
+                        )
                         .for_each(|(baseline_delta_write, (_, result_delta_write))| {
                             assert_eq!(
                                 *baseline_delta_write,
@@ -249,14 +255,14 @@ impl<V: Debug + Clone + PartialEq + Eq + TransactionWrite> BaselineOutput<V> {
 
                 results.iter().skip(committed).for_each(|output| {
                     // Ensure the transaction is skipped based on the output.
-                    assert_eq!(output.0.len(), 0);
-                    assert_eq!(output.1.len(), 0);
-                    assert_eq!(output.2.len(), 0);
-                    assert_eq!(output.4, 0);
+                    assert!(output.writes.is_empty());
+                    assert!(output.deltas.is_empty());
+                    assert!(output.read_results.is_empty());
+                    assert_eq!(output.total_gas, 0);
 
                     // Implies that materialize_delta_writes was never called, as should
                     // be for skipped transactions.
-                    assert_none!(output.3.get());
+                    assert_none!(output.materialized_delta_writes.get());
                 });
             },
             Err(BlockExecutorError::UserError(idx)) => {
diff --git a/aptos-move/block-executor/src/proptest_types/types.rs b/aptos-move/block-executor/src/proptest_types/types.rs
index fd87fe73c18d4..0f6583b5769f8 100644
--- a/aptos-move/block-executor/src/proptest_types/types.rs
+++ b/aptos-move/block-executor/src/proptest_types/types.rs
@@ -277,8 +277,8 @@ impl Default for TransactionGenParams {
 // TODO: consider adding writes to reads (read-before-write). Similar behavior to the Move-VM
 // and may force more testing (since we check read results).
 impl<V: Into<Vec<u8>> + Arbitrary + Clone + Debug + Eq + Sync + Send> TransactionGen<V> {
-    // TODO: disentangle writes and deltas.
     fn writes_and_deltas_from_gen<K: Clone + Hash + Debug + Eq + Ord>(
+        // TODO: disentangle writes and deltas.
         universe: &[K],
         gen: Vec<Vec<(Index, V)>>,
         module_write_fn: &dyn Fn(usize) -> bool,
@@ -527,13 +527,13 @@ where
                         Err(_) => reads_result.push(None),
                     }
                 }
-                ExecutionStatus::Success(MockOutput(
-                    behavior.writes.clone(),
-                    behavior.deltas.clone(),
-                    reads_result,
-                    OnceCell::new(),
-                    behavior.gas,
-                ))
+                ExecutionStatus::Success(MockOutput {
+                    writes: behavior.writes.clone(),
+                    deltas: behavior.deltas.clone(),
+                    read_results: reads_result,
+                    materialized_delta_writes: OnceCell::new(),
+                    total_gas: behavior.gas,
+                })
             },
             MockTransaction::SkipRest => ExecutionStatus::SkipRest(MockOutput::skip_output()),
             MockTransaction::Abort => ExecutionStatus::Abort(txn_idx as usize),
@@ -542,14 +542,14 @@ where
 }
 
 #[derive(Debug)]
-// TODO: name members.
-pub(crate) struct MockOutput<K, V>(
-    pub(crate) Vec<(K, V)>,
-    pub(crate) Vec<(K, DeltaOp)>,
-    pub(crate) Vec<Option<Vec<u8>>>,
-    pub(crate) OnceCell<Vec<(K, WriteOp)>>,
-    pub(crate) u64,
-);
+pub(crate) struct MockOutput<K, V> {
+    // TODO: Split writes into resources & modules.
+    pub(crate) writes: Vec<(K, V)>,
+    pub(crate) deltas: Vec<(K, DeltaOp)>,
+    pub(crate) read_results: Vec<Option<Vec<u8>>>,
+    pub(crate) materialized_delta_writes: OnceCell<Vec<(K, WriteOp)>>,
+    pub(crate) total_gas: u64,
+}
 
 impl<K, V> TransactionOutput for MockOutput<K, V>
 where
@@ -559,32 +559,38 @@ where
     type Txn = MockTransaction<K, V>;
 
     fn get_writes(&self) -> Vec<(K, V)> {
-        self.0.clone()
+        self.writes.clone()
     }
 
     fn get_deltas(&self) -> Vec<(K, DeltaOp)> {
-        self.1.clone()
+        self.deltas.clone()
     }
 
     fn skip_output() -> Self {
-        Self(
-            /*writes = */ vec![],
-            /*deltas = */ vec![],
-            /*read results = */ vec![],
-            /*materialized delta writes = */ OnceCell::new(),
-            /*gas = */ 0,
-        )
+        Self {
+            writes: vec![],
+            deltas: vec![],
+            read_results: vec![],
+            materialized_delta_writes: OnceCell::new(),
+            total_gas: 0,
+        }
     }
 
     fn incorporate_delta_writes(&self, delta_writes: Vec<(K, WriteOp)>) {
-        assert_ok!(self.3.set(delta_writes));
+        assert_ok!(self.materialized_delta_writes.set(delta_writes));
     }
 
     fn fee_statement(&self) -> FeeStatement {
         // First argument is supposed to be total (not important for the test though).
         // Next two arguments are different kinds of execution gas that are counted
         // towards the block limit. We split the total into two pieces for these arguments.
-        // TODO: add variety to generating fee statement based on total gas (self.4).
-        FeeStatement::new(self.4, self.4 / 2, (self.4 + 1) / 2, 0, 0)
+        // TODO: add variety to generating fee statement based on total gas.
+        FeeStatement::new(
+            self.total_gas,
+            self.total_gas / 2,
+            (self.total_gas + 1) / 2,
+            0,
+            0,
+        )
     }
 }
diff --git a/aptos-move/block-executor/src/task.rs b/aptos-move/block-executor/src/task.rs
index 1c35469ec8b9d..ca04c70df3230 100644
--- a/aptos-move/block-executor/src/task.rs
+++ b/aptos-move/block-executor/src/task.rs
@@ -80,7 +80,7 @@ pub trait TransactionOutput: Send + Sync + Debug {
         <Self::Txn as Transaction>::Value,
     )>;
 
-    /// Get the deltas of a transaction from its output.
+    /// Get the aggregator deltas of a transaction from its output.
     fn get_deltas(&self) -> Vec<(<Self::Txn as Transaction>::Key, DeltaOp)>;
 
     /// Execution output for transactions that comes after SkipRest signal.
diff --git a/aptos-move/e2e-tests/src/executor.rs b/aptos-move/e2e-tests/src/executor.rs
index d14bcdb8d4f83..2b1c6da50afaf 100644
--- a/aptos-move/e2e-tests/src/executor.rs
+++ b/aptos-move/e2e-tests/src/executor.rs
@@ -526,7 +526,7 @@ impl FakeExecutor {
             )?;
 
         Ok((
-            output.into_transaction_output(self.get_state_view())?,
+            output.try_into_transaction_output(self.get_state_view())?,
             gas_profiler.finish(),
         ))
     }
@@ -700,7 +700,10 @@ impl FakeExecutor {
                     &ChangeSetConfigs::unlimited_at_gas_feature_version(LATEST_GAS_FEATURE_VERSION),
                 )
                 .expect("Failed to generate txn effects");
-            let (write_set, _delta_change_set, _events) = change_set.unpack();
+            let (write_set, _events) = change_set
+                .try_into_storage_change_set()
+                .expect("Failed to convert to ChangeSet")
+                .into_inner();
             write_set
         };
         self.data_store.add_write_set(&write_set);
@@ -751,7 +754,10 @@ impl FakeExecutor {
                     &ChangeSetConfigs::unlimited_at_gas_feature_version(LATEST_GAS_FEATURE_VERSION),
                 )
                 .expect("Failed to generate txn effects");
-            let (write_set, _delta_change_set, _events) = change_set.unpack();
+            let (write_set, _events) = change_set
+                .try_into_storage_change_set()
+                .expect("Failed to convert to ChangeSet")
+                .into_inner();
             write_set
         };
         self.data_store.add_write_set(&write_set);
@@ -794,7 +800,10 @@ impl FakeExecutor {
             )
             .expect("Failed to generate txn effects");
         // TODO: Support deltas in fake executor.
-        let (write_set, _delta_change_set, _events) = change_set.unpack();
+        let (write_set, _events) = change_set
+            .try_into_storage_change_set()
+            .expect("Failed to convert to ChangeSet")
+            .into_inner();
         Ok(write_set)
     }
 
diff --git a/aptos-move/e2e-testsuite/src/tests/failed_transaction_tests.rs b/aptos-move/e2e-testsuite/src/tests/failed_transaction_tests.rs
index e5a1df56ff8c2..21858b3df0ef7 100644
--- a/aptos-move/e2e-testsuite/src/tests/failed_transaction_tests.rs
+++ b/aptos-move/e2e-testsuite/src/tests/failed_transaction_tests.rs
@@ -8,8 +8,10 @@ use aptos_language_e2e_tests::{common_transactions::peer_to_peer_txn, executor::
 use aptos_memory_usage_tracker::MemoryTrackedGasMeter;
 use aptos_state_view::TStateView;
 use aptos_types::{
+    state_store::state_key::StateKey,
     transaction::ExecutionStatus,
     vm_status::{StatusCode, VMStatus},
+    write_set::WriteOp,
 };
 use aptos_vm::{data_cache::AsMoveResolver, transaction_metadata::TransactionMetadata, AptosVM};
 use aptos_vm_logging::log_schema::AdapterLogSchema;
@@ -56,7 +58,9 @@ fn failed_transaction_cleanup_test() {
         &log_context,
         &change_set_configs,
     );
-    assert!(!out1.write_set().is_empty());
+
+    let write_set: Vec<(&StateKey, &WriteOp)> = out1.change_set().write_set_iter().collect();
+    assert!(!write_set.is_empty());
     assert_eq!(out1.gas_used(), 90_000);
     assert!(!out1.status().is_discarded());
     assert_eq!(
diff --git a/aptos-move/vm-genesis/src/lib.rs b/aptos-move/vm-genesis/src/lib.rs
index 05b6b69560cbd..1e401597dacf7 100644
--- a/aptos-move/vm-genesis/src/lib.rs
+++ b/aptos-move/vm-genesis/src/lib.rs
@@ -140,7 +140,7 @@ pub fn encode_aptos_mainnet_genesis_transaction(
     emit_new_block_and_epoch_event(&mut session);
 
     let configs = ChangeSetConfigs::unlimited_at_gas_feature_version(LATEST_GAS_FEATURE_VERSION);
-    let cs1 = session.finish(&mut (), &configs).unwrap();
+    let mut change_set = session.finish(&mut (), &configs).unwrap();
 
     // Publish the framework, using a different session id, in case both scripts creates tables
     let state_view = GenesisStateView::new();
@@ -151,22 +151,24 @@ pub fn encode_aptos_mainnet_genesis_transaction(
     let id2 = HashValue::new(id2_arr);
     let mut session = move_vm.new_session(&data_cache, SessionId::genesis(id2));
     publish_framework(&mut session, framework);
-    let cs2 = session.finish(&mut (), &configs).unwrap();
-    let change_set = cs1.squash(cs2, &configs).unwrap();
-
-    let (write_set, delta_change_set, events) = change_set.unpack();
+    let additional_change_set = session.finish(&mut (), &configs).unwrap();
+    change_set
+        .squash_additional_change_set(additional_change_set, &configs)
+        .unwrap();
 
     // Publishing stdlib should not produce any deltas around aggregators and map to write ops and
     // not deltas. The second session only publishes the framework module bundle, which should not
     // produce deltas either.
     assert!(
-        delta_change_set.is_empty(),
+        change_set.aggregator_delta_set().is_empty(),
         "non-empty delta change set in genesis"
     );
+    assert!(!change_set.write_set_iter().any(|(_, op)| op.is_deletion()));
+    verify_genesis_write_set(change_set.events());
 
-    assert!(!write_set.iter().any(|(_, op)| op.is_deletion()));
-    verify_genesis_write_set(&events);
-    let change_set = ChangeSet::new(write_set, events);
+    let change_set = change_set
+        .try_into_storage_change_set()
+        .expect("Constructing a ChangeSet from VMChangeSet should always succeed at genesis");
     Transaction::GenesisTransaction(WriteSetPayload::Direct(change_set))
 }
 
@@ -248,7 +250,7 @@ pub fn encode_genesis_change_set(
     emit_new_block_and_epoch_event(&mut session);
 
     let configs = ChangeSetConfigs::unlimited_at_gas_feature_version(LATEST_GAS_FEATURE_VERSION);
-    let cs1 = session.finish(&mut (), &configs).unwrap();
+    let mut change_set = session.finish(&mut (), &configs).unwrap();
 
     let state_view = GenesisStateView::new();
     let data_cache = state_view.as_move_resolver();
@@ -259,22 +261,24 @@ pub fn encode_genesis_change_set(
     let id2 = HashValue::new(id2_arr);
     let mut session = move_vm.new_session(&data_cache, SessionId::genesis(id2));
     publish_framework(&mut session, framework);
-    let cs2 = session.finish(&mut (), &configs).unwrap();
-    let change_set = cs1.squash(cs2, &configs).unwrap();
-
-    let (write_set, delta_change_set, events) = change_set.unpack();
+    let additional_change_set = session.finish(&mut (), &configs).unwrap();
+    change_set
+        .squash_additional_change_set(additional_change_set, &configs)
+        .unwrap();
 
     // Publishing stdlib should not produce any deltas around aggregators and map to write ops and
     // not deltas. The second session only publishes the framework module bundle, which should not
     // produce deltas either.
     assert!(
-        delta_change_set.is_empty(),
+        change_set.aggregator_delta_set().is_empty(),
         "non-empty delta change set in genesis"
     );
 
-    assert!(!write_set.iter().any(|(_, op)| op.is_deletion()));
-    verify_genesis_write_set(&events);
-    ChangeSet::new(write_set, events)
+    assert!(!change_set.write_set_iter().any(|(_, op)| op.is_deletion()));
+    verify_genesis_write_set(change_set.events());
+    change_set
+        .try_into_storage_change_set()
+        .expect("Constructing a ChangeSet from VMChangeSet should always succeed at genesis")
 }
 
 fn validate_genesis_config(genesis_config: &GenesisConfiguration) {
diff --git a/aptos-move/writeset-transaction-generator/src/writeset_builder.rs b/aptos-move/writeset-transaction-generator/src/writeset_builder.rs
index 71bdb61da377b..348a89c6164a9 100644
--- a/aptos-move/writeset-transaction-generator/src/writeset_builder.rs
+++ b/aptos-move/writeset-transaction-generator/src/writeset_builder.rs
@@ -139,8 +139,8 @@ where
     };
 
     // Genesis never produces the delta change set.
-    assert!(change_set.delta_change_set().is_empty());
-
-    let (write_set, _delta_change_set, events) = change_set.unpack();
-    ChangeSet::new(write_set, events)
+    assert!(change_set.aggregator_delta_set().is_empty());
+    change_set
+        .try_into_storage_change_set()
+        .expect("Conversion from VMChangeSet into ChangeSet should always succeed")
 }
diff --git a/types/src/write_set.rs b/types/src/write_set.rs
index aaf3c5107a3f1..a6083da1f17b7 100644
--- a/types/src/write_set.rs
+++ b/types/src/write_set.rs
@@ -301,6 +301,10 @@ impl WriteSetMut {
         self.write_set.insert(item.0, item.1);
     }
 
+    pub fn extend(&mut self, write_ops: impl IntoIterator<Item = (StateKey, WriteOp)>) {
+        self.write_set.extend(write_ops);
+    }
+
     #[inline]
     pub fn is_empty(&self) -> bool {
         self.write_set.is_empty()