adapter: report structured SQLSTATEs for dataflow-execution evaluation errors

src/adapter/src/active_compute_sink.rs

@@ -178,7 +178,9 @@ impl ActiveSubscribe {
                 mz_ore::iter::consolidate_update_iter(merged)
             }
             Err(s) => {
-                self.send(PeekResponseUnary::Error(s));
+                self.send(PeekResponseUnary::Error(AdapterError::Unstructured(
+                    anyhow::Error::msg(s),
+                )));
                 return true;
             }
         };

src/adapter/src/coord/catalog_implications.rs

@@ -39,7 +39,7 @@ use mz_catalog::memory::objects::{
 };
 use mz_cloud_resources::VpcEndpointConfig;
 use mz_compute_client::logging::LogVariant;
-use mz_compute_client::protocol::response::PeekResponse;
+use mz_compute_client::protocol::response::{PeekError, PeekResponse};
 use mz_controller::clusters::{ClusterRole, ReplicaConfig};
 use mz_controller_types::{ClusterId, ReplicaId};
 use mz_ore::collections::CollectionExt;
@@ -911,7 +911,8 @@ impl Coordinator {
             if !peeks_to_drop.is_empty() {
                 for (dep, uuid) in peeks_to_drop {
                     if let Some(pending_peek) = self.remove_pending_peek(&uuid) {
-                        let cancel_reason = PeekResponse::Error(dep.query_terminated_error());
+                        let cancel_reason =
+                            PeekResponse::Error(PeekError::internal(dep.query_terminated_error()));
                         self.controller
                             .compute
                             .cancel_peek(pending_peek.cluster_id, uuid, cancel_reason)

src/adapter/src/coord/peek.rs

@@ -85,13 +85,13 @@ pub(crate) struct PendingPeek {
 #[derive(Debug)]
 pub enum PeekResponseUnary {
     Rows(Box<dyn RowIterator + Send + Sync>),
-    Error(String),
+    Error(AdapterError),
     Canceled,
     /// A dependency was dropped during execution.
     ///
-    /// N.B. This is a bit of a workaround for the fact that our Error variant
-    /// is unstructured and right now we specifically care about this error and
-    /// need to render differently based on context.
+    /// N.B. This is a bit of a workaround for the fact that right now we
+    /// specifically care about this error and need to render differently based
+    /// on context.
     DependencyDropped(DroppedDependency),
 }
 
@@ -1040,7 +1040,7 @@ impl crate::coord::Coordinator {
             let rows = match result {
                 Ok(rows) => rows,
                 Err(e) => {
-                    yield PeekResponseUnary::Error(e.to_string());
+                    yield PeekResponseUnary::Error(AdapterError::Unstructured(anyhow::anyhow!(e)));
                     return;
                 }
             };
@@ -1055,7 +1055,11 @@ impl crate::coord::Coordinator {
                         &duration_histogram,
                     ) {
                         Ok((rows, _size_bytes)) => yield PeekResponseUnary::Rows(Box::new(rows)),
-                        Err(e) => yield PeekResponseUnary::Error(e),
+                        Err(e) => {
+                            yield PeekResponseUnary::Error(AdapterError::Unstructured(
+                                anyhow::Error::msg(e),
+                            ))
+                        }
                     }
                 }
                 PeekResponse::Stashed(response) => {
@@ -1212,7 +1216,11 @@ impl crate::coord::Coordinator {
 
                         match result_rows {
                             Ok(result_rows) => yield PeekResponseUnary::Rows(Box::new(result_rows)),
-                            Err(e) => yield PeekResponseUnary::Error(e),
+                            Err(e) => {
+                                yield PeekResponseUnary::Error(AdapterError::Unstructured(
+                                    anyhow::Error::msg(e),
+                                ))
+                            }
                         }
                     }
 
@@ -1227,7 +1235,7 @@ impl crate::coord::Coordinator {
                     yield PeekResponseUnary::Canceled;
                 }
                 PeekResponse::Error(e) => {
-                    yield PeekResponseUnary::Error(e);
+                    yield PeekResponseUnary::Error(e.into());
                 }
             }
         })

src/adapter/src/coord/sequencer/inner.rs

@@ -2952,7 +2952,7 @@ impl Coordinator {
                                 }
                                 PeekResponseUnary::Canceled => break Err(AdapterError::Canceled),
                                 PeekResponseUnary::Error(e) => {
-                                    break Err(AdapterError::Unstructured(anyhow!(e)));
+                                    break Err(e);
                                 }
                                 PeekResponseUnary::DependencyDropped(dep) => {
                                     break Err(dep.to_concurrent_dependency_drop());

src/adapter/src/error.rs

@@ -35,7 +35,7 @@ use mz_sql::rbac;
 use mz_sql::session::vars::VarError;
 use mz_storage_types::connections::ConnectionValidationError;
 use mz_storage_types::controller::StorageError;
-use mz_storage_types::errors::CollectionMissing;
+use mz_storage_types::errors::{CollectionMissing, DataflowError};
 use smallvec::SmallVec;
 use timely::progress::Antichain;
 use tokio::sync::oneshot;
@@ -68,6 +68,12 @@ pub enum AdapterError {
     DuplicateCursor(String),
     /// An error while evaluating an expression.
     Eval(EvalError),
+    /// A structured error produced while executing a dataflow (e.g. evaluating
+    /// an expression over a collection). Distinct from [`AdapterError::Eval`]
+    /// so that the existing dataflow error message (e.g. the `Evaluation
+    /// error:` prefix) is preserved, while `code` still derives a precise
+    /// SQLSTATE from the inner error.
+    Dataflow(Box<DataflowError>),
     /// An error occurred while planning the statement.
     Explain(ExplainError),
     /// The ID allocator exhausted all valid IDs.
@@ -459,6 +465,20 @@ fn eval_error_code(err: &EvalError) -> SqlState {
     }
 }
 
+/// Maps a [`DataflowError`] produced during dataflow execution to a SQLSTATE.
+///
+/// Evaluation errors are routed through [`eval_error_code`] so they get the
+/// same precise codes as constant folding; the remaining variants are genuine
+/// internal/source errors that stay `INTERNAL_ERROR`.
+fn dataflow_error_code(err: &DataflowError) -> SqlState {
+    match err {
+        DataflowError::EvalError(e) => eval_error_code(e),
+        DataflowError::DecodeError(_)
+        | DataflowError::SourceError(_)
+        | DataflowError::EnvelopeError(_) => SqlState::INTERNAL_ERROR,
+    }
+}
+
 impl AdapterError {
     pub fn into_response(self, severity: Severity) -> ErrorResponse {
         ErrorResponse {
@@ -486,6 +506,10 @@ impl AdapterError {
             }
             AdapterError::Catalog(c) => c.detail(),
             AdapterError::Eval(e) => e.detail(),
+            AdapterError::Dataflow(e) => match &**e {
+                DataflowError::EvalError(e) => e.detail(),
+                _ => None,
+            },
             AdapterError::RelationOutsideTimeDomain { relations, names } => Some(format!(
                 "The following relations in the query are outside the transaction's time domain:\n{}\n{}",
                 relations
@@ -692,6 +716,10 @@ impl AdapterError {
             ),
             AdapterError::Catalog(c) => c.hint(),
             AdapterError::Eval(e) => e.hint(),
+            AdapterError::Dataflow(e) => match &**e {
+                DataflowError::EvalError(e) => e.hint(),
+                _ => None,
+            },
             AdapterError::InvalidClusterReplicaAz { expected, az: _ } => {
                 Some(if expected.is_empty() {
                     "No availability zones configured; do not specify AVAILABILITY ZONE".into()
@@ -792,6 +820,7 @@ impl AdapterError {
             // exhaustively so the catch-all `INTERNAL_ERROR` no longer applies
             // to errors that are really the user's fault. See SQL-326.
             AdapterError::Eval(e) => eval_error_code(e),
+            AdapterError::Dataflow(e) => dataflow_error_code(e),
             AdapterError::Explain(_) => SqlState::INTERNAL_ERROR,
             AdapterError::IdExhaustionError => SqlState::INTERNAL_ERROR,
             AdapterError::Internal(_) => SqlState::INTERNAL_ERROR,
@@ -1102,6 +1131,7 @@ impl fmt::Display for AdapterError {
                 write!(f, "cursor {} already exists", name.quoted())
             }
             AdapterError::Eval(e) => e.fmt(f),
+            AdapterError::Dataflow(e) => e.fmt(f),
             AdapterError::Explain(e) => e.fmt(f),
             AdapterError::IdExhaustionError => f.write_str("ID allocator exhausted all valid IDs"),
             AdapterError::Internal(e) => write!(f, "internal error: {}", e),
@@ -1412,6 +1442,19 @@ impl From<EvalError> for AdapterError {
     }
 }
 
+impl From<mz_compute_client::protocol::response::PeekError> for AdapterError {
+    fn from(e: mz_compute_client::protocol::response::PeekError) -> AdapterError {
+        use mz_compute_client::protocol::response::PeekError;
+        match e {
+            // Preserve the structured dataflow error so that evaluation errors
+            // receive the same precise SQLSTATE that constant folding produces
+            // (see `code`), while keeping the existing error message.
+            PeekError::Dataflow(e) => AdapterError::Dataflow(e),
+            PeekError::Internal(e) => AdapterError::Unstructured(anyhow::Error::msg(e)),
+        }
+    }
+}
+
 impl From<ExplainError> for AdapterError {
     fn from(e: ExplainError) -> AdapterError {
         match e {

src/compute-client/src/controller.rs

@@ -170,7 +170,7 @@ impl PeekNotification {
                     result_size: u64::cast_from(result_size),
                 }
             }
-            PeekResponse::Error(err) => Self::Error(err.clone()),
+            PeekResponse::Error(err) => Self::Error(err.to_string()),
             PeekResponse::Canceled => Self::Canceled,
         }
     }

src/compute-client/src/controller/instance.rs

@@ -1221,7 +1221,9 @@ impl Instance {
             self.deliver_response(response);
         }
         for uuid in to_drop {
-            let response = PeekResponse::Error(ERROR_TARGET_REPLICA_FAILED.into());
+            let response = PeekResponse::Error(crate::protocol::response::PeekError::internal(
+                ERROR_TARGET_REPLICA_FAILED,
+            ));
             self.finish_peek(uuid, response);
         }
 

src/compute-client/src/protocol/response.rs

@@ -9,12 +9,16 @@
 
 //! Compute protocol responses.
 
+use std::fmt;
+
+use mz_expr::EvalError;
 use mz_expr::row::RowCollection;
 use mz_ore::cast::CastFrom;
 use mz_ore::tracing::OpenTelemetryContext;
 use mz_persist_client::batch::ProtoBatch;
 use mz_persist_types::ShardId;
 use mz_repr::{GlobalId, RelationDesc, Timestamp, UpdateCollection};
+use mz_storage_types::errors::DataflowError;
 use serde::{Deserialize, Serialize};
 use timely::progress::frontier::Antichain;
 use uuid::Uuid;
@@ -194,7 +198,7 @@ pub enum PeekResponse {
     /// Results of the peek were stashed in persist batches.
     Stashed(Box<StashedPeekResponse>),
     /// Error of an unsuccessful peek.
-    Error(String),
+    Error(PeekError),
     /// The peek was canceled.
     Canceled,
 }
@@ -210,6 +214,50 @@ impl PeekResponse {
     }
 }
 
+/// The error of an unsuccessful peek.
+///
+/// Errors that arise while evaluating the dataflow (e.g. an arithmetic error
+/// computed over a collection) carry a structured [`DataflowError`], so the
+/// adapter can map them to a precise SQLSTATE just like constant-folded
+/// expressions. Errors from the peek machinery itself (e.g. a failed persist
+/// read or a violated internal invariant) have no structured representation;
+/// they carry an opaque message and are reported as `INTERNAL_ERROR`.
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
+pub enum PeekError {
+    /// A structured error produced while evaluating the dataflow.
+    Dataflow(Box<DataflowError>),
+    /// An internal error from the peek machinery, with no structured form.
+    Internal(String),
+}
+
+impl PeekError {
+    /// Constructs an [`PeekError::Internal`] from anything string-like.
+    pub fn internal(message: impl Into<String>) -> Self {
+        PeekError::Internal(message.into())
+    }
+}
+
+impl fmt::Display for PeekError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            PeekError::Dataflow(err) => err.fmt(f),
+            PeekError::Internal(err) => f.write_str(err),
+        }
+    }
+}
+
+impl From<DataflowError> for PeekError {
+    fn from(err: DataflowError) -> Self {
+        PeekError::Dataflow(Box::new(err))
+    }
+}
+
+impl From<EvalError> for PeekError {
+    fn from(err: EvalError) -> Self {
+        PeekError::Dataflow(Box::new(err.into()))
+    }
+}
+
 /// Response from a peek whose results have been stashed into persist.
 #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
 pub struct StashedPeekResponse {

src/compute-client/src/service.rs

@@ -27,8 +27,8 @@ use uuid::Uuid;
 
 use crate::protocol::command::ComputeCommand;
 use crate::protocol::response::{
-    ComputeResponse, CopyToResponse, FrontiersResponse, PeekResponse, StashedPeekResponse,
-    SubscribeBatch, SubscribeResponse,
+    ComputeResponse, CopyToResponse, FrontiersResponse, PeekError, PeekResponse,
+    StashedPeekResponse, SubscribeBatch, SubscribeResponse,
 };
 
 /// A client to a compute server.
@@ -588,7 +588,7 @@ fn merge_peek_responses(
             "total result exceeds max size of {}",
             ByteSize::b(max_result_size)
         );
-        return Error(err);
+        return Error(PeekError::internal(err));
     }
 
     match (resp1, resp2) {
@@ -625,14 +625,14 @@ fn merge_peek_responses(
                     "shard IDs of stashed responses do not match: \
                              {shard_id1} != {shard_id2}"
                 );
-                return Error("internal error".into());
+                return Error(PeekError::internal("internal error"));
             }
             if relation_desc1 != relation_desc2 {
                 soft_panic_or_log!(
                     "relation descs of stashed responses do not match: \
                              {relation_desc1:?} != {relation_desc2:?}"
                 );
-                return Error("internal error".into());
+                return Error(PeekError::internal("internal error"));
             }
 
             batches1.append(&mut batches2);