diff --git a/nodedb/src/control/server/pgwire/handler/routing/cluster_array.rs b/nodedb/src/control/server/pgwire/handler/routing/cluster_array.rs index 0024ffc96..8982383de 100644 --- a/nodedb/src/control/server/pgwire/handler/routing/cluster_array.rs +++ b/nodedb/src/control/server/pgwire/handler/routing/cluster_array.rs @@ -19,7 +19,7 @@ use crate::control::server::response_shape::redaction::QueryRedaction; use crate::control::server::response_shape::schema::OutputSchema; use crate::control::server::shared::session::SessionId; -use super::super::super::types::{error_to_sqlstate, sqlstate_error}; +use super::super::super::types::{error_to_sqlstate, numeric_code_to_sqlstate, sqlstate_error}; use super::super::core::NodeDbPgHandler; use super::super::plan::{PlanKind, payload_to_response}; use super::super::shape_encode; @@ -128,7 +128,7 @@ impl NodeDbPgHandler { projection, Some(redaction.ctx(&self.state.redaction)), ) - .map_err(|e| sqlstate_error("XX000", e.message()))? + .map_err(|e| sqlstate_error(numeric_code_to_sqlstate(e.code()), e.message()))? { ShapeOutcome::Rows(shaped) => { let (response, notice) = diff --git a/nodedb/src/control/server/pgwire/handler/routing/dispatch_loop.rs b/nodedb/src/control/server/pgwire/handler/routing/dispatch_loop.rs index 01baa406c..7c847600e 100644 --- a/nodedb/src/control/server/pgwire/handler/routing/dispatch_loop.rs +++ b/nodedb/src/control/server/pgwire/handler/routing/dispatch_loop.rs @@ -23,7 +23,9 @@ use crate::control::server::shared::session::SessionId; use crate::types::TenantId; use nodedb_physical::physical_task::{PhysicalTask, PostSetOp}; -use super::super::super::types::{error_to_sqlstate, response_status_to_sqlstate, sqlstate_error}; +use super::super::super::types::{ + error_to_sqlstate, numeric_code_to_sqlstate, response_status_to_sqlstate, sqlstate_error, +}; use super::super::core::NodeDbPgHandler; use super::super::plan::{PlanKind, describe_plan, payload_to_response}; use super::super::shape_encode; @@ -402,7 +404,10 @@ impl NodeDbPgHandler { tenant_id, redaction: Some(redaction.ctx(&self.state.redaction)), }) - .map_err(|e| sqlstate_error("XX000", e.message()))? + // Preserve the shaper error's own SQLSTATE: hardcoding XX000 + // turned a client-visible refusal such as an undefined + // sequence into an internal fault. + .map_err(|e| sqlstate_error(numeric_code_to_sqlstate(e.code()), e.message()))? { ShapeOutcome::Rows(shaped) => { task_rows = Some(shaped.rows.len() as u64); diff --git a/nodedb/src/control/server/pgwire/handler/routing/execute_dml_hooks.rs b/nodedb/src/control/server/pgwire/handler/routing/execute_dml_hooks.rs index 41cbb3a3e..1711cc74d 100644 --- a/nodedb/src/control/server/pgwire/handler/routing/execute_dml_hooks.rs +++ b/nodedb/src/control/server/pgwire/handler/routing/execute_dml_hooks.rs @@ -21,7 +21,7 @@ use crate::control::trigger::dml_hook::DmlWriteInfo; use crate::types::TenantId; use nodedb_physical::physical_task::PhysicalTask; -use super::super::super::types::{error_to_sqlstate, sqlstate_error}; +use super::super::super::types::{error_to_sqlstate, numeric_code_to_sqlstate, sqlstate_error}; use super::super::core::NodeDbPgHandler; use super::super::plan::PlanKind; @@ -332,7 +332,7 @@ impl NodeDbPgHandler { projection, Some(redaction.ctx(&self.state.redaction)), ) - .map_err(|e| sqlstate_error("XX000", e.message()))? + .map_err(|e| sqlstate_error(numeric_code_to_sqlstate(e.code()), e.message()))? { ShapeOutcome::Rows(shaped) => { // Clone write-path DML result (PointUpdate/PointDelete): diff --git a/nodedb/src/control/server/pgwire/handler/routing/gateway_dispatch.rs b/nodedb/src/control/server/pgwire/handler/routing/gateway_dispatch.rs index b7bdd9960..cba3f624a 100644 --- a/nodedb/src/control/server/pgwire/handler/routing/gateway_dispatch.rs +++ b/nodedb/src/control/server/pgwire/handler/routing/gateway_dispatch.rs @@ -22,7 +22,7 @@ use crate::control::server::shared::metering::{PlanMeteringInfo, meter_dispatch} use crate::types::{TenantId, TraceId}; use nodedb_physical::physical_task::PhysicalTask; -use super::super::super::types::sqlstate_error; +use super::super::super::types::{numeric_code_to_sqlstate, sqlstate_error}; use super::super::core::NodeDbPgHandler; use super::super::plan::{PlanKind, multirow_payload_to_response}; use super::super::shape_encode; @@ -75,7 +75,7 @@ fn push_shaped_response( projection, Some(redaction.ctx(&state.redaction)), ) - .map_err(|e| sqlstate_error("XX000", e.message()))? + .map_err(|e| sqlstate_error(numeric_code_to_sqlstate(e.code()), e.message()))? { ShapeOutcome::Rows(shaped) => { let (response, notice) = shape_encode::shaped_query_response(shaped, result_formats); @@ -228,7 +228,7 @@ impl NodeDbPgHandler { projection, Some(redaction.ctx(&self.state.redaction)), ) - .map_err(|e| sqlstate_error("XX000", e.message()))? + .map_err(|e| sqlstate_error(numeric_code_to_sqlstate(e.code()), e.message()))? { ShapeOutcome::Rows(shaped) => { task_rows = Some(task_rows.unwrap_or(0) + shaped.rows.len() as u64); diff --git a/nodedb/src/control/server/pgwire/handler/routing/set_ops.rs b/nodedb/src/control/server/pgwire/handler/routing/set_ops.rs index 2baaaa9a3..736caabca 100644 --- a/nodedb/src/control/server/pgwire/handler/routing/set_ops.rs +++ b/nodedb/src/control/server/pgwire/handler/routing/set_ops.rs @@ -13,7 +13,7 @@ use crate::control::server::response_shape::compose::{self, ShapeOutcome}; use crate::control::server::response_shape::redaction::RedactionCtx; use crate::control::server::response_shape::schema::OutputSchema; -use super::super::super::types::sqlstate_error; +use super::super::super::types::{numeric_code_to_sqlstate, sqlstate_error}; use super::super::plan::{PlanKind, multirow_payload_to_response}; use super::super::shape_encode; @@ -37,7 +37,7 @@ pub(super) fn apply_set_ops( }; Ok( match compose::shape_payload_no_plan(&merged, PlanKind::MultiRow, projection, redaction) - .map_err(|e| sqlstate_error("XX000", e.message()))? + .map_err(|e| sqlstate_error(numeric_code_to_sqlstate(e.code()), e.message()))? { ShapeOutcome::Rows(shaped) => { shape_encode::shaped_query_response(shaped, result_formats) diff --git a/nodedb/src/control/server/pgwire/handler/routing/streaming.rs b/nodedb/src/control/server/pgwire/handler/routing/streaming.rs index 01cfdc0dc..8a0e8aa19 100644 --- a/nodedb/src/control/server/pgwire/handler/routing/streaming.rs +++ b/nodedb/src/control/server/pgwire/handler/routing/streaming.rs @@ -22,7 +22,7 @@ use crate::control::server::response_shape::redaction::QueryRedaction; use crate::control::server::shared::session::SessionId; use super::super::super::types::error_to_sqlstate; -use super::super::super::types::sqlstate_error; +use super::super::super::types::{numeric_code_to_sqlstate, sqlstate_error}; use super::super::core::NodeDbPgHandler; use super::super::plan::{PlanKind, multirow_payload_to_response}; use super::super::stream_response; @@ -141,8 +141,9 @@ impl NodeDbPgHandler { shaping.projection, Some(redaction.ctx(&state.redaction)), ) - .map_err(|e| sqlstate_error("XX000", e.message()))? - { + .map_err(|e| { + sqlstate_error(numeric_code_to_sqlstate(e.code()), e.message()) + })? { ShapeOutcome::Rows(shaped) => { let (response, _notice) = crate::control::server::pgwire::handler::shape_encode::shaped_query_response( shaped, diff --git a/nodedb/src/control/server/pgwire/types/error_map.rs b/nodedb/src/control/server/pgwire/types/error_map.rs index 976634640..48ced622c 100644 --- a/nodedb/src/control/server/pgwire/types/error_map.rs +++ b/nodedb/src/control/server/pgwire/types/error_map.rs @@ -197,7 +197,7 @@ pub fn error_to_sqlstate(err: &crate::Error) -> (&'static str, &'static str, Str /// numeric code, so a constraint violation (say) maps to the same SQLSTATE /// whether it happened locally or on a remote node. Unmapped/unknown codes /// fall back to INTERNAL_ERROR — the behaviour before codes were preserved. -pub(crate) fn numeric_code_to_sqlstate(code: nodedb_types::error::ErrorCode) -> &'static str { +pub fn numeric_code_to_sqlstate(code: nodedb_types::error::ErrorCode) -> &'static str { use nodedb_types::error::ErrorCode as Ec; match code { // Mirrors the `RejectedConstraint` arm. @@ -213,6 +213,27 @@ pub(crate) fn numeric_code_to_sqlstate(code: nodedb_types::error::ErrorCode) -> Ec::DOCUMENT_NOT_FOUND => sqlstate::NO_DATA, // Mirrors the `BadRequest` / `PlanError` arms. Ec::BAD_REQUEST | Ec::PLAN_ERROR => sqlstate::SYNTAX_ERROR, + // A named catalog object (a sequence, most commonly) that does not + // exist. Without this arm the code falls back to INTERNAL_ERROR + // (XX000), so a client-visible "undefined object" reaches pgwire as an + // internal fault — the same error-class leak the response-shaping path + // had. + Ec::UNDEFINED_OBJECT => sqlstate::UNDEFINED_OBJECT, + // The object exists but its prerequisite step has not run, such as + // `currval` before this session called `nextval`. Mirrors the + // `ObjectNotInPrerequisiteState` arm. + Ec::OBJECT_NOT_READY => sqlstate::OBJECT_NOT_IN_PREREQUISITE_STATE, + // A generic lookup miss. Mirrors `crate::Error::DocumentNotFound`'s + // no-data class for the callers that use the generic code. + Ec::NOT_FOUND => sqlstate::NO_DATA, + // The named database does not exist. + Ec::DATABASE_NOT_FOUND => sqlstate::INVALID_CATALOG_NAME, + // SQL is disabled for this server; not a retryable internal fault. + Ec::SQL_NOT_ENABLED => sqlstate::FEATURE_NOT_SUPPORTED, + // A value could not be coerced to the target type. + Ec::TYPE_MISMATCH => sqlstate::CANNOT_COERCE, + // A value outside the target type's range. + Ec::OVERFLOW => sqlstate::NUMERIC_VALUE_OUT_OF_RANGE, // Mirrors the `UndefinedFunction` arm. Ec::UNDEFINED_FUNCTION => sqlstate::UNDEFINED_FUNCTION, // Mirrors the `UndefinedColumn` arm. @@ -266,3 +287,65 @@ pub fn response_status_to_sqlstate( } } } + +#[cfg(test)] +mod tests { + use super::*; + use nodedb_types::error::{ErrorCode as Ec, NodeDbError}; + + /// Every routed shaper call site (`cluster_array`, `dispatch_loop`, + /// `execute_dml_hooks`, `gateway_dispatch`, `set_ops`, `streaming`) renders a + /// shaper error as `sqlstate_error(numeric_code_to_sqlstate(e.code()), ..)`. + /// The rendered code is what a client matches on, so a code that already + /// classifies the fault must reach it as that class. + fn routed_code(err: &NodeDbError) -> String { + let rendered = sqlstate_error(numeric_code_to_sqlstate(err.code()), err.message()); + match rendered { + PgWireError::UserError(info) => info.code, + other => panic!("sqlstate_error must build a UserError, got {other:?}"), + } + } + + /// An undefined named object — the sequence case the routed sites exist to + /// answer — keeps its own class (42704) instead of the XX000 fallback. A + /// client treats XX000 as an internal fault and does not retry; it treats + /// 42704 as a missing object it caused. + #[test] + fn routed_shaper_error_keeps_its_numeric_class() { + let err = NodeDbError::undefined_object("seq_routed_probe"); + assert_eq!(routed_code(&err), sqlstate::UNDEFINED_OBJECT); + assert_eq!(routed_code(&err), "42704"); + assert_ne!(routed_code(&err), sqlstate::INTERNAL_ERROR); + } + + /// The read-path codes the routed sites can surface each map to their + /// Postgres class rather than collapsing into the XX000 fallback. An + /// unmapped code is exactly the leak this pins shut. + #[test] + fn read_path_numeric_codes_keep_their_class() { + let cases = [ + (Ec::UNDEFINED_OBJECT, sqlstate::UNDEFINED_OBJECT), + ( + Ec::OBJECT_NOT_READY, + sqlstate::OBJECT_NOT_IN_PREREQUISITE_STATE, + ), + (Ec::NOT_FOUND, sqlstate::NO_DATA), + (Ec::DATABASE_NOT_FOUND, sqlstate::INVALID_CATALOG_NAME), + (Ec::SQL_NOT_ENABLED, sqlstate::FEATURE_NOT_SUPPORTED), + (Ec::TYPE_MISMATCH, sqlstate::CANNOT_COERCE), + (Ec::OVERFLOW, sqlstate::NUMERIC_VALUE_OUT_OF_RANGE), + ]; + for (numeric, expected) in cases { + let mapped = numeric_code_to_sqlstate(numeric); + assert_eq!( + mapped, expected, + "numeric code {numeric} must keep class {expected}" + ); + assert_ne!( + mapped, + sqlstate::INTERNAL_ERROR, + "numeric code {numeric} must not fall back to XX000" + ); + } + } +} diff --git a/nodedb/src/control/server/pgwire/types/mod.rs b/nodedb/src/control/server/pgwire/types/mod.rs index e4796724b..98c839ac9 100644 --- a/nodedb/src/control/server/pgwire/types/mod.rs +++ b/nodedb/src/control/server/pgwire/types/mod.rs @@ -11,7 +11,8 @@ pub mod parse; pub mod privilege; pub use error_map::{ - error_to_sqlstate, notice_warning, response_status_to_sqlstate, sqlstate_error, + error_to_sqlstate, notice_warning, numeric_code_to_sqlstate, response_status_to_sqlstate, + sqlstate_error, }; pub use field::{ bool_field, bytea_field, float4_array_field, float4_field, float8_array_field, float8_field, diff --git a/nodedb/src/control/server/response_shape/types.rs b/nodedb/src/control/server/response_shape/types.rs index b70d97615..bdf59a5e2 100644 --- a/nodedb/src/control/server/response_shape/types.rs +++ b/nodedb/src/control/server/response_shape/types.rs @@ -147,7 +147,10 @@ pub fn describe_plan(plan: &PhysicalPlan) -> PlanKind { PhysicalPlan::Document(DocumentOp::PointPut { .. }) | PhysicalPlan::Document(DocumentOp::PointInsert { .. }) | PhysicalPlan::Document(DocumentOp::BatchInsert { .. }) + | PhysicalPlan::Kv(KvOp::Insert { .. }) | PhysicalPlan::Kv(KvOp::InsertIfAbsent { .. }) + | PhysicalPlan::Kv(KvOp::Put { .. }) + | PhysicalPlan::Kv(KvOp::BatchPut { .. }) | PhysicalPlan::Columnar(ColumnarOp::Insert { .. }) => DmlResult("INSERT"), PhysicalPlan::Document(DocumentOp::PointUpdate { diff --git a/nodedb/src/data/executor/handlers/kv/crud/write_basic.rs b/nodedb/src/data/executor/handlers/kv/crud/write_basic.rs index 39594631c..292c42046 100644 --- a/nodedb/src/data/executor/handlers/kv/crud/write_basic.rs +++ b/nodedb/src/data/executor/handlers/kv/crud/write_basic.rs @@ -80,7 +80,10 @@ impl CoreLoop { // stored post-image, not an echo of the request. return self.kv_stored_returning_response(task, spec, rls_filters, &[(key, value)]); } - self.response_ok(task) + // A put always writes its row, so the statement affected exactly one: + // the tag is `INSERT 0 1`, never a bare `OK` (pgwire's generic tag for + // a plan that reports nothing). + self.response_affected(task, 1) } /// SQL `INSERT` semantics: write only if key doesn't already exist. @@ -167,7 +170,10 @@ impl CoreLoop { if let Some(spec) = returning { return self.kv_stored_returning_response(task, spec, rls_filters, &[(key, value)]); } - self.response_ok(task) + // An insert writes exactly one row or fails the statement, so the + // affected count is 1 and pgwire renders `INSERT 0 1` — the same tag + // the document engine's point insert produces. + self.response_affected(task, 1) } /// SQL `INSERT ... ON CONFLICT DO NOTHING` semantics: write if absent, diff --git a/nodedb/tests/wire/cases/sql_dml_affected_counts.rs b/nodedb/tests/wire/cases/sql_dml_affected_counts.rs index cc76c3fbc..5c477fdae 100644 --- a/nodedb/tests/wire/cases/sql_dml_affected_counts.rs +++ b/nodedb/tests/wire/cases/sql_dml_affected_counts.rs @@ -286,6 +286,52 @@ async fn kv_multi_key_delete_reports_matched_key_count() { ); } +/// A key-value `INSERT` of a new key writes exactly one row, so its command tag +/// must report `1`. A kv write that answers with no affected count renders as a +/// bare `OK` tag, which the client's tag parser reads as `0` rows — the count is +/// the only signal distinguishing "wrote" from "matched nothing". +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn kv_insert_of_new_key_reports_one() { + let server = TestServer::start().await; + server + .exec("CREATE COLLECTION kv_probe (key TEXT PRIMARY KEY, n INT) WITH (engine='kv')") + .await + .unwrap(); + + let count = affected(&server, "INSERT INTO kv_probe (key, n) VALUES ('a', 1)").await; + assert_eq!( + count, 1, + "a KV INSERT that wrote one key must report 1, not a bare tag the client reads as 0" + ); + assert_eq!( + live_rows(&server, "SELECT count(*) FROM kv_probe WHERE key = 'a'").await, + 1, + "the key must really be present, so the reported count is the honest one" + ); +} + +/// A key-value `UPSERT` overwrites or writes exactly one key, so its tag must +/// report `1` for the same reason: the PUT path always writes the key. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn kv_upsert_of_new_key_reports_one() { + let server = TestServer::start().await; + server + .exec("CREATE COLLECTION kv_probe (key TEXT PRIMARY KEY, n INT) WITH (engine='kv')") + .await + .unwrap(); + + let count = affected(&server, "UPSERT INTO kv_probe (key, n) VALUES ('a', 1)").await; + assert_eq!( + count, 1, + "a KV UPSERT that wrote one key must report 1, not a bare tag the client reads as 0" + ); + assert_eq!( + live_rows(&server, "SELECT count(*) FROM kv_probe WHERE key = 'a'").await, + 1, + "the upserted key must really be present" + ); +} + /// A CRDT document collection routes its PK-targeted delete through the CRDT /// engine, which shares the count contract: a delete that removed the row /// reports `1`.