[indexer] Improve perf of get_dynamic_field_object_id (MystenLabs#11002)

## Description 

- Use derive_dynamic_field_id for O(1) access rather than O(n) traversal

## Test Plan 

- localnet tests 

---
If your changes are not user-facing and not a breaking change, you can
skip the following section. Otherwise, please indicate what changed, and
then add to the Release Notes section as highlighted during the release
process.

### Type of Change (Check all that apply)

- [ ] user-visible impact
- [ ] breaking change for a client SDKs
- [ ] breaking change for FNs (FN binary must upgrade)
- [ ] breaking change for validators or node operators (must upgrade
binaries)
- [ ] breaking change for on-chain data layout
- [ ] necessitate either a data wipe or data migration

### Release notes

crates/sui-core/src/authority.rs

@@ -24,6 +24,7 @@ use prometheus::{
 };
 use serde::de::DeserializeOwned;
 use serde::Serialize;
+use sui_types::TypeTag;
 use tap::TapFallible;
 use tokio::sync::mpsc::unbounded_channel;
 use tokio::sync::oneshot;
@@ -2268,10 +2269,11 @@ impl AuthorityState {
     pub fn get_dynamic_field_object_id(
         &self,
         owner: ObjectID,
-        name: &DynamicFieldName,
+        name_type: TypeTag,
+        name_bcs_bytes: &[u8],
     ) -> SuiResult<Option<ObjectID>> {
         if let Some(indexes) = &self.indexes {
-            indexes.get_dynamic_field_object_id(owner, name)
+            indexes.get_dynamic_field_object_id(owner, name_type, name_bcs_bytes)
         } else {
             Err(SuiError::IndexStoreNotAvailable)
         }

crates/sui-json-rpc/src/indexer_api.rs

@@ -1,20 +1,22 @@
 // Copyright (c) Mysten Labs, Inc.
 // SPDX-License-Identifier: Apache-2.0
 
-use std::str::FromStr;
 use std::sync::Arc;
 
-use anyhow::anyhow;
+use anyhow::{anyhow, Context};
 use async_trait::async_trait;
 use futures::Stream;
 use jsonrpsee::core::error::SubscriptionClosed;
 use jsonrpsee::core::RpcResult;
 use jsonrpsee::types::SubscriptionResult;
 use jsonrpsee::{RpcModule, SubscriptionSink};
+use move_bytecode_utils::layout::TypeLayoutBuilder;
 use serde::Serialize;
+use sui_json::SuiJsonValue;
+use sui_types::MOVE_STDLIB_ADDRESS;
 use tracing::{debug, warn};
 
-use move_core_types::language_storage::TypeTag;
+use move_core_types::language_storage::{StructTag, TypeTag};
 use mysten_metrics::spawn_monitored_task;
 use sui_core::authority::AuthorityState;
 use sui_json_rpc_types::{
@@ -23,7 +25,7 @@ use sui_json_rpc_types::{
     SuiTransactionBlockResponse, SuiTransactionBlockResponseQuery, TransactionBlocksPage,
 };
 use sui_open_rpc::Module;
-use sui_types::base_types::{ObjectID, SuiAddress};
+use sui_types::base_types::{ObjectID, SuiAddress, STD_UTF8_MODULE_NAME, STD_UTF8_STRUCT_NAME};
 use sui_types::digests::TransactionDigest;
 use sui_types::dynamic_field::DynamicFieldName;
 use sui_types::event::EventID;
@@ -39,7 +41,6 @@ const NAME_SERVICE_REVERSE_LOOKUP_KEY: &str = "reverse";
 const NAME_SERVICE_VALUE: &str = "value";
 const NAME_SERVICE_ID: &str = "id";
 const NAME_SERVICE_MARKER: &str = "marker";
-const STRING_TYPE_TAG: &str = "0x1::string::String";
 
 pub fn spawn_subscription<S, T>(mut sink: SubscriptionSink, rx: S)
 where
@@ -255,9 +256,16 @@ impl<R: ReadApiServer> IndexerApiServer for IndexerApi<R> {
         parent_object_id: ObjectID,
         name: DynamicFieldName,
     ) -> RpcResult<SuiObjectResponse> {
+        let DynamicFieldName {
+            type_: name_type,
+            value,
+        } = name.clone();
+        let layout = TypeLayoutBuilder::build_with_types(&name_type, &self.state.database)?;
+        let sui_json_value = SuiJsonValue::new(value)?;
+        let name_bcs_value = sui_json_value.to_bcs_bytes(&layout)?;
         let id = self
             .state
-            .get_dynamic_field_object_id(parent_object_id, &name)
+            .get_dynamic_field_object_id(parent_object_id, name_type, &name_bcs_value)
             .map_err(|e| anyhow!("{e}"))?
             .ok_or_else(|| {
                 anyhow!("Cannot find dynamic field [{name:?}] for object [{parent_object_id}].")
@@ -270,16 +278,22 @@ impl<R: ReadApiServer> IndexerApiServer for IndexerApi<R> {
     async fn resolve_name_service_address(&self, name: String) -> RpcResult<SuiAddress> {
         let dynmaic_field_table_object_id =
             self.get_name_service_dynamic_field_table_object_id(/* reverse_lookup */ false)?;
-        // NOTE: 0x1::string::String is the type tag of fields in dynmaic_field_table
-        let ns_type_tag = TypeTag::from_str(STRING_TYPE_TAG)?;
-        let ns_dynamic_field_name = DynamicFieldName {
-            type_: ns_type_tag,
-            value: name.clone().into(),
-        };
+        // NOTE: 0x1::string::String is the type tag of fields in dynamic_field_table
+        let name_type_tag = TypeTag::Struct(Box::new(StructTag {
+            address: MOVE_STDLIB_ADDRESS,
+            module: STD_UTF8_MODULE_NAME.to_owned(),
+            name: STD_UTF8_STRUCT_NAME.to_owned(),
+            type_params: vec![],
+        }));
+        let name_bcs_value = bcs::to_bytes(&name).context("Unable to serialize name")?;
         // record of the input `name`
         let record_object_id = self
             .state
-            .get_dynamic_field_object_id(dynmaic_field_table_object_id, &ns_dynamic_field_name)
+            .get_dynamic_field_object_id(
+                dynmaic_field_table_object_id,
+                name_type_tag,
+                &name_bcs_value,
+            )
             .map_err(|e| {
                 anyhow!(
                     "Read name service dynamic field table failed with error: {:?}",
@@ -331,15 +345,15 @@ impl<R: ReadApiServer> IndexerApiServer for IndexerApi<R> {
     ) -> RpcResult<Page<String, ObjectID>> {
         let dynmaic_field_table_object_id =
             self.get_name_service_dynamic_field_table_object_id(/* reverse_lookup */ true)?;
-        let addr_type_tag = TypeTag::Address;
-        let ns_dynamic_field_name = DynamicFieldName {
-            type_: addr_type_tag,
-            value: address.to_string().into(),
-        };
-
+        let name_type_tag = TypeTag::Address;
+        let name_bcs_value = bcs::to_bytes(&address).context("Unable to serialize address")?;
         let addr_object_id = self
             .state
-            .get_dynamic_field_object_id(dynmaic_field_table_object_id, &ns_dynamic_field_name)
+            .get_dynamic_field_object_id(
+                dynmaic_field_table_object_id,
+                name_type_tag,
+                &name_bcs_value,
+            )
             .map_err(|e| {
                 anyhow!(
                     "Read name service reverse dynamic field table failed with error: {:?}",

crates/sui-storage/src/indexes.rs

@@ -25,7 +25,7 @@ use sui_types::base_types::{
 };
 use sui_types::base_types::{ObjectInfo, ObjectRef};
 use sui_types::digests::TransactionEventsDigest;
-use sui_types::dynamic_field::{DynamicFieldInfo, DynamicFieldName};
+use sui_types::dynamic_field::{self, DynamicFieldInfo};
 use sui_types::error::{SuiError, SuiResult};
 use sui_types::messages::TransactionEvents;
 use sui_types::object::Owner;
@@ -983,21 +983,48 @@ impl IndexStore {
     pub fn get_dynamic_field_object_id(
         &self,
         object: ObjectID,
-        name: &DynamicFieldName,
+        name_type: TypeTag,
+        name_bcs_bytes: &[u8],
     ) -> SuiResult<Option<ObjectID>> {
         debug!(?object, "get_dynamic_field_object_id");
-        Ok(self
+        let dynamic_field_id =
+            dynamic_field::derive_dynamic_field_id(object, &name_type, name_bcs_bytes).map_err(
+                |e| {
+                    SuiError::Unknown(format!(
+                        "Unable to generate dynamic field id. Got error: {e:?}"
+                    ))
+                },
+            )?;
+
+        if self
             .tables
             .dynamic_field_index
-            .iter()
-            // The object id 0 is the smallest possible
-            .skip_to(&(object, ObjectID::ZERO))?
-            .find(|((object_owner, _), info)| {
-                object_owner == &object
-                    && info.name.type_ == name.type_
-                    && info.name.value == name.value
-            })
-            .map(|(_, object_info)| object_info.object_id))
+            .contains_key(&(object, dynamic_field_id))?
+        {
+            return Ok(Some(dynamic_field_id));
+        }
+
+        let dynamic_object_field_struct = DynamicFieldInfo::dynamic_object_field_wrapper(name_type);
+        let dynamic_object_field_type = TypeTag::Struct(Box::new(dynamic_object_field_struct));
+        let dynamic_object_field_id = dynamic_field::derive_dynamic_field_id(
+            object,
+            &dynamic_object_field_type,
+            name_bcs_bytes,
+        )
+        .map_err(|e| {
+            SuiError::Unknown(format!(
+                "Unable to generate dynamic field id. Got error: {e:?}"
+            ))
+        })?;
+        if let Some(info) = self
+            .tables
+            .dynamic_field_index
+            .get(&(object, dynamic_object_field_id))?
+        {
+            return Ok(Some(info.object_id));
+        }
+
+        Ok(None)
     }
 
     pub fn get_owner_objects(

crates/sui-types/src/dynamic_field.rs

@@ -30,6 +30,9 @@ use std::fmt::{Display, Formatter};
 const DYNAMIC_FIELD_MODULE_NAME: &IdentStr = ident_str!("dynamic_field");
 const DYNAMIC_FIELD_FIELD_STRUCT_NAME: &IdentStr = ident_str!("Field");
 
+const DYNAMIC_OBJECT_FIELD_MODULE_NAME: &IdentStr = ident_str!("dynamic_object_field");
+const DYNAMIC_OBJECT_FIELD_WRAPPER_STRUCT_NAME: &IdentStr = ident_str!("Wrapper");
+
 /// Rust version of the Move sui::dynamic_field::Field type
 #[derive(Clone, Serialize, Deserialize, Debug)]
 pub struct Field<N, V> {
@@ -96,6 +99,15 @@ impl DynamicFieldInfo {
         }
     }
 
+    pub fn dynamic_object_field_wrapper(key: TypeTag) -> StructTag {
+        StructTag {
+            address: SUI_FRAMEWORK_ADDRESS,
+            name: DYNAMIC_OBJECT_FIELD_MODULE_NAME.to_owned(),
+            module: DYNAMIC_OBJECT_FIELD_WRAPPER_STRUCT_NAME.to_owned(),
+            type_params: vec![key],
+        }
+    }
+
     pub fn try_extract_field_name(tag: &StructTag, type_: &DynamicFieldType) -> SuiResult<TypeTag> {
         match (type_, tag.type_params.first()) {
             (DynamicFieldType::DynamicField, Some(name_type)) => Ok(name_type.clone()),
@@ -217,8 +229,8 @@ pub fn is_dynamic_object(move_struct: &MoveStruct) -> bool {
     match move_struct {
         MoveStruct::WithTypes { type_, .. } => {
             matches!(&type_.type_params[0], TypeTag::Struct(tag) if tag.address == SUI_FRAMEWORK_ADDRESS
-        && tag.module.as_str() == "dynamic_object_field"
-        && tag.name.as_str() == "Wrapper")
+        && tag.module.as_ident_str() == DYNAMIC_OBJECT_FIELD_MODULE_NAME
+        && tag.name.as_ident_str() == DYNAMIC_OBJECT_FIELD_WRAPPER_STRUCT_NAME)
         }
         _ => false,
     }