flat: add a way to check for existence without reading out the value (#…

…10510)

core/store/src/flat/chunk_view.rs

@@ -45,6 +45,10 @@ impl FlatStorageChunkView {
         self.flat_storage.get_value(&self.block_hash, key)
     }
 
+    pub fn contains_key(&self, key: &[u8]) -> Result<bool, crate::StorageError> {
+        self.flat_storage.contains_key(&self.block_hash, key)
+    }
+
     pub fn iter_flat_state_entries<'a>(
         &'a self,
         from: Option<&[u8]>,

core/store/src/flat/delta.rs

@@ -155,8 +155,8 @@ impl CachedFlatStateChanges {
         std::mem::size_of::<CryptoHash>() + std::mem::size_of::<Option<ValueRef>>();
 
     /// Returns `Some(Option<ValueRef>)` from delta for the given key. If key is not present, returns None.
-    pub(crate) fn get(&self, key: &[u8]) -> Option<Option<ValueRef>> {
-        self.0.get(&hash(key)).cloned()
+    pub(crate) fn get(&self, key: &[u8]) -> Option<&Option<ValueRef>> {
+        self.0.get(&hash(key))
     }
 
     /// Returns number of all entries.

core/store/src/flat/storage.rs

@@ -304,7 +304,7 @@ impl FlatStorage {
             let changes = guard.get_block_changes(block_hash)?;
             match changes.get(key) {
                 Some(value_ref) => {
-                    return Ok(value_ref.map(|value_ref| FlatStateValue::Ref(value_ref)));
+                    return Ok(value_ref.clone().map(|value_ref| FlatStateValue::Ref(value_ref)));
                 }
                 None => {}
             };
@@ -314,6 +314,30 @@ impl FlatStorage {
         Ok(value)
     }
 
+    /// Same as `get_value()?.is_some()`, but avoids reading out the value.
+    pub fn contains_key(
+        &self,
+        block_hash: &CryptoHash,
+        key: &[u8],
+    ) -> Result<bool, crate::StorageError> {
+        let guard = self.0.read().expect(super::POISONED_LOCK_ERR);
+        let blocks_to_head =
+            guard.get_blocks_to_head(block_hash).map_err(|e| StorageError::from(e))?;
+        for block_hash in blocks_to_head.iter() {
+            // If we found a key in changes, we can return a value because it is the most recent key update.
+            let changes = guard.get_block_changes(block_hash)?;
+            match changes.get(key) {
+                Some(value_ref) => return Ok(value_ref.is_some()),
+                None => {}
+            };
+        }
+
+        let db_key = store_helper::encode_flat_state_db_key(guard.shard_uid, key);
+        Ok(guard.store.exists(crate::DBCol::FlatState, &db_key).map_err(|err| {
+            FlatStorageError::StorageInternalError(format!("failed to read FlatState value: {err}"))
+        })?)
+    }
+
     /// Update the head of the flat storage, including updating the flat state
     /// in memory and on disk and updating the flat state to reflect the state
     /// at the new head. If updating to given head is not possible, returns an