lots of progress

ArroyoSystems · Dec 4, 2023 · 963e5ba · 963e5ba
1 parent 175343c
commit 963e5ba
Show file tree

Hide file tree

Showing 9 changed files with 491 additions and 301 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/arroyo-api/src/pipelines.rs b/arroyo-api/src/pipelines.rs
@@ -61,7 +61,7 @@ async fn compile_sql<'e, E>(
     parallelism: usize,
     auth_data: &AuthData,
     tx: &E,
-) -> anyhow::Result<CompiledSql>
+) -> anyhow::Result<Option<CompiledSql>>
 where
     E: GenericClient,
 {
@@ -293,7 +293,8 @@ pub(crate) async fn create_pipeline<'a>(
                 tx,
             )
             .await
-            .map_err(|e| bad_request(e.to_string()))?;
+            .map_err(|e| bad_request(e.to_string()))?
+            .ok_or_else(|| bad_request("The provided SQL does not contain a query"))?;
             text = Some(sql.query);
             udfs = Some(api_udfs);
             is_preview = sql.preview;
@@ -469,7 +470,7 @@ pub async fn validate_query(
 
     let pipeline_graph_validation_result =
         match compile_sql(validate_query_post.query, &udfs, 1, &auth_data, &client).await {
-            Ok(CompiledSql { mut program, .. }) => {
+            Ok(Some(CompiledSql { mut program, .. })) => {
                 optimizations::optimize(&mut program.graph);
                 let nodes = program
                     .graph
@@ -499,12 +500,19 @@ pub async fn validate_query(
 
                 QueryValidationResult {
                     graph: Some(PipelineGraph { nodes, edges }),
-                    errors: None,
+                    errors: vec![],
+                    missing_query: false,
                 }
             }
+            Ok(None) => QueryValidationResult {
+                graph: None,
+                errors: vec![],
+                missing_query: true,
+            },
             Err(e) => QueryValidationResult {
                 graph: None,
-                errors: Some(vec![e.to_string()]),
+                errors: vec![e.to_string()],
+                missing_query: false,
             },
         };
 

diff --git a/arroyo-rpc/src/api_types/pipelines.rs b/arroyo-rpc/src/api_types/pipelines.rs
@@ -2,6 +2,7 @@ use crate::api_types::udfs::Udf;
 use crate::grpc as grpc_proto;
 use crate::grpc::api as api_proto;
 use serde::{Deserialize, Serialize};
+use serde_json::Value;
 use utoipa::ToSchema;
 
 #[derive(Serialize, Deserialize, Clone, Debug, ToSchema)]
@@ -15,7 +16,9 @@ pub struct ValidateQueryPost {
 #[serde(rename_all = "camelCase")]
 pub struct QueryValidationResult {
     pub graph: Option<PipelineGraph>,
-    pub errors: Option<Vec<String>>,
+    #[serde(default)]
+    pub errors: Vec<String>,
+    pub missing_query: bool,
 }
 
 #[derive(Serialize, Deserialize, Clone, Debug, ToSchema)]
@@ -165,7 +168,7 @@ pub struct OutputData {
     pub operator_id: String,
     pub timestamp: u64,
     pub key: String,
-    pub value: String,
+    pub value: Value,
 }
 
 impl From<grpc_proto::OutputData> for OutputData {
@@ -174,7 +177,8 @@ impl From<grpc_proto::OutputData> for OutputData {
             operator_id: value.operator_id,
             timestamp: value.timestamp,
             key: value.key,
-            value: value.value,
+            value: serde_json::from_str(&value.value)
+                .expect("Received non-JSON data from web sink"),
         }
     }
 }
diff --git a/arroyo-sql-macro/src/lib.rs b/arroyo-sql-macro/src/lib.rs
@@ -181,6 +181,7 @@ fn get_pipeline_module(
         },
     )
     .unwrap()
+    .unwrap()
     .program;
 
     let function = program.make_graph_function();