[adapters] Hoist adhoc query processing to return errors early

Karakatiza666 · Karakatiza666 · commit e783645f5741 · 2026-06-09T20:42:00.000Z
[web-console] Improve ad-hoc error rendering and formatting (remove message duplication)

DataFusion errors raised during query setup were not reported over HTTP:
the backend had already responded with status 200 and simply interrupted
the stream when the error was thrown.

Start the query before deciding the HTTP status. If it fails to start,
respond with HTTP 400 and the error message so the client can surface it.
If it starts cleanly, respond with 200 OK and stream the results as before.

This adds no significant delay - only a few extra milliseconds for query
planning and SQL types header generation; the initial response does not
wait for the first data bytes.

Signed-off-by: Karakatiza666 &lt;bulakh.96@gmail.com&gt;
diff --git a/crates/adapters/src/adhoc.rs b/crates/adapters/src/adhoc.rs
@@ -10,8 +10,8 @@ use datafusion::prelude::*;
 use datafusion::sql::parser::{DFParserBuilder, Statement as DFStatement};
 use datafusion::sql::sqlparser::dialect::dialect_from_str;
 use executor::{
-    hash_query_result, infallible_from_bytestring, stream_arrow_query, stream_json_query,
-    stream_parquet_query, stream_text_query,
+    execute_adhoc_stream, hash_query_result, infallible_from_bytestring, stream_arrow_query,
+    stream_json_query, stream_parquet_query, stream_text_query,
 };
 use feldera_types::query::{AdHocResultFormat, AdhocQueryArgs, MAX_WS_FRAME_SIZE};
 use futures_util::StreamExt;
@@ -46,9 +46,24 @@ async fn adhoc_query_handler(
     mut ws_session: WsSession,
     args: AdhocQueryArgs,
 ) -> Result<(), Closed> {
+    // Set up execution before encoding. Planning and execute-time-setup
+    // errors (e.g. selecting from a non-materialized source) surface here;
+    // report them on the websocket and close, matching the per-format error
+    // handling below.
+    let (record_stream, schema) = match execute_adhoc_stream(df).await {
+        Ok(stream_and_schema) => stream_and_schema,
+        Err(e) => {
+            ws_session
+                .text(serde_json::to_string(&e).unwrap_or_else(|_| e.to_string()))
+                .await?;
+            ws_close(ws_session, CloseCode::Error).await;
+            return Ok(());
+        }
+    };
+
     match args.format {
         AdHocResultFormat::Text => {
-            let mut stream = Box::pin(stream_text_query(df));
+            let mut stream = Box::pin(stream_text_query(record_stream, schema));
             while let Some(res) = stream.next().await {
                 match res {
                     Ok(text) => {
@@ -63,7 +78,7 @@ async fn adhoc_query_handler(
             }
         }
         AdHocResultFormat::Json => {
-            let mut stream = Box::pin(stream_json_query(df));
+            let mut stream = Box::pin(stream_json_query(record_stream));
             while let Some(res) = stream.next().await {
                 match res {
                     Ok(byte_string) => {
@@ -80,7 +95,7 @@ async fn adhoc_query_handler(
             }
         }
         AdHocResultFormat::ArrowIpc => {
-            let mut stream = Box::pin(stream_arrow_query(df));
+            let mut stream = Box::pin(stream_arrow_query(record_stream, schema));
             while let Some(res) = stream.next().await {
                 match res {
                     Ok(bytes) => {
@@ -103,7 +118,7 @@ async fn adhoc_query_handler(
             }
         }
         AdHocResultFormat::Parquet => {
-            let mut stream = Box::pin(stream_parquet_query(df));
+            let mut stream = Box::pin(stream_parquet_query(record_stream, schema));
             while let Some(res) = stream.next().await {
                 match res {
                     Ok(bytes) => ws_session.binary(bytes).await?,
@@ -124,7 +139,7 @@ async fn adhoc_query_handler(
             }
         }
         AdHocResultFormat::Hash => {
-            let hash_result = hash_query_result(df).await;
+            let hash_result = hash_query_result(record_stream, schema).await;
             match hash_result {
                 Ok(hash) => {
                     ws_session.text(hash).await?;
@@ -445,28 +460,33 @@ pub(crate) async fn stream_adhoc_result(
 ) -> Result<HttpResponse, PipelineError> {
     let df = execute_sql(controller, &args.sql).await?;
 
-    // Note that once we are in the stream!{} macros any error that occurs will lead to the connection
-    // in the manager being terminated and a 500 error being returned to the client.
-    // We can't return an error in a stream that is already Response::Ok.
-    //
-    // Sometimes things do tend to fail inside the stream!{} macro, e.g., "select 1/0;" will cause a
-    // division by zero error during query execution. So we return errors according to the chosen
-    // format for text and json, and for parquet we return the 500 error.
+    // Set up execution before committing a response status. Planning and
+    // execute-time-setup errors — e,g, selecting from a non-materialized source —
+    // surface here and are returned as a regular `PipelineError` (HTTP 400),
+    // so the client sees the message rather than a truncated `200 OK` body.
+    let (record_stream, schema) = execute_adhoc_stream(df).await?;
+
+    // Once we hand a stream to `.streaming(...)` the `200 OK` status is
+    // already yielded, so an error raised *mid-stream* (e.g. "select
+    // 1/0;" failing on a later row) can no longer change the status. For
+    // text and json we fold such errors into the response body; for arrow
+    // and parquet the body is simply terminated (and the manager surfaces a 500).
     match args.format {
         AdHocResultFormat::Text => Ok(HttpResponse::Ok()
             .content_type(mime::TEXT_PLAIN)
-            .streaming::<_, Infallible>(infallible_from_bytestring(stream_text_query(df), |e| {
-                format!("ERROR: {}", e).into()
-            }))),
+            .streaming::<_, Infallible>(infallible_from_bytestring(
+                stream_text_query(record_stream, schema),
+                |e| format!("ERROR: {}", e).into(),
+            ))),
         AdHocResultFormat::Json => Ok(HttpResponse::Ok()
             .content_type(mime::APPLICATION_JSON)
             .streaming::<_, Infallible>(infallible_from_bytestring(
-            stream_json_query(df),
+            stream_json_query(record_stream),
             |e| serde_json::to_string(&e).unwrap().into(),
         ))),
         AdHocResultFormat::ArrowIpc => Ok(HttpResponse::Ok()
             .content_type(mime::APPLICATION_OCTET_STREAM)
-            .streaming(stream_arrow_query(df))),
+            .streaming(stream_arrow_query(record_stream, schema))),
         AdHocResultFormat::Parquet => {
             let file_name = format!(
                 "results_{}.parquet",
@@ -475,11 +495,11 @@ pub(crate) async fn stream_adhoc_result(
             Ok(HttpResponse::Ok()
                 .insert_header(header::ContentDisposition::attachment(file_name))
                 .content_type(mime::APPLICATION_OCTET_STREAM)
-                .streaming(stream_parquet_query(df)))
+                .streaming(stream_parquet_query(record_stream, schema)))
         }
         AdHocResultFormat::Hash => Ok(HttpResponse::Ok()
             .content_type(mime::TEXT_PLAIN)
-            .body(hash_query_result(df).await?)),
+            .body(hash_query_result(record_stream, schema).await?)),
     }
 }
 
@@ -684,4 +704,52 @@ mod tests {
         let total_rows: usize = batches.iter().map(|b| b.num_rows()).sum();
         assert_eq!(total_rows, 1);
     }
+
+    /// Selecting from a non-materialized source fails during *execution
+    /// setup*, not planning. `execute_adhoc_stream` must surface that as an
+    /// `Err`, letting the HTTP handler return a non-2xx status before
+    /// committing a `200 OK` whose body would otherwise be silently
+    /// truncated. This is what gives the HTTP transport the same up-front
+    /// error visibility the WebSocket transport already has.
+    #[tokio::test]
+    async fn non_materialized_select_surfaces_error_before_streaming() {
+        use crate::adhoc::table::AdHocTable;
+        use datafusion::arrow::datatypes::{DataType, Field, Schema};
+        use feldera_types::program_schema::SqlIdentifier;
+        use std::collections::BTreeMap;
+        use std::sync::Weak;
+
+        let schema = Arc::new(Schema::new(vec![Field::new("x", DataType::Int64, false)]));
+        let ctx = SessionContext::new_with_state(test_state());
+
+        // A view (no input handle) that is *not* materialized.
+        let table = Arc::new(AdHocTable::new(
+            false, // materialized
+            false, // indexed
+            Weak::new(),
+            None, // input_handle: view, not a base table
+            SqlIdentifier::new("v", false),
+            schema,
+        ));
+        ctx.register_table("v", table).unwrap();
+
+        // The scan reads a consistent snapshot from the session config; an
+        // empty one suffices because execution fails the materialization
+        // check before touching any data.
+        let mut state = ctx.state();
+        set_snapshot(&mut state, Arc::new(BTreeMap::new()));
+
+        // Planning succeeds: the materialization check lives in physical
+        // execution setup, not planning.
+        let df = execute_sql_with_state(state, "SELECT * FROM v")
+            .await
+            .expect("planning a select over a non-materialized view should succeed");
+
+        // The error must surface here, before any response status is set.
+        let msg = match execute_adhoc_stream(df).await {
+            Ok(_) => panic!("selecting from a non-materialized source must fail at setup"),
+            Err(e) => format!("{e}"),
+        };
+        assert!(msg.contains("non-materialized"), "unexpected error: {msg}");
+    }
 }
diff --git a/crates/adapters/src/adhoc/executor.rs b/crates/adapters/src/adhoc/executor.rs
@@ -1,4 +1,5 @@
 use arrow::array::RecordBatch;
+use arrow::datatypes::SchemaRef;
 use arrow::ipc::convert::IpcSchemaEncoder;
 use arrow::ipc::writer::StreamWriter;
 use arrow::util::pretty::pretty_format_batches;
@@ -40,6 +41,31 @@ fn execute_stream(df: DataFrame) -> Receiver<DFResult<SendableRecordBatchStream>
     rx
 }
 
+/// Drives physical planning and execution setup for `df`, returning the
+/// record-batch stream together with its schema.
+///
+/// Planning and execute-time-setup errors — e.g. selecting from a
+/// non-materialized table — surface here, before any result data is produced.
+/// Hoisting this step out of the per-format encoders lets the HTTP handler
+/// translate such an error into a proper 4xx response
+/// instead of a `200 OK` whose body is silently truncated.
+pub(crate) async fn execute_adhoc_stream(
+    df: DataFrame,
+) -> Result<(SendableRecordBatchStream, SchemaRef), PipelineError> {
+    let schema = df.schema().inner().clone();
+    let stream = execute_stream(df)
+        .await
+        .map_err(|e| PipelineError::AdHocQueryError {
+            error: e.to_string(),
+            df: None,
+        })?
+        .map_err(|e| PipelineError::AdHocQueryError {
+            error: e.to_string(),
+            df: Some(Box::new(e)),
+        })?;
+    Ok((stream, schema))
+}
+
 pub(crate) fn infallible_from_bytestring(
     fallible_stream: impl Stream<Item = Result<ByteString, PipelineError>>,
     map_err: impl Fn(PipelineError) -> Bytes + 'static,
@@ -53,13 +79,11 @@ pub(crate) fn infallible_from_bytestring(
 }
 
 pub(crate) fn stream_text_query(
-    df: DataFrame,
+    stream: SendableRecordBatchStream,
+    schema: SchemaRef,
 ) -> impl Stream<Item = Result<ByteString, PipelineError>> {
-    let schema = df.schema().inner().clone();
     try_stream! {
-        let stream_executor = execute_stream(df).await.map_err(|e| PipelineError::AdHocQueryError { error: e.to_string(), df: None })?;
-        let mut stream = stream_executor
-            .map_err(|e| PipelineError::AdHocQueryError { error: e.to_string(), df: Some(Box::new(e)) })?;
+        let mut stream = stream;
 
         let mut headers_sent = false;
         let mut last_line: Option<String> = None;
@@ -159,19 +183,11 @@ impl BatchHasher {
 }
 
 /// Computes an order-independent hash of a DataFrame's result set.
-pub(crate) async fn hash_query_result(df: DataFrame) -> Result<String, PipelineError> {
-    let schema = df.schema().inner().clone();
-
-    let stream_executor = execute_stream(df)
-        .await
-        .map_err(|e| PipelineError::AdHocQueryError {
-            error: e.to_string(),
-            df: None,
-        })?;
-    let mut stream = stream_executor.map_err(|e| PipelineError::AdHocQueryError {
-        error: e.to_string(),
-        df: Some(Box::new(e)),
-    })?;
+pub(crate) async fn hash_query_result(
+    stream: SendableRecordBatchStream,
+    schema: SchemaRef,
+) -> Result<String, PipelineError> {
+    let mut stream = stream;
 
     let mut hasher = BatchHasher::new();
     while let Some(batch) = stream.next().await {
@@ -182,12 +198,10 @@ pub(crate) async fn hash_query_result(df: DataFrame) -> Result<String, PipelineE
 }
 
 pub(crate) fn stream_json_query(
-    df: DataFrame,
+    stream: SendableRecordBatchStream,
 ) -> impl Stream<Item = Result<ByteString, PipelineError>> {
     try_stream! {
-        let stream_executor = execute_stream(df).await.map_err(|e| PipelineError::AdHocQueryError { error: e.to_string(), df: None })?;
-        let mut stream = stream_executor
-            .map_err(|e| PipelineError::AdHocQueryError { error: e.to_string(), df: Some(Box::new(e)) })?;
+        let mut stream = stream;
         while let Some(batch) = stream.next().await {
             let batch = batch.map_err(PipelineError::from)?;
             let mut buf = Vec::with_capacity(4096);
@@ -247,15 +261,11 @@ impl AsyncFileWriter for ChannelWriter {
 /// <https://github.com/apache/arrow-rs/pull/9241>. Once that lands the
 /// per-batch buffering here can be replaced with a direct async sink.
 pub(crate) fn stream_arrow_query(
-    df: DataFrame,
+    stream: SendableRecordBatchStream,
+    schema: SchemaRef,
 ) -> impl Stream<Item = Result<Bytes, DataFusionError>> {
     try_stream! {
-        let schema = df.schema().inner().clone();
-        let mut stream = execute_stream(df)
-            .await
-            .map_err(|e| DataFusionError::Execution(format!(
-                "ad-hoc query worker did not return a stream: {e}"
-            )))??;
+        let mut stream = stream;
 
         // `try_new` writes the schema message to the inner buffer. The
         // `BytesMut` behind `BufMut::writer()` is a single allocation
@@ -294,7 +304,8 @@ pub(crate) fn stream_arrow_query(
 }
 
 pub(crate) fn stream_parquet_query(
-    df: DataFrame,
+    stream: SendableRecordBatchStream,
+    schema: SchemaRef,
 ) -> impl Stream<Item = Result<Bytes, DataFusionError>> {
     // Should probably be smaller than `MAX_WS_FRAME_SIZE`.
     const PARQUET_CHUNK_SIZE: usize = MAX_WS_FRAME_SIZE / 2;
@@ -304,10 +315,7 @@ pub(crate) fn stream_parquet_query(
 
     let mut stream_job = Box::pin(
         async move {
-            let schema = df.schema().inner().clone();
-            let mut stream = execute_stream(df)
-                .await
-                .expect("unable to receive stream")?;
+            let mut stream = stream;
 
             let mut writer = AsyncArrowWriter::try_new(
                 ChannelWriter::new(tx),
@@ -528,8 +536,11 @@ mod tests {
         ctx.register_table("t", Arc::new(mem)).unwrap();
         let df = ctx.sql("SELECT * FROM t ORDER BY id").await.unwrap();
 
+        let (record_stream, schema) = execute_adhoc_stream(df)
+            .await
+            .expect("execute_adhoc_stream failed to set up the query");
         let mut buf = Vec::<u8>::new();
-        let mut stream = Box::pin(stream_arrow_query(df));
+        let mut stream = Box::pin(stream_arrow_query(record_stream, schema));
         while let Some(chunk) = stream.next().await {
             buf.extend_from_slice(&chunk.expect("stream_arrow_query yielded an error"));
         }
@@ -596,8 +607,11 @@ mod tests {
             ctx.register_table("t", Arc::new(mem)).unwrap();
             let df = ctx.sql("SELECT * FROM t").await.unwrap();
 
+            let (record_stream, schema) = execute_adhoc_stream(df)
+                .await
+                .expect("execute_adhoc_stream failed to set up the query");
             let mut buf = Vec::<u8>::new();
-            let mut stream = Box::pin(stream_arrow_query(df));
+            let mut stream = Box::pin(stream_arrow_query(record_stream, schema));
             while let Some(chunk) = stream.next().await {
                 buf.extend_from_slice(&chunk.expect("stream_arrow_query yielded an error"));
             }
diff --git a/js-packages/web-console/src/lib/components/adhoc/Query.svelte b/js-packages/web-console/src/lib/components/adhoc/Query.svelte
@@ -219,7 +219,9 @@
                 </tr>
               {:else if 'error' in row}
                 <tr {style} class={itemHeight} use:selectScope tabindex={-1}>
-                  <td colspan="99999999" class="preset-tonal-error px-2">{row.error}</td>
+                  <td colspan="99999999"
+                    ><div class="rounded bg-error-50-950/50 px-2">{row.error}</div></td
+                  >
                 </tr>
               {:else}
                 <tr {style} class={itemHeight} use:selectScope tabindex={-1}>
diff --git a/js-packages/web-console/src/lib/services/pipelineManager.ts b/js-packages/web-console/src/lib/services/pipelineManager.ts
@@ -841,20 +841,25 @@ const getAuthenticatedFetch = (options?: FetchOptions): typeof globalThis.fetch
 }
 
 function formatValue(details: unknown): string {
+  if (!details) {
+    return ''
+  }
   if (typeof details === 'string') {
     return details
   }
 
-  // Pretty‑print objects, arrays, numbers, booleans, etc.
+  // Pretty‑print objects, arrays, numbers, booleans, etc., dropping any
+  // `error` field that merely duplicates the message shown above it.
   try {
-    return JSON.stringify(details, null, 2)
+    const json = JSON.stringify(details, (key, value) => (key === 'error' ? undefined : value), 2)
+    return json === '{}' ? '' : json
   } catch {
     return String(details)
   }
 }
 
 const apiErrorText = (error: ErrorResponse) => {
-  return `${error.message}${error.details ? `\n${formatValue(error.details)}` : ''}`
+  return `${error.message}\n${formatValue(error.details)}`
 }
 
 const streamingFetch = (
diff --git a/python/tests/runtime/test_adhoc_queries.py b/python/tests/runtime/test_adhoc_queries.py

Original file line number	Diff line number	Diff line change
`@@ -841,20 +841,25 @@ const getAuthenticatedFetch = (options?: FetchOptions): typeof globalThis.fetch`
`841`	`841`	`}`
`842`	`842`
`843`	`843`	`function formatValue(details: unknown): string {`
	`844`	`+ if (!details) {`
	`845`	`+ return ''`
	`846`	`+ }`
`844`	`847`	`if (typeof details === 'string') {`
`845`	`848`	`return details`
`846`	`849`	`}`
`847`	`850`
`848`		`- // Pretty‑print objects, arrays, numbers, booleans, etc.`
	`851`	`+ // Pretty‑print objects, arrays, numbers, booleans, etc., dropping any`
	`852`	+ // `error` field that merely duplicates the message shown above it.
`849`	`853`	`try {`
`850`		`- return JSON.stringify(details, null, 2)`
	`854`	`+ const json = JSON.stringify(details, (key, value) => (key === 'error' ? undefined : value), 2)`
	`855`	`+ return json === '{}' ? '' : json`
`851`	`856`	`} catch {`
`852`	`857`	`return String(details)`
`853`	`858`	`}`
`854`	`859`	`}`
`855`	`860`
`856`	`861`	`const apiErrorText = (error: ErrorResponse) => {`
`857`		- return `${error.message}${error.details ? `\n${formatValue(error.details)}` : ''}`
	`862`	+ return `${error.message}\n${formatValue(error.details)}`
`858`	`863`	`}`
`859`	`864`
`860`	`865`	`const streamingFetch = (`