Add detailed diagnostics for validate_sql API

mwylde · mwylde · commit a980af359552 · 2026-08-03T14:47:18.000-07:00
diff --git a/crates/arroyo-api/src/lib.rs b/crates/arroyo-api/src/lib.rs
@@ -356,6 +356,9 @@ impl IntoResponse for HttpError {
         GlobalUdf,
         GlobalUdfCollection,
         BadData,
+        SqlDiagnostic,
+        SqlSpan,
+        SqlLocation
     )),
     tags(
         (name = "ping", description = "Ping endpoint"),
diff --git a/crates/arroyo-api/src/pipelines.rs b/crates/arroyo-api/src/pipelines.rs
@@ -17,7 +17,7 @@ use crate::{compiler_service, connection_profiles, jobs, types};
 use arroyo_datastream::default_sink;
 use arroyo_rpc::api_types::pipelines::{
     FailureReason, Job, Pipeline, PipelinePatch, PipelinePost, PipelineRestart, PreviewPost,
-    QueryValidationResult, StopType, ValidateQueryPost,
+    QueryValidationResult, SqlDiagnostic, SqlLocation, SqlSpan, StopType, ValidateQueryPost,
 };
 use arroyo_rpc::api_types::udfs::{GlobalUdf, Udf, UdfLanguage};
 use arroyo_rpc::api_types::{JobCollection, PaginationQueryParams, PipelineCollection};
@@ -28,7 +28,7 @@ use arroyo_datastream::logical::{
     ChainedLogicalOperator, LogicalNode, LogicalProgram, OperatorChain, OperatorName,
 };
 use arroyo_formats::ser::ArrowSerializer;
-use arroyo_planner::{ArroyoSchemaProvider, CompiledSql, SqlConfig};
+use arroyo_planner::{ArroyoSchemaProvider, CompiledSql, PlannerError, SqlConfig};
 use arroyo_rpc::formats::Format;
 use arroyo_rpc::grpc::rpc::compiler_grpc_client::CompilerGrpcClient;
 use arroyo_rpc::public_ids::{IdTypes, generate_id};
@@ -65,7 +65,7 @@ async fn compile_sql(
     auth_data: &AuthData,
     validate_only: bool,
     db: &DatabaseSource,
-) -> Result<CompiledSql, ErrorResp> {
+) -> Result<Result<CompiledSql, PlannerError>, ErrorResp> {
     let mut schema_provider = ArroyoSchemaProvider::new();
 
     let global_udfs = fetch_get_udfs(&db.client().await?, &auth_data.organization_id)
@@ -174,15 +174,14 @@ async fn compile_sql(
         schema_provider.add_connection_profile(profile);
     }
 
-    arroyo_planner::parse_and_get_program(
+    Ok(arroyo_planner::parse_and_get_program(
         &query,
         schema_provider,
         SqlConfig {
             default_parallelism: parallelism,
         },
     )
-    .await
-    .map_err(|err| bad_request(err.to_string()))
+    .await)
 }
 
 fn set_parallelism(program: &mut LogicalProgram, parallelism: usize) {
@@ -569,15 +568,15 @@ pub async fn validate_query(
         true,
         &state.database,
     )
-    .await
+    .await?
     {
         Ok(CompiledSql { program, .. }) => QueryValidationResult {
             graph: Some(program.try_into().map_err(log_and_map)?),
             errors: vec![],
         },
         Err(e) => QueryValidationResult {
             graph: None,
-            errors: vec![e.message],
+            errors: e.diagnostics,
         },
     };
 
@@ -660,7 +659,7 @@ async fn create_pipeline_inner(
         false,
         &state.database,
     )
-    .await?;
+    .await??;
 
     let pipeline_id = create_pipeline_int(
         pipeline_post.name,
@@ -718,7 +717,7 @@ pub async fn create_preview_pipeline(
         false,
         &state.database,
     )
-    .await?;
+    .await??;
 
     let pipeline_id = create_pipeline_int(
         format!("preview_{}", to_millis(SystemTime::now())),
diff --git a/crates/arroyo-api/src/rest_utils.rs b/crates/arroyo-api/src/rest_utils.rs
@@ -1,4 +1,5 @@
 use crate::{AuthData, cloud};
+use arroyo_planner::PlannerError;
 use arroyo_rpc::log_event;
 use axum::Json;
 use axum::extract::rejection::JsonRejection;
@@ -67,6 +68,20 @@ pub fn map_delete_err(name: &str, user: &str, error: DbError) -> ErrorResp {
     }
 }
 
+impl From<PlannerError> for ErrorResp {
+    fn from(value: PlannerError) -> Self {
+        let mut message = format!("Failed to plan query: {}", value.error);
+        for d in &value.diagnostics {
+            message.push_str(&format!("\n * {}", d.message));
+        }
+
+        ErrorResp {
+            status_code: StatusCode::BAD_REQUEST,
+            message,
+        }
+    }
+}
+
 impl From<DbError> for ErrorResp {
     fn from(value: DbError) -> Self {
         match value {
diff --git a/crates/arroyo-planner/src/lib.rs b/crates/arroyo-planner/src/lib.rs
@@ -24,7 +24,9 @@ use arroyo_datastream::WindowType;
 
 use builder::NamedNode;
 use datafusion::common::tree_node::TreeNode;
-use datafusion::common::{Column, DFSchema, Result, ScalarValue, not_impl_err, plan_err};
+use datafusion::common::{
+    Column, DFSchema, Diagnostic, Location, Result, ScalarValue, Span, not_impl_err, plan_err,
+};
 use datafusion::datasource::DefaultTableSource;
 #[allow(deprecated)]
 use datafusion::prelude::SessionConfig;
@@ -64,6 +66,7 @@ use arroyo_connectors::connectors;
 use arroyo_datastream::logical::LogicalProgram;
 use arroyo_datastream::optimizers::ChainingOptimizer;
 use arroyo_operator::connector::Connection;
+use arroyo_rpc::api_types::pipelines::{SqlDiagnostic, SqlLocation, SqlSpan};
 use arroyo_rpc::df::ArroyoSchema;
 use arroyo_rpc::{TIMESTAMP_FIELD, duration_from_sql};
 use arroyo_udf_host::ParsedUdfFile;
@@ -79,6 +82,7 @@ use datafusion::prelude::col;
 use sqlparser::ast::{OneOrManyWithParens, Statement};
 use sqlparser::dialect::ArroyoDialect;
 use sqlparser::parser::{Parser, ParserError};
+use sqlparser::tokenizer::{TokenWithSpan, Tokenizer};
 use std::any::Any;
 use std::time::{Duration, SystemTime};
 use std::{collections::HashMap, sync::Arc};
@@ -548,17 +552,57 @@ impl Default for SqlConfig {
     }
 }
 
+#[derive(Debug)]
+pub struct PlannerError {
+    pub error: String,
+    pub diagnostics: Vec<SqlDiagnostic>,
+}
+
 pub async fn parse_and_get_program(
     query: &str,
     schema_provider: ArroyoSchemaProvider,
     config: SqlConfig,
-) -> Result<CompiledSql> {
+) -> Result<CompiledSql, PlannerError> {
     let query = query.to_string();
 
     if query.trim().is_empty() {
-        return plan_err!("Query is empty");
-    }
-    parse_and_get_arrow_program(query, schema_provider, config).await
+        return Err(PlannerError {
+            error: "Query is empty!".to_string(),
+            diagnostics: vec![],
+        });
+    }
+    parse_and_get_arrow_program(query, schema_provider, config)
+        .await
+        .map_err(|e| {
+            let diagnostics = e
+                .iter()
+                .map(|error| {
+                    let diagnostic = error.diagnostic();
+                    SqlDiagnostic {
+                        message: diagnostic
+                            .map(|diagnostic| diagnostic.message.clone())
+                            .unwrap_or_else(|| error.to_string()),
+                        span: diagnostic.and_then(|diagnostic| {
+                            diagnostic.span.map(|span| SqlSpan {
+                                start: SqlLocation {
+                                    line: span.start.line,
+                                    column: span.start.column,
+                                },
+                                end: SqlLocation {
+                                    line: span.end.line,
+                                    column: span.end.column,
+                                },
+                            })
+                        }),
+                    }
+                })
+                .collect();
+
+            PlannerError {
+                error: e.to_string(),
+                diagnostics,
+            }
+        })
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
diff --git a/crates/arroyo-planner/src/tables.rs b/crates/arroyo-planner/src/tables.rs
@@ -18,7 +18,7 @@ use arroyo_rpc::formats::{BadData, Format, Framing, JsonCompression, JsonFormat}
 use arroyo_rpc::grpc::api::ConnectorOp;
 use arroyo_types::ArroyoExtensionType;
 use datafusion::common::tree_node::{TreeNode, TreeNodeRecursion, TreeNodeVisitor};
-use datafusion::common::{Column, DataFusionError, plan_err};
+use datafusion::common::{Column, DataFusionError, Diagnostic, Span, plan_err};
 use datafusion::common::{
     DFSchema, Result, ScalarValue, config::ConfigOptions, plan_datafusion_err,
 };
@@ -57,8 +57,8 @@ use datafusion::{
     },
 };
 use itertools::Itertools;
-use sqlparser::ast;
-use sqlparser::ast::TableConstraint;
+use sqlparser::ast::{self, FunctionArguments, Spanned, TableConstraint};
+use std::ops::ControlFlow;
 use std::sync::Arc;
 use std::{collections::HashMap, time::Duration};
 use tracing::warn;
@@ -140,7 +140,10 @@ fn produce_optimized_plan(
     statement: &Statement,
     schema_provider: &ArroyoSchemaProvider,
 ) -> Result<LogicalPlan> {
-    let sql_to_rel = SqlToRel::new(schema_provider);
+    let sql_to_rel = SqlToRel::new_with_options(
+        schema_provider,
+        datafusion::sql::planner::ParserOptions::new().with_collect_spans(true),
+    );
 
     let plan = sql_to_rel.sql_statement_to_plan(statement.clone())?;
 
diff --git a/crates/arroyo-planner/src/types.rs b/crates/arroyo-planner/src/types.rs
@@ -1,15 +1,14 @@
-use std::{sync::Arc, time::SystemTime};
-
 use arrow::datatypes::{DataType, Field};
-use datafusion::common::{Result, plan_datafusion_err, plan_err};
-
 use arrow_schema::{DECIMAL_DEFAULT_SCALE, DECIMAL128_MAX_PRECISION, IntervalUnit, TimeUnit};
 use arroyo_types::ArroyoExtensionType;
-use datafusion::error::DataFusionError;
+use datafusion::common::{
+    DataFusionError, Diagnostic, Result, Span, plan_datafusion_err, plan_err,
+};
 use datafusion::sql::sqlparser::ast::{
-    ArrayElemTypeDef, DataType as SQLDataType, ExactNumberInfo, TimezoneInfo,
+    ArrayElemTypeDef, DataType as SQLDataType, ExactNumberInfo, Spanned, TimezoneInfo,
 };
 use itertools::Itertools;
+use std::{sync::Arc, time::SystemTime};
 // Pulled from DataFusion
 
 pub(crate) fn convert_data_type(
@@ -117,7 +116,13 @@ fn convert_simple_data_type(
 
             Ok(DataType::Struct(fields.into()))
         }
-        _ => return plan_err!("Unsupported SQL type {sql_type:?}"),
+        SQLDataType::Custom(name, _) => {
+            let message = format!("Unsupported SQL type '{name}'");
+            let diagnostic =
+                Diagnostic::new_error(message.clone(), Span::try_from_sqlparser_span(name.span()));
+            return plan_err!("{message}"; diagnostic = diagnostic);
+        }
+        _ => return plan_err!("Unsupported SQL type '{sql_type}'"),
     };
 
     Ok((dt?, None))
diff --git a/crates/arroyo-rpc/src/api_types/pipelines.rs b/crates/arroyo-rpc/src/api_types/pipelines.rs
@@ -1,9 +1,9 @@
-use std::collections::HashMap;
-
 use crate::api_types::udfs::Udf;
 use crate::errors::ErrorDomain;
 use crate::grpc as grpc_proto;
 use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
+use std::fmt::{Display, Formatter};
 use utoipa::ToSchema;
 
 #[derive(Serialize, Deserialize, Clone, Debug, ToSchema)]
@@ -17,7 +17,34 @@ pub struct ValidateQueryPost {
 #[serde(rename_all = "snake_case")]
 pub struct QueryValidationResult {
     pub graph: Option<PipelineGraph>,
-    pub errors: Vec<String>,
+    pub errors: Vec<SqlDiagnostic>,
+}
+
+#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, ToSchema)]
+pub struct SqlLocation {
+    pub line: u64,
+    pub column: u64,
+}
+
+#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, ToSchema)]
+pub struct SqlSpan {
+    pub start: SqlLocation,
+    pub end: SqlLocation,
+}
+
+#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, ToSchema)]
+pub struct SqlDiagnostic {
+    pub message: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub span: Option<SqlSpan>,
+}
+impl SqlDiagnostic {
+    pub fn message(message: impl Into<String>) -> Self {
+        Self {
+            message: message.into(),
+            span: None,
+        }
+    }
 }
 
 #[derive(Serialize, Deserialize, Clone, Debug, ToSchema)]
diff --git a/crates/arroyo/src/run.rs b/crates/arroyo/src/run.rs
@@ -167,7 +167,7 @@ async fn run_pipeline(
     if !errors.errors.is_empty() {
         eprintln!("There were some issues with the provided query");
         for error in errors.errors {
-            eprintln!("  * {error}");
+            eprintln!("  * {}", error.message);
         }
         exit(1);
     }
diff --git a/crates/integ/tests/api_tests.rs b/crates/integ/tests/api_tests.rs
@@ -223,7 +223,7 @@ async fn basic_pipeline() {
         .unwrap()
         .into_inner();
 
-    assert_eq!(valid.errors, Vec::<String>::new());
+    assert!(valid.errors.is_empty());
     assert!(valid.graph.is_some());
 
     let (pipeline_id, job_id, _) = start_and_monitor(test_id, &query, &[], 10).await.unwrap();
diff --git a/webui/src/gen/api-types.ts b/webui/src/gen/api-types.ts
@@ -907,7 +907,7 @@ export interface components {
             message_name?: string | null;
         };
         QueryValidationResult: {
-            errors: string[];
+            errors: components["schemas"]["SqlDiagnostic"][];
             graph?: components["schemas"]["PipelineGraph"] | null;
         };
         RawBytesFormat: Record<string, never>;
@@ -936,6 +936,20 @@ export interface components {
             required?: boolean;
             readonly sql_name?: string | null;
         });
+        SqlDiagnostic: {
+            message: string;
+            span?: components["schemas"]["SqlSpan"] | null;
+        };
+        SqlLocation: {
+            /** Format: int64 */
+            column: number;
+            /** Format: int64 */
+            line: number;
+        };
+        SqlSpan: {
+            end: components["schemas"]["SqlLocation"];
+            start: components["schemas"]["SqlLocation"];
+        };
         /** @enum {string} */
         StopType: "none" | "checkpoint" | "graceful" | "immediate" | "force";
         StructField: {
diff --git a/webui/src/lib/data_fetching.ts b/webui/src/lib/data_fetching.ts
@@ -33,6 +33,7 @@ export type SubtaskCheckpointGroup = schemas['SubtaskCheckpointGroup'];
 export type GlobalUdf = schemas['GlobalUdf'];
 export type PipelineLocalUdf = schemas['Udf'];
 export type UdfValidationResult = schemas['UdfValidationResult'];
+export type SqlDiagnostic = schemas['SqlDiagnostic'];
 
 const base = window.__ARROYO_BASENAME.replace(/\/$/, '') || '';
 const BASE_URL = `${base}/api`;
diff --git a/webui/src/routes/pipelines/CodeEditor.tsx b/webui/src/routes/pipelines/CodeEditor.tsx
diff --git a/webui/src/routes/pipelines/CreatePipeline.tsx b/webui/src/routes/pipelines/CreatePipeline.tsx
diff --git a/webui/src/routes/pipelines/PipelineEditorTabs.tsx b/webui/src/routes/pipelines/PipelineEditorTabs.tsx

Original file line number	Diff line number	Diff line change
`@@ -167,7 +167,7 @@ async fn run_pipeline(`
`167`	`167`	`if !errors.errors.is_empty() {`
`168`	`168`	`eprintln!("There were some issues with the provided query");`
`169`	`169`	`for error in errors.errors {`
`170`		`- eprintln!(" * {error}");`
	`170`	`+ eprintln!(" * {}", error.message);`
`171`	`171`	`}`
`172`	`172`	`exit(1);`
`173`	`173`	`}`