[cherry-pick][1.12] schemadb: make iterator lazy #13341 (#13349)

* schemadb: make iterator lazy

Meaning, do not advance underlying raw iter early
Sometimes the second advance on the state value CF is exetremely costly
due to deletions by the pruner.

* [consensus] add missing check for commit cert

consensus/consensus-types/src/pipeline/commit_decision.rs

@@ -3,7 +3,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use crate::common::Round;
-use anyhow::Context;
+use anyhow::{ensure, Context};
 use aptos_types::{ledger_info::LedgerInfoWithSignatures, validator_verifier::ValidatorVerifier};
 use serde::{Deserialize, Serialize};
 use std::fmt::{Debug, Display, Formatter};
@@ -48,6 +48,10 @@ impl CommitDecision {
     /// Verifies that the signatures carried in the message forms a valid quorum,
     /// and then verifies the signature.
     pub fn verify(&self, validator: &ValidatorVerifier) -> anyhow::Result<()> {
+        ensure!(
+            !self.ledger_info.commit_info().is_ordered_only(),
+            "Unexpected ordered only commit into"
+        );
         // We do not need to check the author because as long as the signature tree
         // is valid, the message should be valid.
         self.ledger_info

consensus/src/pipeline/execution_client.rs

@@ -377,6 +377,10 @@ impl TExecutionClient for ExecutionProxyClient {
             Err(anyhow::anyhow!("Injected error in sync_to").into())
         });
 
+        if target.commit_info().is_ordered_only() {
+            return Err(anyhow::anyhow!("Sync to ordered only target").into());
+        }
+
         let (reset_tx_to_rand_manager, reset_tx_to_buffer_manager) = {
             let handle = self.handle.read();
             (

storage/schemadb/src/iterator.rs

@@ -12,11 +12,19 @@ pub enum ScanDirection {
     Backward,
 }
 
+enum Status {
+    Initialized,
+    DoneSeek,
+    Advancing,
+    Invalid,
+}
+
 /// DB Iterator parameterized on [`Schema`] that seeks with [`Schema::Key`] and yields
 /// [`Schema::Key`] and [`Schema::Value`]
 pub struct SchemaIterator<'a, S> {
     db_iter: rocksdb::DBRawIterator<'a>,
     direction: ScanDirection,
+    status: Status,
     phantom: PhantomData<S>,
 }
 
@@ -28,6 +36,7 @@ where
         SchemaIterator {
             db_iter,
             direction,
+            status: Status::Initialized,
             phantom: PhantomData,
         }
     }
@@ -38,6 +47,7 @@ where
             .with_label_values(&[S::COLUMN_FAMILY_NAME, "seek_to_first"])
             .start_timer();
         self.db_iter.seek_to_first();
+        self.status = Status::DoneSeek;
     }
 
     /// Seeks to the last key.
@@ -46,6 +56,7 @@ where
             .with_label_values(&[S::COLUMN_FAMILY_NAME, "seek_to_last"])
             .start_timer();
         self.db_iter.seek_to_last();
+        self.status = Status::DoneSeek;
     }
 
     /// Seeks to the first key whose binary representation is equal to or greater than that of the
@@ -59,6 +70,7 @@ where
             .start_timer();
         let key = <SK as SeekKeyCodec<S>>::encode_seek_key(seek_key)?;
         self.db_iter.seek(&key);
+        self.status = Status::DoneSeek;
         Ok(())
     }
 
@@ -75,6 +87,7 @@ where
             .start_timer();
         let key = <SK as SeekKeyCodec<S>>::encode_seek_key(seek_key)?;
         self.db_iter.seek_for_prev(&key);
+        self.status = Status::DoneSeek;
         Ok(())
     }
 
@@ -83,8 +96,19 @@ where
             .with_label_values(&[S::COLUMN_FAMILY_NAME])
             .start_timer();
 
+        if let Status::Advancing = self.status {
+            match self.direction {
+                ScanDirection::Forward => self.db_iter.next(),
+                ScanDirection::Backward => self.db_iter.prev(),
+            }
+        } else {
+            self.status = Status::Advancing;
+        }
+
         if !self.db_iter.valid() {
             self.db_iter.status()?;
+            // advancing an invalid raw iter results in seg fault
+            self.status = Status::Invalid;
             return Ok(None);
         }
 
@@ -97,11 +121,6 @@ where
         let key = <S::Key as KeyCodec<S>>::decode_key(raw_key);
         let value = <S::Value as ValueCodec<S>>::decode_value(raw_value);
 
-        match self.direction {
-            ScanDirection::Forward => self.db_iter.next(),
-            ScanDirection::Backward => self.db_iter.prev(),
-        }
-
         Ok(Some((key?, value?)))
     }
 }