From c62500439981dda87b8bcba212dac24b474ad6a7 Mon Sep 17 00:00:00 2001
From: EnRaiha <15997552+EnRaiha@users.noreply.github.com>
Date: Fri, 18 Sep 2026 23:29:31 +0800
Subject: [PATCH] fix(pgwire): classify the streamed and DDL shaping failures

The streamed path rendered decode and shape failures as XX000 with a raw
message, and the DDL dispatch did the same over a RETURNING shape failure, so
a class the error already carried reached the client as an internal fault.
Both now render through the classified path (error_to_sqlstate for the
stream's crate::Error, numeric_code_to_sqlstate for the shaper's NodeDbError)
with shaping_error_message hygiene. The two 'payload was not a row array'
invariants keep XX000 by decision.
---
 .../server/pgwire/handler/stream_response.rs  | 29 ++++++++++++-------
 .../neutral/collection/dml/parse/dispatch.rs  | 11 +++++--
 2 files changed, 28 insertions(+), 12 deletions(-)

diff --git a/nodedb/src/control/server/pgwire/handler/stream_response.rs b/nodedb/src/control/server/pgwire/handler/stream_response.rs
index a63f7c5ff..761186326 100644
--- a/nodedb/src/control/server/pgwire/handler/stream_response.rs
+++ b/nodedb/src/control/server/pgwire/handler/stream_response.rs
@@ -22,7 +22,7 @@ use crate::control::state::SharedState;
 use crate::data::executor::response_codec::{decode_payload_to_json, decode_payload_value};
 
 use super::super::ddl_encode::col_type_to_field_with_format;
-use super::super::types::{error_to_sqlstate, text_field};
+use super::super::types::{error_to_sqlstate, shaping_error_message, text_field};
 use super::shape_encode::{encode_shaped_row, shaped_query_response};
 
 /// The per-request plumbing a lazily-streamed pgwire response owns for its
@@ -215,10 +215,13 @@ pub(crate) fn streaming_shaped_response(
             }
 
             let value = decode_payload_value(&batch.payload).map_err(|e| {
+                let (severity, code, message) = error_to_sqlstate(&e);
+                let message =
+                    shaping_error_message(crate::error_classify::classify(&e).code(), message);
                 PgWireError::UserError(Box::new(ErrorInfo::new(
-                    "ERROR".to_owned(),
-                    "XX000".to_owned(),
-                    format!("failed to decode streamed batch: {e}"),
+                    severity.to_owned(),
+                    code.to_owned(),
+                    format!("failed to decode streamed batch: {message}"),
                 )))
             })?;
             // Resolved once before the first batch was pulled; this only
@@ -233,10 +236,13 @@ pub(crate) fn streaming_shaped_response(
                 None,
             )
             .map_err(|e| {
+                let (severity, code, message) = error_to_sqlstate(&e);
+                let message =
+                    shaping_error_message(crate::error_classify::classify(&e).code(), message);
                 PgWireError::UserError(Box::new(ErrorInfo::new(
-                    "ERROR".to_owned(),
-                    "XX000".to_owned(),
-                    format!("failed to shape streamed batch: {e}"),
+                    severity.to_owned(),
+                    code.to_owned(),
+                    format!("failed to shape streamed batch: {message}"),
                 )))
             })?;
             for row in &shaped.rows {
@@ -349,10 +355,13 @@ pub(crate) async fn streaming_star_response(
     ) {
         Ok(s) => s,
         Err(e) => {
+            let (severity, code, message) = error_to_sqlstate(&e);
+            let message =
+                shaping_error_message(crate::error_classify::classify(&e).code(), message);
             return single_pgwire_error(PgWireError::UserError(Box::new(ErrorInfo::new(
-                "ERROR".to_owned(),
-                "XX000".to_owned(),
-                format!("failed to shape streamed batch: {e}"),
+                severity.to_owned(),
+                code.to_owned(),
+                format!("failed to shape streamed batch: {message}"),
             ))));
         }
     };
diff --git a/nodedb/src/control/server/shared/ddl/neutral/collection/dml/parse/dispatch.rs b/nodedb/src/control/server/shared/ddl/neutral/collection/dml/parse/dispatch.rs
index 0e87463ab..554cb6b9b 100644
--- a/nodedb/src/control/server/shared/ddl/neutral/collection/dml/parse/dispatch.rs
+++ b/nodedb/src/control/server/shared/ddl/neutral/collection/dml/parse/dispatch.rs
@@ -7,7 +7,9 @@ use crate::control::security::audit::ArcAuditEmitter;
 use crate::control::security::identity::{AuthenticatedIdentity, Permission};
 use crate::control::security::request_scope::RequestAuthScope;
 use crate::control::sequence::SessionSequenceAccess;
-use crate::control::server::pgwire::types::error_to_sqlstate;
+use crate::control::server::pgwire::types::{
+    error_to_sqlstate, numeric_code_to_sqlstate, shaping_error_message,
+};
 use crate::control::server::response_shape::compose::{ShapeOutcome, shape_response_materialized};
 use crate::control::server::response_shape::redaction::QueryRedaction;
 use crate::control::server::response_shape::request::MaterializedShapeRequest;
@@ -443,7 +445,12 @@ pub(in crate::control::server::shared::ddl::neutral::collection) async fn plan_a
                 redaction: Some(redaction.ctx(&state.redaction)),
                 sequences: Some(&sequences),
             })
-            .map_err(|error| ddl_err("XX000", error.message().to_string()))?;
+            .map_err(|error| {
+                ddl_err(
+                    numeric_code_to_sqlstate(error.code()),
+                    shaping_error_message(error.code(), error.message()),
+                )
+            })?;
             // Folded rather than pushed: a statement is ONE result set, however
             // many tasks it planned to.
             if let ShapeOutcome::Rows(shaped) = outcome {