graph, store: Create database sql executor

Author: gusinacio

Cargo.lock

@@ -81,11 +81,8 @@ serde_derive = "1.0.125"
 serde_json = { version = "1.0", features = ["arbitrary_precision"] }
 serde_regex = "1.1.0"
 serde_yaml = "0.9.21"
-slog = { version = "2.7.0", features = [
-    "release_max_level_trace",
-    "max_level_trace",
-] }
-sqlparser = "0.46.0"
+slog = { version = "2.7.0", features = ["release_max_level_trace", "max_level_trace"] }
+sqlparser = { version = "0.46.0", features = ["visitor"] }
 strum = { version = "0.26", features = ["derive"] }
 syn = { version = "2.0.106", features = ["full"] }
 test-store = { path = "./store/test-store" }

Cargo.toml

@@ -16,7 +16,7 @@ use crate::components::transaction_receipt;
 use crate::components::versions::ApiVersion;
 use crate::data::query::Trace;
 use crate::data::store::ethereum::call;
-use crate::data::store::QueryObject;
+use crate::data::store::{QueryObject, SqlQueryObject};
 use crate::data::subgraph::{status, DeploymentFeatures};
 use crate::data::{query::QueryTarget, subgraph::schema::*};
 use crate::prelude::{DeploymentState, NodeId, QueryExecutionError, SubgraphName};
@@ -652,6 +652,8 @@ pub trait QueryStore: Send + Sync {
         query: EntityQuery,
     ) -> Result<(Vec<QueryObject>, Trace), QueryExecutionError>;
 
+    fn execute_sql(&self, sql: &str) -> Result<Vec<SqlQueryObject>, QueryExecutionError>;
+
     async fn is_deployment_synced(&self) -> Result<bool, Error>;
 
     async fn block_ptr(&self) -> Result<Option<BlockPtr>, StoreError>;

graph/src/components/store/traits.rs

@@ -73,6 +73,7 @@ pub enum QueryExecutionError {
     InvalidSubgraphManifest,
     ResultTooBig(usize, usize),
     DeploymentNotFound(String),
+    SqlError(String),
     IdMissing,
     IdNotString,
     InternalError(String),
@@ -135,6 +136,7 @@ impl QueryExecutionError {
             | IdMissing
             | IdNotString
             | InternalError(_) => false,
+            SqlError(_) => false,
         }
     }
 }
@@ -213,7 +215,7 @@ impl fmt::Display for QueryExecutionError {
             }
             InvalidFilterError => write!(f, "Filter must by an object"),
             InvalidOrFilterStructure(fields, example) => {
-                write!(f, "Cannot mix column filters with 'or' operator at the same level. Found column filter(s) {} alongside 'or' operator.\n\n{}", 
+                write!(f, "Cannot mix column filters with 'or' operator at the same level. Found column filter(s) {} alongside 'or' operator.\n\n{}",
                     fields.join(", "), example)
             }
             EntityFieldError(e, a) => {
@@ -281,6 +283,7 @@ impl fmt::Display for QueryExecutionError {
             IdMissing => write!(f, "entity is missing an `id` attribute"),
             IdNotString => write!(f, "entity `id` attribute is not a string"),
             InternalError(msg) => write!(f, "internal error: {}", msg),
+            SqlError(e) => write!(f, "sql error: {}", e),
         }
     }
 }

graph/src/data/query/error.rs

@@ -1102,6 +1102,9 @@ pub struct QueryObject {
     pub entity: r::Object,
 }
 
+/// An object that is returned from a SQL query. It wraps an `r::Value`
+pub struct SqlQueryObject(pub r::Value);
+
 impl CacheWeight for QueryObject {
     fn indirect_weight(&self) -> usize {
         self.parent.indirect_weight() + self.entity.indirect_weight()

graph/src/data/store/mod.rs

@@ -32,6 +32,8 @@ git-testament = "0.2.6"
 itertools = "0.14.0"
 hex = "0.4.3"
 pretty_assertions = "1.4.1"
+sqlparser = { workspace = true }
+thiserror = { workspace = true }
 
 [dev-dependencies]
 clap.workspace = true

store/postgres/Cargo.toml

@@ -12,8 +12,9 @@ use graph::components::store::{
     PruningStrategy, QueryPermit, StoredDynamicDataSource, VersionStats,
 };
 use graph::components::versions::VERSIONS;
+use graph::data::graphql::IntoValue;
 use graph::data::query::Trace;
-use graph::data::store::IdList;
+use graph::data::store::{IdList, SqlQueryObject};
 use graph::data::subgraph::{status, SPEC_VERSION_0_0_6};
 use graph::data_source::CausalityRegion;
 use graph::derive::CheapClone;
@@ -54,7 +55,7 @@ use crate::dynds::DataSourcesTable;
 use crate::primary::{DeploymentId, Primary};
 use crate::relational::index::{CreateIndex, IndexList, Method};
 use crate::relational::{self, Layout, LayoutCache, SqlName, Table};
-use crate::relational_queries::FromEntityData;
+use crate::relational_queries::{FromEntityData, JSONData};
 use crate::{advisory_lock, catalog, retry};
 use crate::{detail, ConnectionPool};
 use crate::{dynds, primary::Site};
@@ -290,6 +291,24 @@ impl DeploymentStore {
         layout.query(&logger, conn, query)
     }
 
+    pub(crate) fn execute_sql(
+        &self,
+        conn: &mut PgConnection,
+        query: &str,
+    ) -> Result<Vec<SqlQueryObject>, QueryExecutionError> {
+        let query = diesel::sql_query(query);
+
+        // Execute the provided SQL query
+        let results = query
+            .load::<JSONData>(conn)
+            .map_err(|e| QueryExecutionError::SqlError(e.to_string()))?;
+
+        Ok(results
+            .into_iter()
+            .map(|e| SqlQueryObject(e.into_value()))
+            .collect::<Vec<_>>())
+    }
+
     fn check_intf_uniqueness(
         &self,
         conn: &mut PgConnection,

store/postgres/src/deployment_store.rs

@@ -30,6 +30,7 @@ pub mod query_store;
 mod relational;
 mod relational_queries;
 mod retry;
+mod sql;
 mod store;
 mod store_events;
 mod subgraph_store;

store/postgres/src/lib.rs

@@ -2,9 +2,10 @@ use std::collections::HashMap;
 use std::time::Instant;
 
 use crate::deployment_store::{DeploymentStore, ReplicaId};
+use crate::sql::Parser;
 use graph::components::store::{DeploymentId, QueryPermit, QueryStore as QueryStoreTrait};
 use graph::data::query::Trace;
-use graph::data::store::QueryObject;
+use graph::data::store::{QueryObject, SqlQueryObject};
 use graph::prelude::*;
 use graph::schema::{ApiSchema, InputSchema};
 
@@ -16,6 +17,7 @@ pub(crate) struct QueryStore {
     store: Arc<DeploymentStore>,
     chain_store: Arc<crate::ChainStore>,
     api_version: Arc<ApiVersion>,
+    sql_parser: Result<Parser, StoreError>,
 }
 
 impl QueryStore {
@@ -26,12 +28,16 @@ impl QueryStore {
         replica_id: ReplicaId,
         api_version: Arc<ApiVersion>,
     ) -> Self {
+        let sql_parser = store
+            .find_layout(site.clone())
+            .map(|layout| Parser::new(layout));
         QueryStore {
             site,
             replica_id,
             store,
             chain_store,
             api_version,
+            sql_parser,
         }
     }
 }
@@ -57,6 +63,25 @@ impl QueryStoreTrait for QueryStore {
             })
     }
 
+    fn execute_sql(
+        &self,
+        sql: &str,
+    ) -> Result<Vec<SqlQueryObject>, graph::prelude::QueryExecutionError> {
+        let mut conn = self
+            .store
+            .get_replica_conn(self.replica_id)
+            .map_err(|e| QueryExecutionError::SqlError(format!("SQL error: {}", e)))?;
+
+        let parser = self
+            .sql_parser
+            .as_ref()
+            .map_err(|e| QueryExecutionError::SqlError(format!("SQL error: {}", e)))?;
+
+        let sql = parser.parse_and_validate(sql)?;
+
+        self.store.execute_sql(&mut conn, &sql)
+    }
+
     /// Return true if the deployment with the given id is fully synced,
     /// and return false otherwise. Errors from the store are passed back up
     async fn is_deployment_synced(&self) -> Result<bool, Error> {

store/postgres/src/query_store.rs

@@ -14,6 +14,8 @@ use diesel::sql_types::{Array, BigInt, Binary, Bool, Int8, Integer, Jsonb, Text,
 use diesel::QuerySource as _;
 use graph::components::store::write::{EntityWrite, RowGroup, WriteChunk};
 use graph::components::store::{Child as StoreChild, DerivedEntityQuery};
+
+use graph::data::graphql::IntoValue;
 use graph::data::store::{Id, IdType, NULL};
 use graph::data::store::{IdList, IdRef, QueryObject};
 use graph::data::value::{Object, Word};
@@ -439,6 +441,47 @@ pub fn parse_id(id_type: IdType, json: serde_json::Value) -> Result<Id, StoreErr
     }
 }
 
+#[derive(QueryableByName, Debug)]
+pub struct JSONData {
+    #[diesel(sql_type = Jsonb)]
+    pub data: serde_json::Value,
+}
+
+impl IntoValue for JSONData {
+    fn into_value(self) -> r::Value {
+        JSONData::to_value(self.data)
+    }
+}
+
+impl JSONData {
+    pub fn to_value(data: serde_json::Value) -> r::Value {
+        match data {
+            serde_json::Value::Null => r::Value::Null,
+            serde_json::Value::Bool(b) => r::Value::Boolean(b),
+            serde_json::Value::Number(n) => {
+                if let Some(i) = n.as_i64() {
+                    r::Value::Int(i)
+                } else {
+                    r::Value::Float(n.as_f64().unwrap())
+                }
+            }
+            serde_json::Value::String(s) => r::Value::String(s),
+            serde_json::Value::Array(vals) => {
+                let vals: Vec<_> = vals.into_iter().map(JSONData::to_value).collect::<Vec<_>>();
+                r::Value::List(vals)
+            }
+            serde_json::Value::Object(map) => {
+                let mut m = std::collections::BTreeMap::new();
+                for (k, v) in map {
+                    let value = JSONData::to_value(v);
+                    m.insert(Word::from(k), value);
+                }
+                r::Value::object(m)
+            }
+        }
+    }
+}
+
 /// Helper struct for retrieving entities from the database. With diesel, we
 /// can only run queries that return columns whose number and type are known
 /// at compile time. Because of that, we retrieve the actual data for an