[vm] Separation of resources, modules and aggregators (#9256)

TLDR; Cherrypicks some changes from #8227.

In Move VM we know what is a resource, what is a module. On top of
that adapter has aggregators. All these have been merged into a single
`WriteSet` previously.

Now, we split write sets into 3:
- Resources
- Modules
- Aggregators

As a result, the code was cleaned-up.

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:

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();