fix(core, graph): simplify working with identifiers

Author: isum

core/src/amp_subgraph/runner/latest_blocks.rs

@@ -2,16 +2,12 @@ use alloy::primitives::BlockNumber;
 use anyhow::anyhow;
 use arrow::array::RecordBatch;
 use futures::{future::try_join_all, stream::BoxStream, StreamExt, TryFutureExt};
-use graph::{
-    amp::{
-        client::ResponseBatch,
-        codec::{utils::block_number_decoder, Decoder},
-        common::Ident,
-        error::IsDeterministic,
-        manifest::DataSource,
-        Client,
-    },
-    cheap_clone::CheapClone,
+use graph::amp::{
+    client::ResponseBatch,
+    codec::{utils::block_number_decoder, Decoder},
+    error::IsDeterministic,
+    manifest::DataSource,
+    Client,
 };
 use itertools::Itertools;
 use slog::debug;
@@ -43,7 +39,7 @@ impl LatestBlocks {
                     .map(move |(j, table)| ((i, j), &data_source.source.dataset, table))
             })
             .flatten()
-            .unique_by(|(_, dataset, table)| (dataset.cheap_clone(), table.cheap_clone()))
+            .unique_by(|(_, dataset, table)| (dataset.to_string(), table.to_string()))
             .map(|(table_ptr, dataset, table)| {
                 latest_block(&cx, dataset, table)
                     .map_ok(move |latest_block| (table_ptr, latest_block))
@@ -127,8 +123,8 @@ fn indexing_completed(data_source: &DataSource, latest_synced_block: &Option<Blo
 
 async fn latest_block<AC>(
     cx: &Context<AC>,
-    dataset: &Ident,
-    table: &Ident,
+    dataset: &str,
+    table: &str,
 ) -> Result<BlockNumber, Error>
 where
     AC: Client,
@@ -148,8 +144,8 @@ where
 
 async fn latest_block_changed<AC>(
     cx: &Context<AC>,
-    dataset: &Ident,
-    table: &Ident,
+    dataset: &str,
+    table: &str,
     latest_block: BlockNumber,
 ) -> Result<(), Error>
 where

core/src/amp_subgraph/runner/reorg_handler.rs

@@ -4,7 +4,6 @@ use futures::{future::try_join_all, StreamExt, TryFutureExt};
 use graph::{
     amp::{
         client::{LatestBlockBeforeReorg, RequestMetadata, ResponseBatch, ResumeStreamingQuery},
-        common::Ident,
         Client,
     },
     blockchain::block_stream::FirehoseCursor,
@@ -124,8 +123,8 @@ where
 async fn detect_reorg<AC>(
     cx: &Context<AC>,
     network: &str,
-    dataset: &Ident,
-    table: &Ident,
+    dataset: &str,
+    table: &str,
     latest_synced_block_number: BlockNumber,
     latest_synced_block_hash: BlockHash,
 ) -> Result<Option<LatestBlockBeforeReorg>, Error>

graph/src/amp/codec/mod.rs

@@ -7,14 +7,16 @@ mod value_decoder;
 
 pub mod utils;
 
-use std::collections::{BTreeMap, HashMap};
+use std::{
+    collections::{BTreeMap, HashMap},
+    sync::Arc,
+};
 
 use anyhow::{anyhow, bail, Context, Result};
 use arrow::array::{Array, RecordBatch};
 
 use self::{list_decoder::ListDecoder, mapping_decoder::MappingDecoder, name_cache::NameCache};
 use crate::{
-    amp::common::Ident,
     data::{
         graphql::TypeExt,
         store::{Id, IdType, Value},
@@ -163,7 +165,7 @@ impl Codec {
             .fields()
             .into_iter()
             .zip(record_batch.columns())
-            .map(|(field, array)| Ok((self.ident(field.name())?, array.as_ref())))
+            .map(|(field, array)| Ok((self.ident(field.name()), array.as_ref())))
             .collect::<Result<HashMap<_, _>>>()?;
 
         let mut value_decoders = BTreeMap::new();
@@ -194,7 +196,7 @@ impl Codec {
     fn value_decoder<'a>(
         &mut self,
         field: &'a Field,
-        columns: &HashMap<Ident, &'a dyn Array>,
+        columns: &HashMap<Arc<str>, &'a dyn Array>,
     ) -> Result<Option<Box<dyn Decoder<Value> + 'a>>> {
         // VIDs are auto-generated
         if field.name.eq_ignore_ascii_case("vid") {
@@ -206,7 +208,7 @@ impl Codec {
             return Ok(None);
         }
 
-        let normalized_name = self.ident(&field.name)?;
+        let normalized_name = self.ident(&field.name);
         let array = match columns.get(&normalized_name) {
             Some(&array) => array,
             None => {
@@ -230,7 +232,7 @@ impl Codec {
         Ok(Some(decoder))
     }
 
-    fn ident(&mut self, name: impl AsRef<str>) -> Result<Ident> {
+    fn ident(&mut self, name: impl AsRef<str>) -> Arc<str> {
         self.name_cache.ident(name.as_ref())
     }
 }

graph/src/amp/codec/name_cache.rs

@@ -1,12 +1,12 @@
-use std::collections::HashMap;
+use std::{collections::HashMap, sync::Arc};
 
-use anyhow::Result;
+use inflector::Inflector;
 
-use crate::{amp::common::Ident, cheap_clone::CheapClone};
+use crate::cheap_clone::CheapClone;
 
-/// Caches identifiers that are used to match Arrow columns and subgraph entity fields.
+/// Normalizes and caches identifiers that are used to match Arrow columns and subgraph entity fields.
 pub(super) struct NameCache {
-    cache: HashMap<Box<str>, Ident>,
+    cache: HashMap<Box<str>, Arc<str>>,
 }
 
 impl NameCache {
@@ -17,18 +17,18 @@ impl NameCache {
         }
     }
 
-    /// Returns the identifier for the given name.
+    /// Normalizes and returns the identifier for the given name.
     ///
     /// If the identifier exists in the cache, returns the cached version.
-    /// Otherwise, creates a new identifier, caches it, and returns it.
-    pub(super) fn ident(&mut self, name: &str) -> Result<Ident> {
+    /// Otherwise, creates a new normalized identifier, caches it, and returns it.
+    pub(super) fn ident(&mut self, name: &str) -> Arc<str> {
         if let Some(ident) = self.cache.get(name) {
-            return Ok(ident.cheap_clone());
+            return ident.cheap_clone();
         }
 
-        let ident = Ident::new(name)?;
+        let ident: Arc<str> = name.to_camel_case().into();
         self.cache.insert(name.into(), ident.cheap_clone());
 
-        Ok(ident)
+        ident
     }
 }

graph/src/amp/common/ident.rs

@@ -1,7 +1,3 @@
-mod ident;
-
-pub use self::ident::Ident;
-
 pub(super) mod column_aliases {
     pub(in crate::amp) static BLOCK_NUMBER: &[&str] = &[
         "_block_num",   // Meta column present in all tables