graph,graphql,server,store: Subgraph Sql Service

Cargo.toml

@@ -10,6 +10,7 @@ members = [
     "store/*",
     "substreams/*",
     "graph",
+    "graph/derive",
     "tests",
     "graph/derive",
 ]
@@ -24,7 +25,13 @@ repository = "https://github.com/graphprotocol/graph-node"
 license = "MIT OR Apache-2.0"
 
 [workspace.dependencies]
-diesel = { version = "2.1.3", features = ["postgres", "serde_json", "numeric", "r2d2", "chrono"] }
+diesel = { version = "2.1.3", features = [
+    "postgres",
+    "serde_json",
+    "numeric",
+    "r2d2",
+    "chrono",
+] }
 diesel-derive-enum = { version = "2.1.0", features = ["postgres"] }
 diesel_derives = "2.1.3"
 diesel-dynamic-schema = "0.2.1"

graph/src/components/store/traits.rs

@@ -14,7 +14,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};
@@ -575,6 +575,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/data/graphql/ext.rs

@@ -7,12 +7,13 @@ use crate::prelude::s::{
     TypeDefinition, Value,
 };
 use crate::prelude::{ValueType, ENV_VARS};
-use crate::schema::{META_FIELD_TYPE, SCHEMA_TYPE_NAME};
+use crate::schema::{META_FIELD_TYPE, SCHEMA_TYPE_NAME, SQL_FIELD_TYPE};
 use std::collections::{BTreeMap, HashMap};
 
 pub trait ObjectTypeExt {
     fn field(&self, name: &str) -> Option<&Field>;
     fn is_meta(&self) -> bool;
+    fn is_sql(&self) -> bool;
 }
 
 impl ObjectTypeExt for ObjectType {
@@ -23,6 +24,10 @@ impl ObjectTypeExt for ObjectType {
     fn is_meta(&self) -> bool {
         self.name == META_FIELD_TYPE
     }
+
+    fn is_sql(&self) -> bool {
+        self.name == SQL_FIELD_TYPE
+    }
 }
 
 impl ObjectTypeExt for InterfaceType {
@@ -33,6 +38,10 @@ impl ObjectTypeExt for InterfaceType {
     fn is_meta(&self) -> bool {
         false
     }
+
+    fn is_sql(&self) -> bool {
+        false
+    }
 }
 
 pub trait DocumentExt {

graph/src/data/graphql/object_or_interface.rs

@@ -134,4 +134,11 @@ impl<'a> ObjectOrInterface<'a> {
             ObjectOrInterface::Interface(i) => i.is_meta(),
         }
     }
+
+    pub fn is_sql(&self) -> bool {
+        match self {
+            ObjectOrInterface::Object(o) => o.is_sql(),
+            ObjectOrInterface::Interface(i) => i.is_sql(),
+        }
+    }
 }

graph/src/data/query/error.rs

@@ -75,6 +75,7 @@ pub enum QueryExecutionError {
     InvalidSubgraphManifest,
     ResultTooBig(usize, usize),
     DeploymentNotFound(String),
+    SqlError(String),
     IdMissing,
     IdNotString,
     ConstraintViolation(String),
@@ -139,6 +140,7 @@ impl QueryExecutionError {
             | IdMissing
             | IdNotString
             | ConstraintViolation(_) => false,
+            SqlError(_) => false,
         }
     }
 }
@@ -289,6 +291,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"),
             ConstraintViolation(msg) => write!(f, "internal constraint violated: {}", msg),
+            SqlError(e) => write!(f, "sql error: {}", e),
         }
     }
 }

graph/src/data/store/mod.rs

@@ -1085,6 +1085,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/schema/api.rs

@@ -5,17 +5,24 @@ use std::sync::Arc;
 use anyhow::Context;
 use graphql_parser::Pos;
 use lazy_static::lazy_static;
-use thiserror::Error;
 
-use crate::cheap_clone::CheapClone;
 use crate::data::graphql::{ObjectOrInterface, ObjectTypeExt, TypeExt};
 use crate::data::store::IdType;
-use crate::env::ENV_VARS;
-use crate::schema::{ast, META_FIELD_NAME, META_FIELD_TYPE, SCHEMA_TYPE_NAME};
+use crate::schema::{
+    ast, META_FIELD_NAME, META_FIELD_TYPE, SCHEMA_TYPE_NAME, SQL_FIELD_NAME, SQL_FIELD_TYPE,
+    SQL_INPUT_TYPE,
+};
 
 use crate::data::graphql::ext::{
     camel_cased_names, DefinitionExt, DirectiveExt, DocumentExt, ValueExt,
 };
+use crate::prelude::s::*;
+use crate::prelude::*;
+use thiserror::Error;
+
+use crate::cheap_clone::CheapClone;
+use crate::env::ENV_VARS;
+
 use crate::derive::CheapClone;
 use crate::prelude::{q, r, s, DeploymentHash};
 
@@ -356,6 +363,7 @@ pub(in crate::schema) fn api_schema(
     // Refactor: Don't clone the schema.
     let mut api = init_api_schema(input_schema)?;
     add_meta_field_type(&mut api.document);
+    add_sql_field_type(&mut api.document);
     add_types_for_object_types(&mut api, input_schema)?;
     add_types_for_interface_types(&mut api, input_schema)?;
     add_types_for_aggregation_types(&mut api, input_schema)?;
@@ -465,6 +473,21 @@ fn add_meta_field_type(api: &mut s::Document) {
         .extend(META_FIELD_SCHEMA.definitions.iter().cloned());
 }
 
+// Adds a global `SqlOutput` type to the schema. The `sql` field
+// accepts values of this type
+fn add_sql_field_type(schema: &mut Document) {
+    lazy_static! {
+        static ref SQL_FIELD_SCHEMA: Document = {
+            let schema = include_str!("sql.graphql");
+            parse_schema(schema).expect("the schema `sql.graphql` is invalid")
+        };
+    }
+
+    schema
+        .definitions
+        .extend(SQL_FIELD_SCHEMA.definitions.iter().cloned());
+}
+
 fn add_types_for_object_types(
     api: &mut Schema,
     schema: &InputSchema,
@@ -1061,6 +1084,7 @@ fn add_query_type(api: &mut s::Document, input_schema: &InputSchema) -> Result<(
     fields.append(&mut agg_fields);
     fields.append(&mut fulltext_fields);
     fields.push(meta_field());
+    fields.push(sql_field());
 
     let typedef = s::TypeDefinition::Object(s::ObjectType {
         position: Pos::default(),
@@ -1159,6 +1183,7 @@ fn add_subscription_type(
         .collect::<Vec<s::Field>>();
     fields.append(&mut agg_fields);
     fields.push(meta_field());
+    fields.push(sql_field());
 
     let typedef = s::TypeDefinition::Object(s::ObjectType {
         position: Pos::default(),
@@ -1304,6 +1329,31 @@ fn meta_field() -> s::Field {
     META_FIELD.clone()
 }
 
+fn sql_field() -> s::Field {
+    lazy_static! {
+        static ref SQL_FIELD: s::Field = s::Field {
+            position: Pos::default(),
+            description: Some("Access to SQL queries".to_string()),
+            name: SQL_FIELD_NAME.to_string(),
+            arguments: vec![
+                InputValue {
+                    position: Pos::default(),
+                    description: None,
+                    name: String::from("input"),
+                    value_type: Type::NonNullType(Box::new(Type::NamedType(SQL_INPUT_TYPE.to_string()))),
+                    default_value: None,
+                    directives: vec![],
+
+                }
+            ],
+            field_type: Type::NamedType(SQL_FIELD_TYPE.to_string()),
+            directives: vec![],
+        };
+    }
+
+    SQL_FIELD.clone()
+}
+
 #[cfg(test)]
 mod tests {
     use crate::{

graph/src/schema/mod.rs

@@ -42,6 +42,12 @@ pub const INTROSPECTION_SCHEMA_FIELD_NAME: &str = "__schema";
 pub const META_FIELD_TYPE: &str = "_Meta_";
 pub const META_FIELD_NAME: &str = "_meta";
 
+pub const SQL_FIELD_TYPE: &str = "SqlOutput";
+pub const SQL_JSON_FIELD_TYPE: &str = "SqlJSONOutput";
+pub const SQL_CSV_FIELD_TYPE: &str = "SqlCSVOutput";
+pub const SQL_INPUT_TYPE: &str = "SqlInput";
+pub const SQL_FIELD_NAME: &str = "sql";
+
 pub const INTROSPECTION_TYPE_FIELD_NAME: &str = "__type";
 
 pub const BLOCK_FIELD_TYPE: &str = "_Block_";

graph/src/schema/sql.graphql

@@ -0,0 +1,67 @@
+scalar String
+scalar JSONObject
+scalar SqlVariable
+
+
+enum SqlFormat {
+  JSON
+  CSV
+}
+
+input SqlInput {
+  """
+  The SQL query to execute. The query may contain positional parameters
+  that are passed in the `parameters` field.
+  """
+  query: String!
+  """
+  The SQL query parameters. The parameters are passed to the SQL query
+  as positional parameters. The parameters are converted to the SQL
+  types based on the GraphQL types of the parameters.
+  """
+  parameters: [SqlVariable]
+
+  """
+  The format of the SQL query result. The default format is JSON.
+  """
+  format: SqlFormat = JSON
+}
+
+
+# The type names are purposely awkward to minimize the risk of them
+# colliding with user-supplied types
+"The type for the top-level sql field"
+type SqlJSONOutput {
+  "The columns returned by the query"
+  columns: [String!]!
+
+  """
+  The SQL query result row count.
+  """
+  rowCount: Int!
+
+  """
+  The SQL query result rows. Each row is represented as a list of values.
+  The values are represented as JSON values.
+  """
+  rows: [JSONObject!]!
+}
+
+
+type SqlCSVOutput {
+
+  "The columns returned by the query"
+  columns: [String!]!
+
+  """
+  The SQL query result rows. Represented as a CSV string
+  """
+  result: String
+  """
+  Number of SQL rows being returned
+  """
+  rowCount: Int!
+}
+
+union SqlOutput = SqlJSONOutput | SqlCSVOutput
+