fix(core, graph): resolve metric conflicts, use quoted table references in sql

Author: isum

Cargo.lock

@@ -35,7 +35,7 @@ impl Metrics {
         let stopwatch = StopwatchMetrics::new(
             logger.cheap_clone(),
             deployment.cheap_clone(),
-            "amp-process",
+            "process",
             metrics_registry.cheap_clone(),
             store.shard().to_string(),
         );
@@ -123,7 +123,11 @@ impl DeploymentHead {
             .new_int_gauge(
                 "deployment_head",
                 "Tracks the most recent block number processed by a deployment",
-                const_labels,
+                const_labels
+                    .into_iter()
+                    .map(|(k, v)| (k.to_string(), v.to_string()))
+                    // TODO: Pass the network from the Amp manifest
+                    .chain([("network".to_string(), "".to_string())]),
             )
             .expect("failed to register `deployment_head` gauge");
 
@@ -139,7 +143,7 @@ impl DeploymentHead {
     }
 }
 
-/// Tracks the target block number of a deployment.
+/// Tracks the maximum block number currently available for indexing within a deployment.
 pub(super) struct DeploymentTarget(IntGauge);
 
 impl DeploymentTarget {
@@ -149,8 +153,8 @@ impl DeploymentTarget {
     ) -> Self {
         let int_gauge = metrics_registry
             .new_int_gauge(
-                "amp_deployment_target",
-                "Tracks the target block number of a deployment",
+                "deployment_target",
+                "Tracks the maximum block number currently available for indexing within a deployment",
                 const_labels,
             )
             .expect("failed to register `amp_deployment_target` gauge");
@@ -217,7 +221,7 @@ impl IndexingDuration {
     ) -> Self {
         let int_counter = metrics_registry
             .new_int_counter(
-                "amp_deployment_indexing_duration_seconds",
+                "deployment_indexing_duration_seconds",
                 "Tracks the total duration in seconds of deployment indexing",
                 const_labels,
             )

core/src/amp_subgraph/metrics.rs

@@ -439,13 +439,6 @@ impl Monitor {
                         error!(logger, "Subgraph panicked";
                             "error" => ?e
                         );
-
-                        // TODO: Maybe abort the entire process on panic and require a full graph-node restart.
-                        //       Q: Should a bug that is triggered in a specific subgraph affect everything?
-                        //       Q: How to make this failure loud enough so it is not missed?
-                        //
-                        // println!("Subgraph panicked");
-                        // std::process::abort();
                     }
                     Err(e) => {
                         error!(logger, "Subgraph failed";

core/src/amp_subgraph/monitor.rs

@@ -133,6 +133,7 @@ async fn process_record_batch_group<AC>(
         })?;
     }
 
+    let section = cx.metrics.stopwatch.start_section("as_modifications");
     let ModificationsAndCache {
         modifications,
         entity_lfu_cache,
@@ -142,7 +143,9 @@ async fn process_record_batch_group<AC>(
         .await
         .map_err(Error::from)
         .map_err(|e| e.context("failed to extract entity modifications from the state"))?;
+    section.end();
 
+    let _section = cx.metrics.stopwatch.start_section("transact_block");
     let is_close_to_chain_head = latest_block.saturating_sub(block_number) <= 100;
 
     cx.store

core/src/amp_subgraph/runner/data_processing.rs

@@ -553,7 +553,7 @@ Amp-powered subgraphs feature introduces the following new ENV variables:
 In addition to reporting updates to the existing `deployment_status`, `deployment_head`, `deployment_synced` and `deployment_blocks_processed_count`
 metrics, Amp-powered subgraphs feature introduces the following new metrics:
 
-- `amp_deployment_target` – Tracks the target block number of a deployment.
-- `amp_deployment_indexing_duration_seconds` – Tracks the total duration in seconds of deployment indexing.
+- `deployment_target` – Tracks the maximum block number currently available for indexing within a deployment.
+- `deployment_indexing_duration_seconds` – Tracks the total duration in seconds of deployment indexing.
 
-Additionally, the `deployment_sync_secs` is extended with a new `amp-process` stage and new sections specific to the Amp indexing process.
+Additionally, the `deployment_sync_secs` is extended with new sections specific to the Amp indexing process.

docs/amp-powered-subgraphs.md

@@ -99,6 +99,7 @@ alloy.workspace = true
 arrow-flight.workspace = true
 arrow.workspace = true
 half.workspace = true
+indoc.workspace = true
 lazy-regex.workspace = true
 sqlparser-latest.workspace = true
 tokio-util.workspace = true

graph/Cargo.toml

@@ -7,6 +7,7 @@ use alloy::{
 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;
@@ -86,14 +87,7 @@ impl RawDataSource {
             .map_err(|e| e.source_context("invalid `source`"))?;
 
         let transformer = transformer
-            .resolve(
-                &logger,
-                link_resolver,
-                amp_client,
-                input_schema,
-                &network,
-                &source,
-            )
+            .resolve(&logger, link_resolver, amp_client, input_schema, &source)
             .await
             .map_err(|e| e.source_context("invalid `transformer`"))?;
 
@@ -159,6 +153,12 @@ impl RawSource {
         }
         Self::validate_tables(&tables)?;
 
+        let dataset = normalize_sql_ident(&dataset);
+        let tables = tables
+            .into_iter()
+            .map(|table| normalize_sql_ident(&table))
+            .collect_vec();
+
         let address = address.unwrap_or(Address::ZERO);
         let start_block = start_block.unwrap_or(BlockNumber::MIN);
         let end_block = end_block.unwrap_or(BlockNumber::MAX);
@@ -232,7 +232,6 @@ impl RawTransformer {
         link_resolver: &dyn LinkResolver,
         amp_client: &impl amp::Client,
         input_schema: Option<&InputSchema>,
-        network: &str,
         source: &Source,
     ) -> Result<Transformer, Error> {
         let Self {
@@ -248,7 +247,6 @@ impl RawTransformer {
             link_resolver,
             amp_client,
             input_schema,
-            network,
             tables,
             source,
             &abis,
@@ -306,7 +304,6 @@ impl RawTransformer {
         link_resolver: &dyn LinkResolver,
         amp_client: &impl amp::Client,
         input_schema: Option<&InputSchema>,
-        network: &str,
         tables: Vec<RawTable>,
         source: &Source,
         abis: &[Abi],
@@ -335,7 +332,6 @@ impl RawTransformer {
                     link_resolver,
                     amp_client,
                     input_schema,
-                    network,
                     source,
                     abis,
                 )
@@ -431,7 +427,6 @@ impl RawTable {
         link_resolver: &dyn LinkResolver,
         amp_client: &impl amp::Client,
         input_schema: Option<&InputSchema>,
-        network: &str,
         source: &Source,
         abis: &[Abi],
     ) -> Result<Table, Error> {
@@ -459,7 +454,6 @@ impl RawTable {
             logger,
             amp_client,
             input_schema,
-            network,
             source,
             query,
             schema.clone(),
@@ -557,7 +551,6 @@ impl RawTable {
         logger: &Logger,
         amp_client: &impl amp::Client,
         input_schema: Option<&InputSchema>,
-        network: &str,
         source: &Source,
         query: ValidQuery,
         schema: Schema,
@@ -586,15 +579,7 @@ impl RawTable {
         let context_sources_iter = source
             .tables
             .iter()
-            .map(|table| (source.dataset.as_str(), table.as_str()))
-            // TODO: Replace hardcoded values with schema metadata sources when available
-            .chain(match network {
-                "ethereum-mainnet" => vec![("edgeandnode/ethereum_mainnet", "blocks")],
-                "base-mainnet" => vec![("edgeandnode/base_mainnet", "blocks")],
-                "base-sepolia" => vec![("edgeandnode/base_sepolia", "blocks")],
-                "arbitrum-one" => vec![("edgeandnode/arbitrum_one", "blocks")],
-                _ => vec![],
-            });
+            .map(|table| (source.dataset.as_str(), table.as_str()));
 
         for (dataset, table) in context_sources_iter {
             let context_logger = logger.new(slog::o!(
@@ -719,3 +704,10 @@ fn validate_ident(s: &str) -> Result<(), Error> {
     }
     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(),
+    }
+}

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

@@ -134,24 +134,25 @@ mod tests {
         let block_range_query = new_block_range_query(&query, block_number_column, &block_range);
 
         assert_eq!(
-            block_range_query,
+            block_range_query.to_string(),
             parse_query(
                 r#"
-                WITH block_range_1164572571450379730 AS (
-                    SELECT * FROM "d" WHERE _block_num BETWEEN 0 AND 1000000
+                WITH block_range_14621009630487609643 AS (
+                    SELECT * FROM d WHERE _block_num BETWEEN 0 AND 1000000
                 ),
-                source_1164572571450379730 AS (
-                    SELECT a, b, c FROM block_range_1164572571450379730 AS d
+                source_14621009630487609643 AS (
+                    SELECT a, b, c FROM block_range_14621009630487609643 AS d
                 )
                 SELECT
-                    source_1164572571450379730.*
+                    source_14621009630487609643.*
                 FROM
-                    source_1164572571450379730
+                    source_14621009630487609643
                 ORDER BY
-                    source_1164572571450379730.b
+                    source_14621009630487609643.b
                 "#
             )
-            .unwrap(),
+            .unwrap()
+            .to_string(),
         )
     }
 
@@ -163,27 +164,28 @@ mod tests {
         let block_range_query = new_block_range_query(&query, block_number_column, &block_range);
 
         assert_eq!(
-            block_range_query,
+            block_range_query.to_string(),
             parse_query(
                 r#"
-                WITH block_range_1164572571450379730 AS (
-                    SELECT * FROM "d" WHERE _block_num BETWEEN 0 AND 1000000
+                WITH block_range_14621009630487609643 AS (
+                    SELECT * FROM d WHERE _block_num BETWEEN 0 AND 1000000
                 ),
-                block_range_13063992259633584610 AS (
-                    SELECT * FROM "e" WHERE _block_num BETWEEN 0 AND 1000000
+                block_range_12377422807768256314 AS (
+                    SELECT * FROM e WHERE _block_num BETWEEN 0 AND 1000000
                 ),
-                source_13063992259633584610 AS (
-                    SELECT a, b, c FROM block_range_1164572571450379730 AS d JOIN block_range_13063992259633584610 AS e ON e.e = d.d
+                source_12377422807768256314 AS (
+                    SELECT a, b, c FROM block_range_14621009630487609643 AS d JOIN block_range_12377422807768256314 AS e ON e.e = d.d
                 )
                 SELECT
-                    source_13063992259633584610.*
+                    source_12377422807768256314.*
                 FROM
-                    source_13063992259633584610
+                    source_12377422807768256314
                 ORDER BY
-                    source_13063992259633584610.b
+                    source_12377422807768256314.b
                 "#
             )
-            .unwrap(),
+            .unwrap()
+            .to_string(),
         )
     }
 }