graph, node: move normalize_sql_ident to shared location and normalize context idents at config load time

Author: isum

graph/src/amp/manifest/data_source/raw.rs

@@ -8,7 +8,6 @@ use anyhow::anyhow;
 use arrow::{array::RecordBatch, datatypes::Schema};
 use futures03::future::try_join_all;
 use itertools::Itertools;
-use lazy_regex::regex_is_match;
 use semver::Version;
 use serde::Deserialize;
 use slog::{debug, Logger};
@@ -22,7 +21,9 @@ use crate::{
             auto_block_hash_decoder, auto_block_number_decoder, auto_block_timestamp_decoder,
         },
         error::IsDeterministic,
-        sql::{BlockRangeQueryBuilder, ContextQuery, ValidQuery},
+        sql::{
+            normalize_sql_ident, validate_ident, BlockRangeQueryBuilder, ContextQuery, ValidQuery,
+        },
     },
     components::link_resolver::{LinkResolver, LinkResolverContext},
     data::subgraph::DeploymentHash,
@@ -80,7 +81,8 @@ impl RawDataSource {
         let logger = logger.new(slog::o!("data_source" => name.clone()));
         debug!(logger, "Resolving data source");
 
-        validate_ident(&name).map_err(|e| e.source_context("invalid `name`"))?;
+        validate_ident(&name)
+            .map_err(|e| Error::InvalidValue(e).source_context("invalid `name`"))?;
         Self::validate_kind(kind)?;
 
         let source = source
@@ -380,7 +382,8 @@ impl RawAbi {
     ) -> Result<Abi, Error> {
         let Self { name, file } = self;
 
-        validate_ident(&name).map_err(|e| e.source_context("invalid `name`"))?;
+        validate_ident(&name)
+            .map_err(|e| Error::InvalidValue(e).source_context("invalid `name`"))?;
         let contract = Self::resolve_contract(logger, link_resolver, file).await?;
 
         Ok(Abi { name, contract })
@@ -448,7 +451,8 @@ impl RawTable {
     ) -> Result<Table, Error> {
         let Self { name, query, file } = self;
 
-        validate_ident(&name).map_err(|e| e.source_context("invalid `name`"))?;
+        validate_ident(&name)
+            .map_err(|e| Error::InvalidValue(e).source_context("invalid `name`"))?;
         let query = match Self::resolve_query(query, source, abis)? {
             Some(query) => query,
             None => Self::resolve_file(logger, link_resolver, file, source, abis).await?,
@@ -459,7 +463,7 @@ impl RawTable {
 
         for field in schema.fields() {
             validate_ident(field.name()).map_err(|e| {
-                e.source_context(format!(
+                Error::InvalidValue(e).source_context(format!(
                     "invalid query output schema: invalid column '{}'",
                     field.name()
                 ))
@@ -685,22 +689,6 @@ impl IsDeterministic for Error {
     }
 }
 
-fn validate_ident(s: &str) -> Result<(), Error> {
-    if !regex_is_match!("^[a-zA-Z_][a-zA-Z0-9_-]{0,100}$", s) {
-        return Err(Error::InvalidValue(
-            anyhow!("invalid identifier '{s}': must start with a letter or an underscore, and contain only letters, numbers, hyphens, and underscores")
-        ));
-    }
-    Ok(())
-}
-
-fn normalize_sql_ident(s: &str) -> String {
-    match validate_ident(s) {
-        Ok(()) => s.to_lowercase(),
-        Err(_e) => sqlparser_latest::ast::Ident::with_quote('"', s).to_string(),
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;

graph/src/amp/sql/mod.rs

@@ -1,3 +1,30 @@
 pub mod query_builder;
 
 pub use self::query_builder::{BlockRangeQueryBuilder, ContextQuery, ValidQuery};
+
+use anyhow::anyhow;
+use lazy_regex::regex_is_match;
+
+/// Validates that `s` is a simple SQL identifier: starts with a letter or
+/// underscore and contains only `[a-zA-Z0-9_-]`, up to 101 characters.
+pub fn validate_ident(s: &str) -> Result<(), anyhow::Error> {
+    if !regex_is_match!("^[a-zA-Z_][a-zA-Z0-9_-]{0,100}$", s) {
+        return Err(anyhow!(
+            "invalid identifier '{s}': must start with a letter or an underscore, \
+             and contain only letters, numbers, hyphens, and underscores"
+        ));
+    }
+    Ok(())
+}
+
+/// Normalizes a SQL identifier for safe interpolation into SQL `FROM` clauses.
+///
+/// Simple identifiers (matching `validate_ident`) are lowercased and returned
+/// unquoted. Identifiers with special characters are double-quoted using
+/// `sqlparser_latest::ast::Ident::with_quote`.
+pub fn normalize_sql_ident(s: &str) -> String {
+    match validate_ident(s) {
+        Ok(()) => s.to_lowercase(),
+        Err(_) => sqlparser_latest::ast::Ident::with_quote('"', s).to_string(),
+    }
+}

node/src/config.rs

@@ -1,4 +1,5 @@
 use graph::{
+    amp::sql::normalize_sql_ident,
     anyhow::Error,
     blockchain::BlockchainKind,
     components::network_provider::{AmpChainConfig, AmpChainNames, ChainName},
@@ -155,8 +156,8 @@ impl Config {
                     AmpChainConfig {
                         address: uri,
                         token: amp.token.clone(),
-                        context_dataset: amp.context_dataset.clone(),
-                        context_table: amp.context_table.clone(),
+                        context_dataset: normalize_sql_ident(&amp.context_dataset),
+                        context_table: normalize_sql_ident(&amp.context_table),
                         network: amp.network.clone(),
                     },
                 );
@@ -2528,4 +2529,84 @@ fdw_pool_size = [
         assert_eq!(amp.context_table, "blocks");
         assert_eq!(amp.network, Some("ethereum-mainnet".to_string()));
     }
+
+    #[test]
+    fn amp_chain_config_normalizes_context_idents() {
+        let section = toml::from_str::<ChainSection>(
+            r#"
+            ingestor = "block_ingestor_node"
+            [mainnet]
+            shard = "primary"
+            provider = []
+            [mainnet.amp]
+            address = "http://localhost:50051"
+            context_dataset = "ns/data@v1"
+            context_table = "my/table@2"
+            "#,
+        )
+        .unwrap();
+
+        let config = Config {
+            node: NodeId::new("test").unwrap(),
+            general: None,
+            stores: {
+                let mut s = std::collections::BTreeMap::new();
+                s.insert(
+                    "primary".to_string(),
+                    toml::from_str::<Shard>(r#"connection = "postgresql://u:p@h/db""#).unwrap(),
+                );
+                s
+            },
+            chains: section,
+            deployment: toml::from_str("[[rule]]\nshards = [\"primary\"]\nindexers = [\"test\"]")
+                .unwrap(),
+        };
+
+        let map = config.amp_chain_configs().unwrap();
+        let mainnet = map.get("mainnet").expect("mainnet should be in map");
+
+        // Identifiers with special characters should be double-quoted
+        assert_eq!(mainnet.context_dataset, "\"ns/data@v1\"");
+        assert_eq!(mainnet.context_table, "\"my/table@2\"");
+    }
+
+    #[test]
+    fn amp_chain_config_lowercases_simple_context_idents() {
+        let section = toml::from_str::<ChainSection>(
+            r#"
+            ingestor = "block_ingestor_node"
+            [mainnet]
+            shard = "primary"
+            provider = []
+            [mainnet.amp]
+            address = "http://localhost:50051"
+            context_dataset = "Eth"
+            context_table = "Blocks"
+            "#,
+        )
+        .unwrap();
+
+        let config = Config {
+            node: NodeId::new("test").unwrap(),
+            general: None,
+            stores: {
+                let mut s = std::collections::BTreeMap::new();
+                s.insert(
+                    "primary".to_string(),
+                    toml::from_str::<Shard>(r#"connection = "postgresql://u:p@h/db""#).unwrap(),
+                );
+                s
+            },
+            chains: section,
+            deployment: toml::from_str("[[rule]]\nshards = [\"primary\"]\nindexers = [\"test\"]")
+                .unwrap(),
+        };
+
+        let map = config.amp_chain_configs().unwrap();
+        let mainnet = map.get("mainnet").expect("mainnet should be in map");
+
+        // Simple identifiers should be lowercased and unquoted
+        assert_eq!(mainnet.context_dataset, "eth");
+        assert_eq!(mainnet.context_table, "blocks");
+    }
 }