graphql, store: add parse and validation for custom sql

Signed-off-by: Gustavo Inacio <[email protected]>

create store/postgres/src/sql

validator and formater: full refactor

Signed-off-by: Gustavo Inacio <[email protected]>

refactor: move sql to store/postgres/src/sql

Signed-off-by: Gustavo Inacio <[email protected]>

parser: test for array of byte fixed

Signed-off-by: Tümay Tuzcu <[email protected]>

parser: block_range and block$ added as available columns for tables,
sequence functions moved to black list

Signed-off-by: Tümay Tuzcu <[email protected]>

parser: use latest block as cte filter

parser: fix escape columns and tables

Fixes STU-217

Signed-off-by: Gustavo Inacio <[email protected]>

Author: gusinacio

Cargo.lock

@@ -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"

Cargo.toml

@@ -16,6 +16,7 @@ use graph::schema::{
 };
 use graph::schema::{ErrorPolicy, BLOCK_FIELD_TYPE};
 
+
 use crate::execution::{ast as a, Query};
 use crate::metrics::GraphQLMetrics;
 use crate::prelude::{ExecutionContext, Resolver};

graphql/src/store/resolver.rs

@@ -30,6 +30,8 @@ git-testament = "0.2.5"
 itertools = "0.12.0"
 hex = "0.4.3"
 pretty_assertions = "1.4.0"
+sqlparser = { version = "0.40.0", features = ["visitor"] }
+thiserror = "1.0.25"
 
 [dev-dependencies]
 clap = "3.2.25"

store/postgres/Cargo.toml

@@ -33,6 +33,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,6 +2,7 @@ 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, SqlQueryObject};
@@ -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,
         }
     }
 }
@@ -65,7 +71,15 @@ impl QueryStoreTrait for QueryStore {
             .store
             .get_replica_conn(self.replica_id)
             .map_err(|e| QueryExecutionError::SqlError(format!("SQL error: {}", e)))?;
-        self.store.execute_sql(&mut conn, sql)
+
+        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,

store/postgres/src/query_store.rs

@@ -0,0 +1,78 @@
+use sqlparser::ast::{ObjectName, Statement, TableFactor, VisitMut, VisitorMut};
+use std::ops::ControlFlow;
+
+pub struct Formatter<'a> {
+    prelude: &'a str,
+}
+
+impl<'a> Formatter<'a> {
+    pub fn new(prelude: &'a str) -> Self {
+        Self { prelude }
+    }
+
+    fn prepend_prefix_to_object_name_mut(&self, name: &mut ObjectName) {
+        let table_identifier: &mut Vec<_> = &mut name.0;
+        // remove all but the last identifier
+        table_identifier.drain(0..table_identifier.len() - 1);
+    }
+
+    pub fn format(&mut self, statement: &mut Statement) -> String {
+        statement.visit(self);
+
+        format!(
+            "{} SELECT to_jsonb(sub.*) AS data FROM ( {} ) AS sub",
+            self.prelude, statement
+        )
+    }
+}
+
+impl VisitorMut for Formatter<'_> {
+    type Break = ();
+
+    fn pre_visit_table_factor(
+        &mut self,
+        table_factor: &mut TableFactor,
+    ) -> ControlFlow<Self::Break> {
+        if let TableFactor::Table { name, .. } = table_factor {
+            self.prepend_prefix_to_object_name_mut(name);
+        }
+        ControlFlow::Continue(())
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+    use crate::sql::constants::SQL_DIALECT;
+    const CTE_PREFIX: &str = "WITH swap AS (
+            SELECT 
+            id,
+            amount_in,
+            amount_out,
+            concat('0x',encode(token_in,'hex') as token_in,
+            concat('0x',token_out,'hex') AS token_out 
+            FROM 
+            sdg1.swap
+        )";
+
+    #[test]
+    fn format_sql() {
+        let mut formatter = Formatter::new(CTE_PREFIX);
+
+        let sql = "SELECT token_in, SUM(amount_in) AS amount FROM unknown.swap GROUP BY token_in";
+
+        let mut statements = sqlparser::parser::Parser::parse_sql(&SQL_DIALECT, sql).unwrap();
+
+        let mut statement = statements.get_mut(0).unwrap();
+
+        let result = formatter.format(&mut statement);
+
+        assert_eq!(
+            result,
+            format!(
+                "{} SELECT to_jsonb(sub.*) AS data FROM ( {} ) AS sub",
+                CTE_PREFIX, "SELECT token_in, SUM(amount_in) AS amount FROM swap GROUP BY token_in"
+            )
+        );
+    }
+}

store/postgres/src/sql/constants.rs

@@ -0,0 +1,10 @@
+mod constants;
+mod formatter;
+mod parser;
+mod validation;
+
+use std::collections::{HashMap, HashSet};
+
+pub(self) type Schema = HashMap<String, HashSet<String>>; // HashMap<Table, HashSet<Column>>
+
+pub use parser::Parser;