[table info][1/4] separate indexer async v2 db from aptosdb (#11799)

* separate indexer async v2 db from aptosdb

* address comments

* remove unrelated changes

api/Cargo.toml

@@ -19,6 +19,7 @@ aptos-bcs-utils = { workspace = true }
 aptos-build-info = { workspace = true }
 aptos-config = { workspace = true }
 aptos-crypto = { workspace = true }
+aptos-db-indexer = { workspace = true }
 aptos-framework =  { workspace = true }
 aptos-gas-schedule = { workspace = true }
 aptos-global-constants = { workspace = true }

api/openapi-spec-generator/src/fake_context.rs

@@ -16,5 +16,6 @@ pub fn get_fake_context() -> Context {
         Arc::new(MockDbReaderWriter),
         mempool.ac_client,
         NodeConfig::default(),
+        None, /* table info reader */
     )
 }

api/src/accounts.rs

@@ -351,7 +351,10 @@ impl Account {
                     .latest_state_view_poem(&self.latest_ledger_info)?;
                 let converted_resources = state_view
                     .as_move_resolver()
-                    .as_converter(self.context.db.clone())
+                    .as_converter(
+                        self.context.db.clone(),
+                        self.context.table_info_reader.clone(),
+                    )
                     .try_into_resources(resources.iter().map(|(k, v)| (k.clone(), v.as_slice())))
                     .context("Failed to build move resource response from data in DB")
                     .map_err(|err| {
@@ -545,7 +548,10 @@ impl Account {
             })?;
 
         resolver
-            .as_converter(self.context.db.clone())
+            .as_converter(
+                self.context.db.clone(),
+                self.context.table_info_reader.clone(),
+            )
             .move_struct_fields(resource_type, &bytes)
             .context("Failed to convert move structs from storage")
             .map_err(|err| {

api/src/context.rs

@@ -18,6 +18,7 @@ use aptos_api_types::{
 };
 use aptos_config::config::{NodeConfig, RoleType};
 use aptos_crypto::HashValue;
+use aptos_db_indexer::table_info_reader::TableInfoReader;
 use aptos_gas_schedule::{AptosGasParameters, FromOnChainGasSchedule};
 use aptos_logger::{error, info, warn, Schema};
 use aptos_mempool::{MempoolClientRequest, MempoolClientSender, SubmissionStatus};
@@ -76,6 +77,7 @@ pub struct Context {
     gas_limit_cache: Arc<RwLock<GasLimitCache>>,
     view_function_stats: Arc<FunctionStats>,
     simulate_txn_stats: Arc<FunctionStats>,
+    pub table_info_reader: Option<Arc<dyn TableInfoReader>>,
 }
 
 impl std::fmt::Debug for Context {
@@ -90,6 +92,7 @@ impl Context {
         db: Arc<dyn DbReader>,
         mp_sender: MempoolClientSender,
         node_config: NodeConfig,
+        table_info_reader: Option<Arc<dyn TableInfoReader>>,
     ) -> Self {
         let (view_function_stats, simulate_txn_stats) = {
             let log_per_call_stats = node_config.api.periodic_function_stats_sec.is_some();
@@ -126,6 +129,7 @@ impl Context {
             })),
             view_function_stats,
             simulate_txn_stats,
+            table_info_reader,
         }
     }
 
@@ -661,7 +665,7 @@ impl Context {
 
         let state_view = self.latest_state_view_poem(ledger_info)?;
         let resolver = state_view.as_move_resolver();
-        let converter = resolver.as_converter(self.db.clone());
+        let converter = resolver.as_converter(self.db.clone(), self.table_info_reader.clone());
         let txns: Vec<aptos_api_types::Transaction> = data
             .into_iter()
             .map(|t| {
@@ -692,7 +696,7 @@ impl Context {
 
         let state_view = self.latest_state_view_poem(ledger_info)?;
         let resolver = state_view.as_move_resolver();
-        let converter = resolver.as_converter(self.db.clone());
+        let converter = resolver.as_converter(self.db.clone(), self.table_info_reader.clone());
         let txns: Vec<aptos_api_types::Transaction> = data
             .into_iter()
             .map(|t| {

api/src/events.rs

@@ -186,7 +186,10 @@ impl EventsApi {
                     .context
                     .latest_state_view_poem(&latest_ledger_info)?
                     .as_move_resolver()
-                    .as_converter(self.context.db.clone())
+                    .as_converter(
+                        self.context.db.clone(),
+                        self.context.table_info_reader.clone(),
+                    )
                     .try_into_versioned_events(&events)
                     .context("Failed to convert events from storage into response")
                     .map_err(|err| {

api/src/runtime.rs

@@ -10,6 +10,7 @@ use crate::{
 };
 use anyhow::Context as AnyhowContext;
 use aptos_config::config::{ApiConfig, NodeConfig};
+use aptos_db_indexer::table_info_reader::TableInfoReader;
 use aptos_logger::info;
 use aptos_mempool::MempoolClientSender;
 use aptos_storage_interface::DbReader;
@@ -34,11 +35,12 @@ pub fn bootstrap(
     chain_id: ChainId,
     db: Arc<dyn DbReader>,
     mp_sender: MempoolClientSender,
+    table_info_reader: Option<Arc<dyn TableInfoReader>>,
 ) -> anyhow::Result<Runtime> {
     let max_runtime_workers = get_max_runtime_workers(&config.api);
     let runtime = aptos_runtimes::spawn_named_runtime("api".into(), Some(max_runtime_workers));
 
-    let context = Context::new(chain_id, db, mp_sender, config.clone());
+    let context = Context::new(chain_id, db, mp_sender, config.clone(), table_info_reader);
 
     attach_poem_to_runtime(runtime.handle(), context.clone(), config, false)
         .context("Failed to attach poem to runtime")?;
@@ -339,6 +341,7 @@ mod tests {
             ChainId::test(),
             context.db.clone(),
             context.mempool.ac_client.clone(),
+            None,
         );
         assert!(ret.is_ok());
 

api/src/state.rs

@@ -315,7 +315,10 @@ impl StateApi {
             AcceptType::Json => {
                 let resource = state_view
                     .as_move_resolver()
-                    .as_converter(self.context.db.clone())
+                    .as_converter(
+                        self.context.db.clone(),
+                        self.context.table_info_reader.clone(),
+                    )
                     .try_into_resource(&resource_type, &bytes)
                     .context("Failed to deserialize resource data retrieved from DB")
                     .map_err(|err| {
@@ -421,7 +424,10 @@ impl StateApi {
             .state_view(ledger_version.map(|inner| inner.0))?;
 
         let resolver = state_view.as_move_resolver();
-        let converter = resolver.as_converter(self.context.db.clone());
+        let converter = resolver.as_converter(
+            self.context.db.clone(),
+            self.context.table_info_reader.clone(),
+        );
 
         // Convert key to lookup version for DB
         let vm_key = converter

api/src/tests/converter_test.rs

@@ -21,7 +21,7 @@ async fn test_value_conversion() {
 
     let state_view = context.latest_state_view();
     let resolver = state_view.as_move_resolver();
-    let converter = resolver.as_converter(context.db);
+    let converter = resolver.as_converter(context.db, None);
 
     assert_value_conversion(&converter, "u8", 1i32, VmMoveValue::U8(1));
     assert_value_conversion(&converter, "u64", "1", VmMoveValue::U64(1));

api/src/transactions.rs

@@ -762,7 +762,10 @@ impl TransactionsApi {
                         let timestamp =
                             self.context.get_block_timestamp(ledger_info, txn.version)?;
                         resolver
-                            .as_converter(self.context.db.clone())
+                            .as_converter(
+                                self.context.db.clone(),
+                                self.context.table_info_reader.clone(),
+                            )
                             .try_into_onchain_transaction(timestamp, txn)
                             .context("Failed to convert on chain transaction to Transaction")
                             .map_err(|err| {
@@ -774,7 +777,10 @@ impl TransactionsApi {
                             })?
                     },
                     TransactionData::Pending(txn) => resolver
-                        .as_converter(self.context.db.clone())
+                        .as_converter(
+                            self.context.db.clone(),
+                            self.context.table_info_reader.clone(),
+                        )
                         .try_into_pending_transaction(*txn)
                         .context("Failed to convert on pending transaction to Transaction")
                         .map_err(|err| {
@@ -946,7 +952,10 @@ impl TransactionsApi {
                 .context
                 .latest_state_view_poem(ledger_info)?
                 .as_move_resolver()
-                .as_converter(self.context.db.clone())
+                .as_converter(
+                    self.context.db.clone(),
+                    self.context.table_info_reader.clone(),
+                )
                 .try_into_signed_transaction_poem(data.0, self.context.chain_id())
                 .context("Failed to create SignedTransaction from SubmitTransactionRequest")
                 .map_err(|err| {
@@ -1025,7 +1034,7 @@ impl TransactionsApi {
                 .map(|(index, txn)| {
                     self.context
                         .latest_state_view_poem(ledger_info)?.as_move_resolver()
-                        .as_converter(self.context.db.clone())
+                        .as_converter(self.context.db.clone(), self.context.table_info_reader.clone())
                         .try_into_signed_transaction_poem(txn, self.context.chain_id())
                         .context(format!("Failed to create SignedTransaction from SubmitTransactionRequest at position {}", index))
                         .map_err(|err| {
@@ -1117,7 +1126,7 @@ impl TransactionsApi {
 
                     // We provide the pending transaction so that users have the hash associated
                     let pending_txn = resolver
-                            .as_converter(self.context.db.clone())
+                            .as_converter(self.context.db.clone(), self.context.table_info_reader.clone())
                             .try_into_pending_transaction_poem(txn)
                             .context("Failed to build PendingTransaction from mempool response, even though it said the request was accepted")
                             .map_err(|err| SubmitTransactionError::internal_with_code(
@@ -1350,7 +1359,10 @@ impl TransactionsApi {
         let state_view = self.context.latest_state_view_poem(&ledger_info)?;
         let resolver = state_view.as_move_resolver();
         let raw_txn: RawTransaction = resolver
-            .as_converter(self.context.db.clone())
+            .as_converter(
+                self.context.db.clone(),
+                self.context.table_info_reader.clone(),
+            )
             .try_into_raw_transaction_poem(request.transaction, self.context.chain_id())
             .context("The given transaction is invalid")
             .map_err(|err| {

api/src/view_function.rs

@@ -97,7 +97,7 @@ fn view_request(
         ViewFunctionRequest::Json(data) => {
             let resolver = state_view.as_move_resolver();
             resolver
-                .as_converter(context.db.clone())
+                .as_converter(context.db.clone(), context.table_info_reader.clone())
                 .convert_view_function(data.0)
                 .map_err(|err| {
                     BasicErrorWith404::bad_request_with_code(
@@ -171,7 +171,7 @@ fn view_request(
         AcceptType::Json => {
             let resolver = state_view.as_move_resolver();
             let return_types = resolver
-                .as_converter(context.db.clone())
+                .as_converter(context.db.clone(), context.table_info_reader.clone())
                 .function_return_types(&view_function)
                 .and_then(|tys| {
                     tys.into_iter()
@@ -191,7 +191,7 @@ fn view_request(
                 .zip(return_types.into_iter())
                 .map(|(v, ty)| {
                     resolver
-                        .as_converter(context.db.clone())
+                        .as_converter(context.db.clone(), context.table_info_reader.clone())
                         .try_into_move_value(&ty, &v)
                 })
                 .collect::<anyhow::Result<Vec<_>>>()

api/test-context/src/test_context.rs

@@ -130,7 +130,6 @@ pub fn new_test_context(
                 false, /* indexer */
                 BUFFERED_STATE_TARGET_ITEMS,
                 DEFAULT_MAX_NUM_NODES_PER_LRU_CACHE_SHARD,
-                false, /* indexer async v2 */
             )
             .unwrap(),
         )
@@ -146,6 +145,7 @@ pub fn new_test_context(
         db.clone(),
         mempool.ac_client.clone(),
         node_config.clone(),
+        None, /* table info reader */
     );
 
     // Configure the testing depending on which API version we're testing.

api/types/Cargo.toml

@@ -16,6 +16,7 @@ rust-version = { workspace = true }
 anyhow = { workspace = true }
 aptos-config = { workspace = true }
 aptos-crypto = { workspace = true }
+aptos-db-indexer = { workspace = true }
 aptos-framework = { workspace = true }
 aptos-logger = { workspace = true }
 aptos-openapi = { workspace = true }
@@ -35,3 +36,4 @@ poem-openapi = { workspace = true }
 poem-openapi-derive = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
+