graph, graphql, server: Add rudimentary SQL query support

Author: lutter

graph/src/components/graphql.rs

@@ -1,6 +1,7 @@
-use crate::data::query::QueryResults;
 use crate::data::query::{Query, QueryTarget};
-use crate::prelude::DeploymentHash;
+use crate::data::query::{QueryResults, SqlQueryReq};
+use crate::data::store::SqlQueryObject;
+use crate::prelude::{DeploymentHash, QueryExecutionError};
 
 use async_trait::async_trait;
 use std::sync::Arc;
@@ -28,6 +29,11 @@ pub trait GraphQlRunner: Send + Sync + 'static {
     ) -> QueryResults;
 
     fn metrics(&self) -> Arc<dyn GraphQLMetrics>;
+
+    async fn run_sql_query(
+        self: Arc<Self>,
+        req: SqlQueryReq,
+    ) -> Result<Vec<SqlQueryObject>, QueryExecutionError>;
 }
 
 pub trait GraphQLMetrics: Send + Sync + 'static {

graph/src/data/query/mod.rs

@@ -6,6 +6,6 @@ mod trace;
 
 pub use self::cache_status::CacheStatus;
 pub use self::error::{QueryError, QueryExecutionError};
-pub use self::query::{Query, QueryTarget, QueryVariables};
+pub use self::query::{Query, QueryTarget, QueryVariables, SqlQueryMode, SqlQueryReq};
 pub use self::result::{LatestBlockInfo, QueryResult, QueryResults};
 pub use self::trace::Trace;

graph/src/data/query/query.rs

@@ -1,7 +1,8 @@
 use serde::de::Deserializer;
-use serde::Deserialize;
+use serde::{Deserialize, Serialize};
 use std::collections::{BTreeMap, HashMap};
 use std::convert::TryFrom;
+use std::hash::{DefaultHasher, Hash as _, Hasher as _};
 use std::ops::{Deref, DerefMut};
 use std::sync::Arc;
 
@@ -165,3 +166,26 @@ impl Query {
         }
     }
 }
+
+#[derive(Copy, Clone, Debug, Deserialize, Serialize)]
+#[serde(rename_all = "snake_case")]
+pub enum SqlQueryMode {
+    Data,
+    Info,
+}
+
+#[derive(Clone, Debug, Deserialize, Serialize)]
+pub struct SqlQueryReq {
+    pub deployment: DeploymentHash,
+    pub query: String,
+    pub mode: SqlQueryMode,
+}
+
+impl SqlQueryReq {
+    pub fn query_hash(&self) -> u64 {
+        let mut hasher = DefaultHasher::new();
+        self.deployment.hash(&mut hasher);
+        self.query.hash(&mut hasher);
+        hasher.finish()
+    }
+}

graph/src/data/store/mod.rs

@@ -1080,6 +1080,7 @@ pub struct QueryObject {
 }
 
 /// An object that is returned from a SQL query. It wraps an `r::Value`
+#[derive(CacheWeight, Serialize)]
 pub struct SqlQueryObject(pub r::Value);
 
 impl CacheWeight for QueryObject {

graphql/src/runner.rs

@@ -4,12 +4,14 @@ use std::time::Instant;
 use crate::metrics::GraphQLMetrics;
 use crate::prelude::{QueryExecutionOptions, StoreResolver};
 use crate::query::execute_query;
+use graph::data::query::{CacheStatus, SqlQueryReq};
+use graph::data::store::SqlQueryObject;
 use graph::futures03::future;
-use graph::prelude::MetricsRegistry;
 use graph::prelude::{
     async_trait, o, CheapClone, DeploymentState, GraphQLMetrics as GraphQLMetricsTrait,
     GraphQlRunner as GraphQlRunnerTrait, Logger, Query, QueryExecutionError, ENV_VARS,
 };
+use graph::prelude::{ApiVersion, MetricsRegistry};
 use graph::{data::graphql::load_manager::LoadManager, prelude::QueryStoreManager};
 use graph::{
     data::query::{LatestBlockInfo, QueryResults, QueryTarget},
@@ -251,4 +253,43 @@ where
     fn metrics(&self) -> Arc<dyn GraphQLMetricsTrait> {
         self.graphql_metrics.clone()
     }
+
+    async fn run_sql_query(
+        self: Arc<Self>,
+        req: SqlQueryReq,
+    ) -> Result<Vec<SqlQueryObject>, QueryExecutionError> {
+        let store = self
+            .store
+            .query_store(QueryTarget::Deployment(
+                req.deployment.clone(),
+                ApiVersion::default(),
+            ))
+            .await?;
+
+        let query_hash = req.query_hash();
+        self.load_manager
+            .decide(
+                &store.wait_stats(),
+                store.shard(),
+                store.deployment_id(),
+                query_hash,
+                &req.query,
+            )
+            .to_result()?;
+
+        let query_start = Instant::now();
+        let result = store
+            .execute_sql(&req.query)
+            .map_err(|e| QueryExecutionError::from(e));
+
+        self.load_manager.record_work(
+            store.shard(),
+            store.deployment_id(),
+            query_hash,
+            query_start.elapsed(),
+            CacheStatus::Miss,
+        );
+
+        result
+    }
 }

server/http/src/service.rs

@@ -9,6 +9,8 @@ use graph::components::server::query::ServerResponse;
 use graph::components::server::query::ServerResult;
 use graph::components::versions::ApiVersion;
 use graph::data::query::QueryResult;
+use graph::data::query::SqlQueryMode;
+use graph::data::query::SqlQueryReq;
 use graph::data::subgraph::DeploymentHash;
 use graph::data::subgraph::SubgraphName;
 use graph::env::ENV_VARS;
@@ -21,6 +23,8 @@ use graph::hyper::{body::Body, header::HeaderValue};
 use graph::hyper::{Method, Request, Response, StatusCode};
 use graph::prelude::serde_json;
 use graph::prelude::serde_json::json;
+use graph::prelude::CacheWeight as _;
+use graph::prelude::QueryError;
 use graph::semver::VersionReq;
 use graph::slog::error;
 use graph::slog::Logger;
@@ -195,6 +199,51 @@ where
         Ok(result.as_http_response())
     }
 
+    async fn handle_sql_query<T: Body>(&self, request: Request<T>) -> ServerResult {
+        let body = request
+            .collect()
+            .await
+            .map_err(|_| ServerError::InternalError("Failed to read request body".into()))?
+            .to_bytes();
+        let sql_req: SqlQueryReq = serde_json::from_slice(&body)
+            .map_err(|e| ServerError::ClientError(format!("{}", e)))?;
+
+        let mode = sql_req.mode;
+        let result = self
+            .graphql_runner
+            .cheap_clone()
+            .run_sql_query(sql_req)
+            .await
+            .map_err(|e| ServerError::QueryError(QueryError::from(e)));
+
+        use SqlQueryMode::*;
+        let response_obj = match (result, mode) {
+            (Ok(result), Info) => {
+                json!({
+                    "count": result.len(),
+                    "bytes" : result.weight(),
+                })
+            }
+            (Ok(result), Data) => {
+                json!({
+                    "data": result,
+                })
+            }
+            (Err(e), _) => json!({
+                "error": e.to_string(),
+            }),
+        };
+
+        let response_str = serde_json::to_string(&response_obj).unwrap();
+
+        Ok(Response::builder()
+            .status(200)
+            .header(ACCESS_CONTROL_ALLOW_ORIGIN, "*")
+            .header(CONTENT_TYPE, "application/json")
+            .body(Full::from(response_str))
+            .unwrap())
+    }
+
     // Handles OPTIONS requests
     fn handle_graphql_options<T>(&self, _request: Request<T>) -> ServerResult {
         Ok(Response::builder()
@@ -327,7 +376,9 @@ where
                 let dest = format!("/{}/graphql", filtered_path);
                 self.handle_temp_redirect(dest)
             }
-
+            (Method::POST, &["subgraphs", "sql"] | &["subgraphs", "sql", ""]) => {
+                self.handle_sql_query(req).await
+            }
             (Method::POST, &["subgraphs", "id", subgraph_id]) => {
                 self.handle_graphql_query_by_id(subgraph_id.to_owned(), req)
                     .await
@@ -395,14 +446,15 @@ where
 
 #[cfg(test)]
 mod tests {
+    use graph::data::store::SqlQueryObject;
     use graph::data::value::{Object, Word};
     use graph::http_body_util::{BodyExt, Full};
     use graph::hyper::body::Bytes;
     use graph::hyper::header::{CONTENT_LENGTH, CONTENT_TYPE};
     use graph::hyper::{Method, Request, StatusCode};
     use graph::prelude::serde_json::json;
 
-    use graph::data::query::{QueryResults, QueryTarget};
+    use graph::data::query::{QueryResults, QueryTarget, SqlQueryReq};
     use graph::prelude::*;
 
     use crate::test_utils;
@@ -449,6 +501,13 @@ mod tests {
         fn metrics(&self) -> Arc<dyn GraphQLMetrics> {
             Arc::new(TestGraphQLMetrics)
         }
+
+        async fn run_sql_query(
+            self: Arc<Self>,
+            _req: SqlQueryReq,
+        ) -> Result<Vec<SqlQueryObject>, QueryExecutionError> {
+            unimplemented!()
+        }
     }
 
     #[tokio::test]

server/http/tests/server.rs

@@ -1,4 +1,7 @@
-use graph::http::StatusCode;
+use graph::{
+    data::{query::SqlQueryReq, store::SqlQueryObject},
+    http::StatusCode,
+};
 use std::time::Duration;
 
 use graph::data::{
@@ -66,6 +69,13 @@ impl GraphQlRunner for TestGraphQlRunner {
     fn metrics(&self) -> Arc<dyn GraphQLMetrics> {
         Arc::new(TestGraphQLMetrics)
     }
+
+    async fn run_sql_query(
+        self: Arc<Self>,
+        _req: SqlQueryReq,
+    ) -> Result<Vec<SqlQueryObject>, QueryExecutionError> {
+        unimplemented!();
+    }
 }
 
 #[cfg(test)]