[Storage] Add get_state_value API for supporting tables in StateView

Closes: aptos-labs#344

api/src/context.rs

@@ -108,9 +108,10 @@ impl Context {
         account: AccountAddress,
         version: u64,
     ) -> Result<Option<AccountStateBlob>> {
-        let (state_value, _) = self
-            .db
-            .get_state_value_with_proof_by_version(StateKey::AccountAddressKey(account), version)?;
+        let (state_value, _) = self.db.get_state_value_with_proof_by_version(
+            &StateKey::AccountAddressKey(account),
+            version,
+        )?;
         Ok(state_value.map(AccountStateBlob::from))
     }
 

aptos-move/aptos-transactional-test-harness/src/diem_test_harness.rs

@@ -465,7 +465,7 @@ impl<'a> DiemTestAdapter<'a> {
         ));
         let account_blob = self
             .storage
-            .get(&account_access_path)
+            .get_by_access_path(&account_access_path)
             .unwrap()
             .ok_or_else(|| {
                 format_err!(
@@ -490,7 +490,7 @@ impl<'a> DiemTestAdapter<'a> {
 
         let balance_blob = self
             .storage
-            .get(&balance_access_path)
+            .get_by_access_path(&balance_access_path)
             .unwrap()
             .ok_or_else(|| {
                 format_err!(

aptos-move/aptos-validator-interface/src/lib.rs

@@ -15,6 +15,7 @@ use aptos_types::{
     contract_event::EventWithProof,
     event::EventKey,
     on_chain_config::ValidatorSet,
+    state_store::{state_key::StateKey, state_value::StateValue},
     transaction::{Transaction, Version},
 };
 use move_binary_format::file_format::CompiledModule;
@@ -27,10 +28,20 @@ pub trait AptosValidatorInterface: Sync {
         account: AccountAddress,
         version: Version,
     ) -> Result<Option<AccountState>>;
+
+    fn get_state_value_by_version(
+        &self,
+        state_key: &StateKey,
+        version: Version,
+    ) -> Result<Option<StateValue>>;
+
     fn get_events(&self, key: &EventKey, start_seq: u64, limit: u64)
         -> Result<Vec<EventWithProof>>;
+
     fn get_committed_transactions(&self, start: Version, limit: u64) -> Result<Vec<Transaction>>;
+
     fn get_latest_version(&self) -> Result<Version>;
+
     fn get_version_by_account_sequence(
         &self,
         account: AccountAddress,
@@ -109,7 +120,7 @@ impl<'a> DebuggerStateView<'a> {
 }
 
 impl<'a> StateView for DebuggerStateView<'a> {
-    fn get(&self, access_path: &AccessPath) -> Result<Option<Vec<u8>>> {
+    fn get_by_access_path(&self, access_path: &AccessPath) -> Result<Option<Vec<u8>>> {
         match self.version {
             None => Ok(None),
             Some(ver) => match self
@@ -122,6 +133,16 @@ impl<'a> StateView for DebuggerStateView<'a> {
         }
     }
 
+    fn get_state_value(&self, state_key: &StateKey) -> Result<Option<Vec<u8>>> {
+        match self.version {
+            None => Ok(None),
+            Some(ver) => Ok(self
+                .db
+                .get_state_value_by_version(state_key, ver)?
+                .map(|x| x.bytes)),
+        }
+    }
+
     fn is_genesis(&self) -> bool {
         false
     }

aptos-move/aptos-validator-interface/src/storage_interface.rs

@@ -9,7 +9,7 @@ use aptos_types::{
     account_state::AccountState,
     contract_event::EventWithProof,
     event::EventKey,
-    state_store::state_key::StateKey,
+    state_store::{state_key::StateKey, state_value::StateValue},
     transaction::{Transaction, Version},
 };
 use aptosdb::AptosDB;
@@ -36,12 +36,23 @@ impl AptosValidatorInterface for DBDebuggerInterface {
         version: Version,
     ) -> Result<Option<AccountState>> {
         self.0
-            .get_state_value_with_proof_by_version(StateKey::AccountAddressKey(account), version)?
+            .get_state_value_with_proof_by_version(&StateKey::AccountAddressKey(account), version)?
             .0
             .map(|s| AccountState::try_from(&s))
             .transpose()
     }
 
+    fn get_state_value_by_version(
+        &self,
+        state_key: &StateKey,
+        version: Version,
+    ) -> Result<Option<StateValue>> {
+        Ok(self
+            .0
+            .get_state_value_with_proof_by_version(state_key, version)?
+            .0)
+    }
+
     fn get_events(
         &self,
         key: &EventKey,

aptos-move/aptos-vm/src/adapter_common.rs

@@ -159,7 +159,11 @@ fn preload_cache(signature_verified_block: &[PreprocessedTransaction], data_view
     // want to fine tune the number of threads launched here in the future.
     addresses_to_preload
         .into_par_iter()
-        .map(|addr| data_view.get(&AccessPath::new(addr, Vec::new())).ok()?)
+        .map(|addr| {
+            data_view
+                .get_by_access_path(&AccessPath::new(addr, Vec::new()))
+                .ok()?
+        })
         .collect::<Vec<Option<Vec<u8>>>>();
 }
 

aptos-move/aptos-vm/src/aptos_vm.rs

@@ -478,7 +478,7 @@ impl AptosVM {
         // access path that the write set is going to update.
         for (ap, _) in write_set.iter() {
             state_view
-                .get(ap)
+                .get_by_access_path(ap)
                 .map_err(|_| VMStatus::Error(StatusCode::STORAGE_ERROR))?;
         }
         Ok(())

aptos-move/aptos-vm/src/data_cache.rs

@@ -10,6 +10,7 @@ use aptos_state_view::{StateView, StateViewId};
 use aptos_types::{
     access_path::AccessPath,
     on_chain_config::ConfigStorage,
+    state_store::state_key::StateKey,
     vm_status::StatusCode,
     write_set::{WriteOp, WriteSet},
 };
@@ -70,14 +71,14 @@ impl<'a, S: StateView> StateViewCache<'a, S> {
 
 impl<'block, S: StateView> StateView for StateViewCache<'block, S> {
     // Get some data either through the cache or the `StateView` on a cache miss.
-    fn get(&self, access_path: &AccessPath) -> anyhow::Result<Option<Vec<u8>>> {
+    fn get_by_access_path(&self, access_path: &AccessPath) -> anyhow::Result<Option<Vec<u8>>> {
         fail_point!("move_adapter::data_cache::get", |_| Err(format_err!(
             "Injected failure in data_cache::get"
         )));
 
         match self.data_map.get(access_path) {
             Some(opt_data) => Ok(opt_data.clone()),
-            None => match self.data_view.get(access_path) {
+            None => match self.data_view.get_by_access_path(access_path) {
                 Ok(remote_data) => Ok(remote_data),
                 // TODO: should we forward some error info?
                 Err(e) => {
@@ -98,6 +99,27 @@ impl<'block, S: StateView> StateView for StateViewCache<'block, S> {
         }
     }
 
+    fn get_state_value(&self, state_key: &StateKey) -> anyhow::Result<Option<Vec<u8>>> {
+        //TODO: Add a caching layer on this once the VM write set starts populating state_value changes.
+        match self.data_view.get_state_value(state_key) {
+            Ok(remote_data) => Ok(remote_data),
+            Err(e) => {
+                // create an AdapterLogSchema from the `data_view` in scope. This log_context
+                // does not carry proper information about the specific transaction and
+                // context, but this error is related to the given `StateView` rather
+                // than the transaction.
+                // Also this API does not make it easy to plug in a context
+                let log_context = AdapterLogSchema::new(self.data_view.id(), 0);
+                CRITICAL_ERRORS.inc();
+                error!(
+                    log_context,
+                    "[VM, StateView] Error getting data from storage for {:?}", state_key
+                );
+                Err(e)
+            }
+        }
+    }
+
     fn is_genesis(&self) -> bool {
         self.data_view.is_genesis()
     }
@@ -129,7 +151,7 @@ impl<'block, S: StateView> ResourceResolver for StateViewCache<'block, S> {
 
 impl<'block, S: StateView> ConfigStorage for StateViewCache<'block, S> {
     fn fetch_config(&self, access_path: AccessPath) -> Option<Vec<u8>> {
-        self.get(&access_path).ok()?
+        self.get_by_access_path(&access_path).ok()?
     }
 }
 
@@ -143,7 +165,7 @@ impl<'a, S: StateView> RemoteStorage<'a, S> {
 
     pub fn get(&self, access_path: &AccessPath) -> PartialVMResult<Option<Vec<u8>>> {
         self.0
-            .get(access_path)
+            .get_by_access_path(access_path)
             .map_err(|_| PartialVMError::new(StatusCode::STORAGE_ERROR))
     }
 }

aptos-move/aptos-vm/src/parallel_executor/storage_wrapper.rs

@@ -4,7 +4,7 @@
 use crate::data_cache::RemoteStorage;
 use aptos_parallel_executor::executor::MVHashMapView;
 use aptos_state_view::{StateView, StateViewId};
-use aptos_types::{access_path::AccessPath, write_set::WriteOp};
+use aptos_types::{access_path::AccessPath, state_store::state_key::StateKey, write_set::WriteOp};
 use move_binary_format::errors::VMError;
 use move_core_types::{
     account_address::AccountAddress,
@@ -35,17 +35,22 @@ impl<'a, S: StateView> StateView for VersionedView<'a, S> {
     }
 
     // Get some data either through the cache or the `StateView` on a cache miss.
-    fn get(&self, access_path: &AccessPath) -> anyhow::Result<Option<Vec<u8>>> {
+    fn get_by_access_path(&self, access_path: &AccessPath) -> anyhow::Result<Option<Vec<u8>>> {
         match self.hashmap_view.read(access_path) {
             Ok(Some(v)) => Ok(match v.as_ref() {
                 WriteOp::Value(w) => Some(w.clone()),
                 WriteOp::Deletion => None,
             }),
-            Ok(None) => self.base_view.get(access_path),
+            Ok(None) => self.base_view.get_by_access_path(access_path),
             Err(err) => Err(err),
         }
     }
 
+    fn get_state_value(&self, state_key: &StateKey) -> anyhow::Result<Option<Vec<u8>>> {
+        // TODO: Add a caching layer on this once the VM write set starts populating state_value changes.
+        self.base_view.get_state_value(state_key)
+    }
+
     fn is_genesis(&self) -> bool {
         self.base_view.is_genesis()
     }

aptos-move/e2e-tests/src/data_store.rs

@@ -9,6 +9,7 @@ use aptos_state_view::StateView;
 use aptos_types::{
     access_path::AccessPath,
     on_chain_config::ConfigStorage,
+    state_store::state_key::StateKey,
     transaction::ChangeSet,
     write_set::{WriteOp, WriteSet},
 };
@@ -37,13 +38,17 @@ pub static GENESIS_CHANGE_SET_FRESH: Lazy<ChangeSet> =
 /// `RemoteCache` is needed.
 #[derive(Debug, Default, Clone, Serialize, Deserialize)]
 pub struct FakeDataStore {
-    data: HashMap<AccessPath, Vec<u8>>,
+    account_data: HashMap<AccessPath, Vec<u8>>,
+    state_data: HashMap<StateKey, Vec<u8>>,
 }
 
 impl FakeDataStore {
     /// Creates a new `FakeDataStore` with the provided initial data.
     pub fn new(data: HashMap<AccessPath, Vec<u8>>) -> Self {
-        FakeDataStore { data }
+        FakeDataStore {
+            account_data: data,
+            state_data: HashMap::new(),
+        }
     }
 
     /// Adds a [`WriteSet`] to this data store.
@@ -64,14 +69,14 @@ impl FakeDataStore {
     ///
     /// Returns the previous data if the key was occupied.
     pub fn set(&mut self, access_path: AccessPath, data_blob: Vec<u8>) -> Option<Vec<u8>> {
-        self.data.insert(access_path, data_blob)
+        self.account_data.insert(access_path, data_blob)
     }
 
     /// Deletes a key from this data store.
     ///
     /// Returns the previous data if the key was occupied.
     pub fn remove(&mut self, access_path: &AccessPath) -> Option<Vec<u8>> {
-        self.data.remove(access_path)
+        self.account_data.remove(access_path)
     }
 
     /// Adds an [`AccountData`] to this data store.
@@ -90,26 +95,30 @@ impl FakeDataStore {
 
     /// Yields a reference to the internal data structure of the global state
     pub fn inner(&self) -> &HashMap<AccessPath, Vec<u8>> {
-        &self.data
+        &self.account_data
     }
 }
 
 impl ConfigStorage for FakeDataStore {
     fn fetch_config(&self, access_path: AccessPath) -> Option<Vec<u8>> {
-        StateView::get(self, &access_path).unwrap_or_default()
+        StateView::get_by_access_path(self, &access_path).unwrap_or_default()
     }
 }
 
 // This is used by the `execute_block` API.
 // TODO: only the "sync" get is implemented
 impl StateView for FakeDataStore {
-    fn get(&self, access_path: &AccessPath) -> Result<Option<Vec<u8>>> {
+    fn get_by_access_path(&self, access_path: &AccessPath) -> Result<Option<Vec<u8>>> {
         // Since the data is in-memory, it can't fail.
-        Ok(self.data.get(access_path).cloned())
+        Ok(self.account_data.get(access_path).cloned())
+    }
+
+    fn get_state_value(&self, state_key: &StateKey) -> Result<Option<Vec<u8>>> {
+        Ok(self.state_data.get(state_key).cloned())
     }
 
     fn is_genesis(&self) -> bool {
-        self.data.is_empty()
+        self.account_data.is_empty()
     }
 }
 

aptos-move/e2e-tests/src/executor.rs

@@ -277,7 +277,7 @@ impl FakeExecutor {
             *addr,
             DiemAccountResource::struct_tag(),
         ));
-        let data_blob = StateView::get(&self.data_store, &ap)
+        let data_blob = StateView::get_by_access_path(&self.data_store, &ap)
             .expect("account must exist in data store")
             .unwrap_or_else(|| panic!("Can't fetch account resource for {}", addr));
         bcs::from_bytes(data_blob.as_slice()).ok()
@@ -303,7 +303,7 @@ impl FakeExecutor {
         let currency_code_tag = type_tag_for_currency_code(balance_currency_code);
         let balance_resource_tag = BalanceResource::struct_tag_for_currency(currency_code_tag);
         let ap = AccessPath::resource_access_path(ResourceKey::new(*addr, balance_resource_tag));
-        StateView::get(&self.data_store, &ap)
+        StateView::get_by_access_path(&self.data_store, &ap)
             .unwrap_or_else(|_| panic!("account {:?} must exist in data store", addr))
             .map(|data_blob| {
                 bcs::from_bytes(data_blob.as_slice()).expect("Failure decoding balance resource")
@@ -448,7 +448,7 @@ impl FakeExecutor {
 
     /// Get the blob for the associated AccessPath
     pub fn read_from_access_path(&self, path: &AccessPath) -> Option<Vec<u8>> {
-        StateView::get(&self.data_store, path).unwrap()
+        StateView::get_by_access_path(&self.data_store, path).unwrap()
     }
 
     /// Verifies the given transaction by running it through the VM verifier.