feat(query): jump to PostgreSQL error position from result messages

crates/dbx-core/src/backend_error.rs

@@ -125,6 +125,11 @@ pub struct BackendError {
     diagnostics: Option<BackendErrorDiagnostics>,
     #[serde(skip_serializing_if = "Option::is_none")]
     help_url: Option<String>,
+    /// Driver-reported SQL error position, when the adapter provides one
+    /// (currently native PostgreSQL). Optional field, added without bumping the
+    /// envelope version; clients that do not understand it simply ignore it.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    error_position: Option<crate::sql_error_position::SqlErrorPosition>,
 }
 
 impl BackendError {
@@ -270,6 +275,14 @@ impl BackendError {
         )
     }
 
+    /// Create a SQL failure envelope while retaining bounded native driver detail
+    /// and a driver-reported error position.
+    pub fn from_sql_detail_with_position(message: &str, position: crate::sql_error_position::SqlErrorPosition) -> Self {
+        let mut error = Self::from_sql_detail(message);
+        error.error_position = Some(position);
+        error
+    }
+
     /// Adapt a DuckDB worker error while retaining both the native detail and
     /// the worker protocol code for diagnostics at the public boundary.
     pub fn from_duckdb_worker_error(code: &str, message: &str) -> Self {
@@ -350,6 +363,10 @@ impl BackendError {
         self.diagnostics.as_ref()
     }
 
+    pub fn error_position(&self) -> Option<crate::sql_error_position::SqlErrorPosition> {
+        self.error_position
+    }
+
     fn new(
         entry: &'static CatalogEntry,
         source: BackendErrorSource,
@@ -372,6 +389,7 @@ impl BackendError {
             detail,
             diagnostics,
             help_url: entry.help_url.map(str::to_string),
+            error_position: None,
         }
     }
 }
@@ -673,6 +691,8 @@ fn bounded_text(value: &str, max_bytes: usize) -> String {
 }
 
 fn bounded_detail(message: &str) -> Option<String> {
+    let stripped = crate::sql_error_position::strip_marker(message);
+    let message = stripped.as_str();
     if message.trim().is_empty() {
         return None;
     }
@@ -684,7 +704,10 @@ fn bounded_detail(message: &str) -> Option<String> {
 fn bounded_native_detail(message: &str) -> Option<String> {
     // This path is only for typed native SQL failures. Do not infer or rewrite
     // SQL content here; preserving the driver's diagnostic is the contract.
-    let trimmed = message.trim();
+    // The position transport suffix is stripped because it is surfaced through
+    // the structured `errorPosition` field, not as message text.
+    let stripped = crate::sql_error_position::strip_marker(message);
+    let trimmed = stripped.trim();
     if trimmed.is_empty() {
         return None;
     }
@@ -1141,6 +1164,34 @@ mod tests {
         assert_eq!(payload["origin"]["adapter"], "native");
     }
 
+    #[test]
+    fn sql_error_with_position_exposes_typed_error_position_and_keeps_v1_detail() {
+        let position = crate::sql_error_position::SqlErrorPosition { line: 2, column: 6, offset: 12 };
+        let payload = serde_json::to_value(BackendError::from_sql_detail_with_position(
+            "ERROR: relation \"missing\" does not exist",
+            position,
+        ))
+        .unwrap();
+
+        assert_eq!(payload["code"], "DBX-JDBC-4001");
+        assert_eq!(payload["detail"], "ERROR: relation \"missing\" does not exist");
+        assert_eq!(payload["errorPosition"]["line"], 2);
+        assert_eq!(payload["errorPosition"]["column"], 6);
+        assert_eq!(payload["errorPosition"]["offset"], 12);
+        assert_eq!(payload["origin"]["adapter"], "native");
+
+        // SQL errors without a position must not emit the optional field.
+        let plain = serde_json::to_value(BackendError::from_sql_detail("ERROR: nope")).unwrap();
+        assert!(plain.get("errorPosition").is_none());
+    }
+
+    #[test]
+    fn detail_sanitizers_strip_the_position_transport_marker() {
+        let message = format!("ERROR: relation does not exist{}", crate::sql_error_position::encode_marker(15));
+        assert_eq!(bounded_native_detail(&message).as_deref(), Some("ERROR: relation does not exist"));
+        assert_eq!(bounded_detail(&message).as_deref(), Some("ERROR: relation does not exist"));
+    }
+
     #[test]
     fn duckdb_worker_sql_error_preserves_worker_code_detail_and_driver_origin() {
         let payload = serde_json::to_value(BackendError::from_duckdb_worker_error(

crates/dbx-core/src/db/postgres.rs

@@ -1480,6 +1480,25 @@ fn escape_tsvector_lexeme(value: &str) -> String {
 }
 
 fn pg_error_to_string(err: tokio_postgres::Error) -> String {
+    let Some(db_error) = err.as_db_error() else {
+        return err.to_string();
+    };
+    let mut message = db_error.to_string();
+    // Carry the server-reported cursor position across the `db` layer's
+    // `Result<_, String>` boundary; `query.rs` resolves it against the executed
+    // statement and strips the suffix before the message reaches any client.
+    if let Some(tokio_postgres::error::ErrorPosition::Original(cursor)) = db_error.position() {
+        message.push_str(&crate::sql_error_position::encode_marker(*cursor));
+    }
+    message
+}
+
+/// Same as [`pg_error_to_string`] but never carries a cursor position.
+///
+/// Used for infrastructure/setup statements (search_path, BEGIN/ROLLBACK, …)
+/// whose SQL is not the statement the user is editing: a marker from those would
+/// be resolved against the user's SQL and point at the wrong place.
+fn pg_error_to_string_plain(err: tokio_postgres::Error) -> String {
     err.as_db_error().map(ToString::to_string).unwrap_or_else(|| err.to_string())
 }
 
@@ -2822,7 +2841,10 @@ async fn set_automatic_postgres_timezone(client: &deadpool_postgres::Client, tim
                 }
             }
             Err(error) => {
-                return Err(format!("PostgreSQL SET timezone failed after connecting: {}", pg_error_to_string(error)));
+                return Err(format!(
+                    "PostgreSQL SET timezone failed after connecting: {}",
+                    pg_error_to_string_plain(error)
+                ));
             }
         }
     }
@@ -8025,7 +8047,7 @@ pub(crate) async fn execute_postgres_infra_statement(
     tokio::time::timeout(timeout_duration, client.execute_typed(sql, &[]))
         .await
         .map_err(|_| format!("PostgreSQL {stage} timed out after {} seconds", timeout_duration.as_secs()))?
-        .map_err(pg_error_to_string)
+        .map_err(pg_error_to_string_plain)
 }
 
 pub(crate) async fn wait_postgres_operation<T, F>(

crates/dbx-core/src/lib.rs

@@ -93,6 +93,7 @@ pub mod sql_analysis;
 pub mod sql_diagnostics;
 pub mod sql_dialect;
 pub mod sql_editability;
+pub mod sql_error_position;
 pub mod sql_file_import;
 pub mod sql_parser;
 pub mod sql_risk;

crates/dbx-core/src/query.rs

@@ -68,23 +68,43 @@ pub enum PoolErrorAction {
 #[derive(Debug, Clone)]
 pub enum QueryExecutionError {
     Agent(AgentCallError),
-    DuckDb { code: String, message: String },
-    Canceled { stage: AgentErrorStage, operation_outcome: AgentOperationOutcome },
+    DuckDb {
+        code: String,
+        message: String,
+    },
+    Canceled {
+        stage: AgentErrorStage,
+        operation_outcome: AgentOperationOutcome,
+    },
     Timeout(String),
     Sql(String),
+    /// Native PostgreSQL SQL failure whose driver-reported cursor position was
+    /// resolved against the executed statement text. Kept distinct from
+    /// [`Self::Sql`] so the position survives `classify_query_error` and reaches
+    /// [`Self::into_backend_error`] as a typed field instead of being parsed back
+    /// out of a string.
+    SqlWithPosition {
+        message: String,
+        position: crate::sql_error_position::SqlErrorPosition,
+    },
     Legacy(String),
 }
 
 impl QueryExecutionError {
     pub fn into_legacy_string(self) -> String {
-        match self {
+        let mut message = match self {
             Self::Agent(error) => error.into_legacy_string(),
             Self::DuckDb { message, .. } => message,
             Self::Canceled { .. } => canceled_error(),
             Self::Timeout(error) => error,
             Self::Sql(error) => error,
+            Self::SqlWithPosition { message, .. } => message,
             Self::Legacy(error) => error,
-        }
+        };
+        // Defensive: any remaining transport marker (e.g. from a driver error
+        // that never passed through the resolve step) must not reach clients.
+        while crate::sql_error_position::take_marker(&mut message).is_some() {}
+        message
     }
 
     pub fn into_backend_error(self) -> crate::backend_error::BackendError {
@@ -98,6 +118,9 @@ impl QueryExecutionError {
             }
             Self::Timeout(error) => crate::backend_error::BackendError::from_timeout_detail(&error),
             Self::Sql(error) => crate::backend_error::BackendError::from_sql_detail(&error),
+            Self::SqlWithPosition { message, position } => {
+                crate::backend_error::BackendError::from_sql_detail_with_position(&message, position)
+            }
             Self::Legacy(error) => crate::backend_error::BackendError::from_legacy_string(&error),
         }
     }
@@ -111,6 +134,9 @@ impl QueryExecutionError {
             canceled @ Self::Canceled { .. } => canceled,
             Self::Timeout(error) => Self::Timeout(query_error_with_omitted_sql_context(&error, sql)),
             Self::Sql(error) => Self::Sql(append_typed_sql_error_context(&error, sql)),
+            Self::SqlWithPosition { message, position } => {
+                Self::SqlWithPosition { message: append_typed_sql_error_context(&message, sql), position }
+            }
             Self::Legacy(error) => Self::Legacy(query_error_with_omitted_sql_context(&error, sql)),
         }
     }
@@ -122,6 +148,9 @@ impl QueryExecutionError {
             canceled @ Self::Canceled { .. } => canceled,
             Self::Timeout(error) => Self::Timeout(format!("{error}; {context}")),
             Self::Sql(error) => Self::Sql(format!("{error}; {context}")),
+            Self::SqlWithPosition { message, position } => {
+                Self::SqlWithPosition { message: format!("{message}; {context}"), position }
+            }
             Self::Legacy(error) => Self::Legacy(format!("{error}; {context}")),
         }
     }
@@ -129,7 +158,12 @@ impl QueryExecutionError {
     fn as_agent_error(&self) -> Option<&AgentCallError> {
         match self {
             Self::Agent(error) => Some(error),
-            Self::DuckDb { .. } | Self::Canceled { .. } | Self::Timeout(_) | Self::Sql(_) | Self::Legacy(_) => None,
+            Self::DuckDb { .. }
+            | Self::Canceled { .. }
+            | Self::Timeout(_)
+            | Self::Sql(_)
+            | Self::SqlWithPosition { .. }
+            | Self::Legacy(_) => None,
         }
     }
 }
@@ -142,6 +176,7 @@ impl std::fmt::Display for QueryExecutionError {
             Self::Canceled { .. } => formatter.write_str(QUERY_CANCELED),
             Self::Timeout(error) => formatter.write_str(error),
             Self::Sql(error) => formatter.write_str(error),
+            Self::SqlWithPosition { message, .. } => formatter.write_str(message),
             Self::Legacy(error) => formatter.write_str(error),
         }
     }
@@ -1490,6 +1525,7 @@ fn query_execution_error_action(
         QueryExecutionError::DuckDb { message, .. } => query_pool_error_action(db_type, sql, message),
         QueryExecutionError::Timeout(message)
         | QueryExecutionError::Sql(message)
+        | QueryExecutionError::SqlWithPosition { message, .. }
         | QueryExecutionError::Legacy(message) => query_pool_error_action(db_type, sql, message),
         QueryExecutionError::Agent(_) => unreachable!("Agent errors return above"),
     }
@@ -2345,7 +2381,23 @@ async fn do_execute_typed(
             if let Some(duckdb_error) = typed_duckdb_error {
                 return QueryExecutionError::DuckDb { code: duckdb_error.code, message: duckdb_error.message };
             }
-            typed_agent_error.map_or_else(|| QueryExecutionError::Legacy(error), QueryExecutionError::Agent)
+            if let Some(agent_error) = typed_agent_error {
+                return QueryExecutionError::Agent(agent_error);
+            }
+            // PostgreSQL reports a cursor position as a marker suffix on the
+            // driver message. Resolve it here, while the executed statement text
+            // is still available, into a typed field. The marker is stripped even
+            // when it cannot be resolved, so it can never reach a user-facing
+            // message. The PostgreSQL driver also backs Redshift/GaussDB/Kwdb/
+            // QuestDB/openGauss, so this is not gated on `DatabaseType::Postgres`.
+            if error.contains(crate::sql_error_position::SQL_ERROR_POSITION_MARKER) {
+                let (message, position) = crate::sql_error_position::take_message_position(&error, sql);
+                return match position {
+                    Some(position) => QueryExecutionError::SqlWithPosition { message, position },
+                    None => QueryExecutionError::Legacy(message),
+                };
+            }
+            QueryExecutionError::Legacy(error)
         })
         .map_err(|error| classify_query_error(pool_db_type, error))
 }
@@ -8749,6 +8801,46 @@ for line in sys.stdin:
         assert_eq!(error.into_backend_error().code(), "DBX-JDBC-1002");
     }
 
+    #[test]
+    fn postgres_sql_error_with_position_survives_classification_and_legacy_rendering() {
+        let sql = "SELECT *\nFROM no_such_table";
+        let cursor = sql.find("no_such_table").unwrap() as u32 + 1;
+        let raw = format!(
+            "ERROR: relation \"no_such_table\" does not exist{}",
+            crate::sql_error_position::encode_marker(cursor)
+        );
+        let (message, position) = crate::sql_error_position::take_message_position(&raw, sql);
+        let error = QueryExecutionError::SqlWithPosition { message, position: position.unwrap() };
+
+        // The transport marker must never reach the user-facing message.
+        let legacy = error.clone().into_legacy_string();
+        assert!(!legacy.contains(crate::sql_error_position::SQL_ERROR_POSITION_MARKER));
+        assert_eq!(legacy, "ERROR: relation \"no_such_table\" does not exist");
+
+        // Classification must not downgrade the typed variant.
+        let classified = classify_query_error(Some(DatabaseType::Postgres), error.clone());
+        assert!(matches!(classified, QueryExecutionError::SqlWithPosition { .. }));
+
+        let backend_error = classified.into_backend_error();
+        assert_eq!(backend_error.code(), "DBX-JDBC-4001");
+        let position = backend_error.error_position().expect("position must survive into the envelope");
+        assert_eq!((position.line, position.column), (2, 6));
+    }
+
+    #[test]
+    fn legacy_rendering_strips_a_leftover_transport_marker() {
+        // Guards the driver-error path that never went through the resolve step
+        // (e.g. a PostgreSQL-family driver error shown as a legacy string).
+        let error = QueryExecutionError::Legacy(format!(
+            "ERROR: boom{}{}",
+            crate::sql_error_position::encode_marker(9),
+            crate::sql_error_position::encode_marker(2)
+        ));
+        let rendered = error.into_legacy_string();
+        assert_eq!(rendered, "ERROR: boom");
+        assert!(!rendered.contains(crate::sql_error_position::SQL_ERROR_POSITION_MARKER));
+    }
+
     #[test]
     fn duckdb_worker_error_preserves_catalog_identity_and_detail() {
         let error = QueryExecutionError::DuckDb {

crates/dbx-core/src/sql_error_position.rs

@@ -0,0 +1,255 @@
+//! Driver-reported SQL error positions (currently PostgreSQL only).
+//!
+//! PostgreSQL reports a 1-based *character* cursor position for many SQL
+//! errors ([`tokio_postgres::error::ErrorPosition::Original`]). The native
+//! driver error path in `db::postgres` only exposes a `String` to the query
+//! layer, so the raw cursor is carried across that `String` ABI as a small
+//! machine-readable suffix and resolved back into a typed [`SqlErrorPosition`]
+//! as soon as the executed statement text is available.
+//!
+//! The suffix is a position carrier, not an error classifier: nothing about the
+//! error category, retryability or recovery is inferred from it, and
+//! [`strip_marker`] guarantees it can never surface in a user-facing message.
+
+use serde::{Deserialize, Serialize};
+
+/// Suffix appended to a driver error message to carry a raw PostgreSQL cursor
+/// position across the `Result<_, String>` boundary of the `db` layer.
+pub const SQL_ERROR_POSITION_MARKER: &str = "\nDBX_SQL_ERROR_POSITION:";
+
+/// A SQL error position relative to the statement text that was actually sent
+/// to the database.
+///
+/// `line`/`column` are 1-based and counted in Unicode scalar values (matching
+/// PostgreSQL's character-based cursor positions); `offset` is the 0-based
+/// scalar-value index into the statement text.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct SqlErrorPosition {
+    pub line: u32,
+    pub column: u32,
+    pub offset: u32,
+}
+
+impl SqlErrorPosition {
+    /// Convert a PostgreSQL 1-based character cursor into a line/column/offset.
+    ///
+    /// Returns `None` when the cursor is unusable (zero, non-positive, or the
+    /// statement is empty). A cursor past the end (`position == len + 1`, e.g.
+    /// "unexpected end of input") is clamped to the last character.
+    pub fn from_pg_cursor(statement_sql: &str, cursor: u32) -> Option<Self> {
+        if cursor == 0 {
+            return None;
+        }
+        let total = statement_sql.chars().count() as u32;
+        if total == 0 {
+            return None;
+        }
+        let target = (cursor - 1).min(total - 1);
+        let mut line = 1u32;
+        let mut column = 1u32;
+        for (index, ch) in statement_sql.chars().enumerate() {
+            if index as u32 == target {
+                return Some(Self { line, column, offset: target });
+            }
+            if ch == '\n' {
+                line += 1;
+                column = 1;
+            } else {
+                column += 1;
+            }
+        }
+        None
+    }
+}
+
+/// Encode a raw cursor as the transport suffix.
+pub fn encode_marker(cursor: u32) -> String {
+    format!("{SQL_ERROR_POSITION_MARKER}{cursor}")
+}
+
+/// Remove every marker from `message` and return the cursors they carried.
+///
+/// Markers are removed even when followed by more text (the read-only
+/// transaction path can append a cleanup error after one, e.g.
+/// `...<marker>15; rollback failed`), so the trailing text is preserved. A
+/// malformed or absent suffix leaves `message` untouched and is not collected.
+fn take_all_markers(message: &mut String) -> Vec<u32> {
+    let mut cursors = Vec::new();
+    while let Some(cursor) = take_marker(message) {
+        cursors.push(cursor);
+    }
+    cursors
+}
+
+/// Remove a marker from `message` and return the raw cursor it carried.
+///
+/// See [`take_all_markers`] for the multi-marker behaviour.
+pub fn take_marker(message: &mut String) -> Option<u32> {
+    let index = message.rfind(SQL_ERROR_POSITION_MARKER)?;
+    let digits_start = index + SQL_ERROR_POSITION_MARKER.len();
+    let digits_end = message[digits_start..]
+        .find(|ch: char| !ch.is_ascii_digit())
+        .map_or(message.len(), |offset| digits_start + offset);
+    let cursor = message[digits_start..digits_end].parse::<u32>().ok()?;
+    message.replace_range(index..digits_end, "");
+    Some(cursor)
+}
+
+/// Return `message` with every transport suffix stripped.
+pub fn strip_marker(message: &str) -> String {
+    let mut owned = message.to_string();
+    let _ = take_all_markers(&mut owned);
+    owned
+}
+
+/// Strip all transport suffixes from `message` and, when possible, resolve the
+/// carried cursor against the executed statement text into a typed position.
+///
+/// The returned message is always marker-free. The position is only resolved
+/// when exactly one marker was present: a message carrying several markers (e.g.
+/// a user-statement error merged with an infrastructure error) cannot be
+/// attributed to a single statement, and guessing would point at the wrong spot.
+pub fn take_message_position(message: &str, statement_sql: &str) -> (String, Option<SqlErrorPosition>) {
+    let mut cleaned = message.to_string();
+    let cursors = take_all_markers(&mut cleaned);
+    if cursors.len() != 1 {
+        return (cleaned, None);
+    }
+    let position = SqlErrorPosition::from_pg_cursor(statement_sql, cursors[0]);
+    (cleaned, position)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn cursor_maps_to_line_and_column() {
+        let sql = "SELECT *\nFROM no_such_table";
+        // 1-based cursor of the 'n' in no_such_table (second line, column 6).
+        let cursor = sql.find("no_such_table").unwrap() as u32 + 1;
+        let position = SqlErrorPosition::from_pg_cursor(sql, cursor).unwrap();
+        assert_eq!(position.line, 2);
+        assert_eq!(position.column, 6);
+        assert_eq!(position.offset, sql[..sql.find("no_such_table").unwrap()].chars().count() as u32);
+    }
+
+    #[test]
+    fn first_character_is_line_one_column_one() {
+        let position = SqlErrorPosition::from_pg_cursor("SELECT 1", 1).unwrap();
+        assert_eq!((position.line, position.column, position.offset), (1, 1, 0));
+    }
+
+    #[test]
+    fn crlf_treats_carriage_return_as_a_column() {
+        let sql = "SELECT 1\r\nFROM t";
+        let cursor = sql.find("FROM").unwrap() as u32 + 1;
+        let position = SqlErrorPosition::from_pg_cursor(sql, cursor).unwrap();
+        assert_eq!(position.line, 2);
+        assert_eq!(position.column, 1);
+    }
+
+    #[test]
+    fn multi_byte_characters_count_as_one_column() {
+        // '表' is 3 UTF-8 bytes; cursor positions are character based.
+        let sql = "SELECT '表' FROM t";
+        let from_byte = sql.find("FROM").unwrap();
+        let cursor = sql[..from_byte].chars().count() as u32 + 1;
+        let position = SqlErrorPosition::from_pg_cursor(sql, cursor).unwrap();
+        assert_eq!(position.line, 1);
+        assert_eq!(position.column, 12);
+        assert_eq!(position.offset, 11);
+    }
+
+    #[test]
+    fn cursor_past_end_is_clamped() {
+        let sql = "SELECT";
+        let position = SqlErrorPosition::from_pg_cursor(sql, 999).unwrap();
+        assert_eq!(position.offset, 5);
+        assert_eq!(position.column, 6);
+    }
+
+    #[test]
+    fn zero_cursor_and_empty_sql_are_rejected() {
+        assert!(SqlErrorPosition::from_pg_cursor("SELECT 1", 0).is_none());
+        assert!(SqlErrorPosition::from_pg_cursor("", 1).is_none());
+    }
+
+    #[test]
+    fn marker_round_trips_and_strips() {
+        let mut message = format!("ERROR: relation does not exist{}", encode_marker(15));
+        assert_eq!(take_marker(&mut message), Some(15));
+        assert_eq!(message, "ERROR: relation does not exist");
+
+        let mut without_marker = "ERROR: nope".to_string();
+        assert_eq!(take_marker(&mut without_marker), None);
+        assert_eq!(without_marker, "ERROR: nope");
+    }
+
+    #[test]
+    fn malformed_marker_is_left_untouched() {
+        let text = format!("{SQL_ERROR_POSITION_MARKER}not-a-number");
+        let mut message = text.clone();
+        assert_eq!(take_marker(&mut message), None);
+        assert_eq!(message, text);
+    }
+
+    #[test]
+    fn marker_followed_by_text_is_stripped_and_preserves_the_tail() {
+        let mut message = format!("ERROR: boom{}; rollback failed", encode_marker(7));
+        assert_eq!(take_marker(&mut message), Some(7));
+        assert_eq!(message, "ERROR: boom; rollback failed");
+    }
+
+    #[test]
+    fn take_message_position_strips_marker_and_resolves_position() {
+        let sql = "SELECT 1\nFROM missing";
+        let cursor = sql.find("missing").unwrap() as u32 + 1;
+        let raw = format!("ERROR: relation \"missing\" does not exist{}", encode_marker(cursor));
+        let (message, position) = take_message_position(&raw, sql);
+        assert_eq!(message, "ERROR: relation \"missing\" does not exist");
+        let position = position.unwrap();
+        assert_eq!(position.line, 2);
+        assert_eq!(position.column, 6);
+    }
+
+    #[test]
+    fn take_message_position_still_strips_when_the_cursor_is_unusable() {
+        // cursor 0 cannot be resolved, but the marker must not leak.
+        let raw = format!("ERROR: boom{}", encode_marker(0));
+        let (message, position) = take_message_position(&raw, "SELECT 1");
+        assert_eq!(message, "ERROR: boom");
+        assert!(position.is_none());
+
+        // Empty statement: nothing to map onto.
+        let raw = format!("ERROR: boom{}", encode_marker(5));
+        let (message, position) = take_message_position(&raw, "");
+        assert_eq!(message, "ERROR: boom");
+        assert!(position.is_none());
+    }
+
+    #[test]
+    fn take_message_position_without_marker_is_identity() {
+        let (message, position) = take_message_position("ERROR: x", "SELECT 1");
+        assert_eq!(message, "ERROR: x");
+        assert!(position.is_none());
+    }
+
+    #[test]
+    fn multiple_markers_are_all_stripped_and_never_resolved() {
+        // A user-statement error merged with an infrastructure error carries two
+        // cursors; the position cannot be attributed to a single statement.
+        let raw =
+            format!("ERROR: relation missing does not exist{}; cleanup failed{}", encode_marker(15), encode_marker(3));
+        let (message, position) = take_message_position(&raw, "SELECT * FROM missing");
+        assert_eq!(message, "ERROR: relation missing does not exist; cleanup failed");
+        assert!(position.is_none());
+        assert_eq!(strip_marker(&raw), "ERROR: relation missing does not exist; cleanup failed");
+    }
+
+    #[test]
+    fn strip_marker_without_marker_is_identity() {
+        assert_eq!(strip_marker("ERROR: x"), "ERROR: x");
+    }
+}

crates/dbx-core/tests/live_postgres_error_position.rs

@@ -0,0 +1,233 @@
+#![recursion_limit = "256"]
+//! Live PostgreSQL verification of the SQL error-position plumbing.
+//!
+//! Requires a connectable PostgreSQL-family server pointed at by
+//! `DBX_LIVE_POSTGRES_*` (same variables as the other `live_postgres_*` tests).
+//! `DBX_LIVE_PG_FAMILY=opengauss` switches the full-funnel test to an openGauss
+//! connection config, to prove the position is not gated on `DatabaseType::Postgres`.
+//!
+//! ```text
+//! DBX_LIVE_POSTGRES_HOST=... DBX_LIVE_POSTGRES_PORT=... \
+//! DBX_LIVE_POSTGRES_USER=... DBX_LIVE_POSTGRES_PASSWORD=... \
+//! DBX_LIVE_POSTGRES_DATABASE=... \
+//! cargo test -p dbx-core --no-default-features --test live_postgres_error_position -- --ignored
+//! ```
+
+use std::time::Duration;
+
+use dbx_core::connection::AppState;
+use dbx_core::db::postgres;
+use dbx_core::models::connection::{ConnectionConfig, DatabaseType};
+use dbx_core::query::{execute_multi_core_with_options_for_client_and_progress_typed, QueryExecutionOptions};
+use dbx_core::sql_error_position::{encode_marker, take_message_position, SqlErrorPosition, SQL_ERROR_POSITION_MARKER};
+use dbx_core::storage::Storage;
+
+fn live_env() -> (String, u16, String, String, String) {
+    let host = std::env::var("DBX_LIVE_POSTGRES_HOST").unwrap_or_else(|_| "127.0.0.1".to_string());
+    let port = std::env::var("DBX_LIVE_POSTGRES_PORT").ok().and_then(|value| value.parse().ok()).unwrap_or(5432);
+    let user = std::env::var("DBX_LIVE_POSTGRES_USER").unwrap_or_else(|_| "postgres".to_string());
+    let password = std::env::var("DBX_LIVE_POSTGRES_PASSWORD").unwrap_or_default();
+    let database = std::env::var("DBX_LIVE_POSTGRES_DATABASE").unwrap_or_else(|_| "postgres".to_string());
+    (host, port, user, password, database)
+}
+
+fn postgres_url() -> String {
+    let (host, port, user, password, database) = live_env();
+    format!("postgres://{}:{}@{host}:{port}/{database}", encode_userinfo(&user), encode_userinfo(&password))
+}
+
+/// Percent-encode the URL userinfo so secrets containing `@`/`:` survive parsing.
+fn encode_userinfo(value: &str) -> String {
+    let mut encoded = String::with_capacity(value.len());
+    for character in value.chars() {
+        match character {
+            '%' => encoded.push_str("%25"),
+            '@' => encoded.push_str("%40"),
+            ':' => encoded.push_str("%3A"),
+            '/' => encoded.push_str("%2F"),
+            other => encoded.push(other),
+        }
+    }
+    encoded
+}
+
+fn live_db_type() -> DatabaseType {
+    if std::env::var("DBX_LIVE_PG_FAMILY").is_ok_and(|value| value.eq_ignore_ascii_case("opengauss")) {
+        DatabaseType::OpenGauss
+    } else {
+        DatabaseType::Postgres
+    }
+}
+
+fn expected_position(sql: &str, offending: &str) -> SqlErrorPosition {
+    let byte_index = sql.find(offending).unwrap();
+    let offset = sql[..byte_index].chars().count() as u32;
+    let line = sql[..byte_index].matches('\n').count() as u32 + 1;
+    let line_start = sql[..byte_index].rfind('\n').map_or(0, |index| index + 1);
+    let column = sql[line_start..byte_index].chars().count() as u32 + 1;
+    SqlErrorPosition { line, column, offset }
+}
+
+#[tokio::test]
+#[ignore = "requires DBX_LIVE_POSTGRES_* pointing at a PostgreSQL-family database"]
+async fn live_pg_error_positions_resolve_against_the_executed_statement() {
+    let pool = postgres::connect(&postgres_url(), Duration::from_secs(10)).await.expect("connect");
+
+    // (sql, offending token). The first case is the exact shape DBX sends for the
+    // user's `select  * from no_such_table` after pagination appends LIMIT.
+    let cases = [
+        ("select  * from no_such_table LIMIT 100", "no_such_table"),
+        ("SELECT *\nFROM no_such_table", "no_such_table"),
+        ("SELECT no_such_col FROM (SELECT 1 AS x) AS t", "no_such_col"),
+        ("INSERT INTO no_such_table VALUES (1)", "no_such_table"),
+    ];
+
+    for (sql, offending) in cases {
+        let message = postgres::execute_query(&pool, sql).await.expect_err("statement must fail");
+        assert!(message.contains(SQL_ERROR_POSITION_MARKER), "driver error must carry the marker: {message}");
+
+        let (cleaned, position) = take_message_position(&message, sql);
+        assert!(!cleaned.contains(SQL_ERROR_POSITION_MARKER), "marker must be stripped: {cleaned}");
+        assert_eq!(position, Some(expected_position(sql, offending)), "position for `{sql}`");
+    }
+
+    let syntax_sql = "SELECT * FROMM no_such_table";
+    let message = postgres::execute_query(&pool, syntax_sql).await.expect_err("syntax error");
+    let (_, position) = take_message_position(&message, syntax_sql);
+    assert_eq!(position, Some(expected_position(syntax_sql, "FROMM")), "syntax error should point at FROMM");
+
+    pool.close();
+}
+
+/// Full backend funnel: AppState + connection config → `execute_multi_core` →
+/// `BackendError`. This is where the old `pool_db_type == Postgres` gate lived, so
+/// running it with `DBX_LIVE_PG_FAMILY=opengauss` guards the regression.
+#[tokio::test]
+#[ignore = "requires DBX_LIVE_POSTGRES_* pointing at a PostgreSQL-family database"]
+async fn live_pg_multi_core_reports_error_position_in_the_backend_envelope() {
+    let (host, port, user, password, database) = live_env();
+    let db_type = live_db_type();
+
+    let dir = std::env::temp_dir().join(format!("dbx-live-pg-position-{}", uuid::Uuid::new_v4().simple()));
+    std::fs::create_dir_all(&dir).unwrap();
+    let storage = Storage::open(&dir.join("storage.db")).await.unwrap();
+    let state = AppState::new(storage);
+    let connection_id = "live-pg-position";
+    state.configs.write().await.insert(
+        connection_id.to_string(),
+        live_connection_config(connection_id, db_type, &host, port, &user, &password, &database),
+    );
+
+    let sql = "select  * from no_such_table LIMIT 100";
+    let result = execute_multi_core_with_options_for_client_and_progress_typed(
+        &state,
+        connection_id,
+        &database,
+        sql,
+        None,
+        None,
+        QueryExecutionOptions { timeout_secs: Some(30), ..Default::default() },
+        None,
+    )
+    .await;
+
+    let backend_error = result.expect_err("single-statement failure must surface as an error").into_backend_error();
+    assert_eq!(backend_error.code(), "DBX-JDBC-4001");
+    assert_eq!(
+        backend_error.error_position(),
+        Some(expected_position(sql, "no_such_table")),
+        "db_type={db_type:?} detail={:?}",
+        backend_error.detail()
+    );
+    let detail = backend_error.detail().unwrap_or_default();
+    assert!(!detail.contains(SQL_ERROR_POSITION_MARKER), "detail must not leak the marker: {detail}");
+    assert!(detail.contains("no_such_table"), "detail should keep the native message: {detail}");
+
+    let _ = std::fs::remove_dir_all(&dir);
+}
+
+/// A second case proving the multi-marker guard: a marker-containing error that
+/// gets another marker merged in must not resolve to a wrong position.
+#[test]
+fn marker_guard_rejects_ambiguous_merged_errors() {
+    let sql = "SELECT 1";
+    let raw = format!("ERROR: boom{}{}", encode_marker(1), encode_marker(50));
+    let (cleaned, position) = take_message_position(&raw, sql);
+    assert!(position.is_none());
+    assert_eq!(cleaned, "ERROR: boom");
+}
+
+fn live_connection_config(
+    id: &str,
+    db_type: DatabaseType,
+    host: &str,
+    port: u16,
+    user: &str,
+    password: &str,
+    database: &str,
+) -> ConnectionConfig {
+    ConnectionConfig {
+        docs_notes_path: None,
+        id: id.to_string(),
+        name: id.to_string(),
+        note: String::new(),
+        db_type,
+        driver_profile: None,
+        driver_label: None,
+        url_params: None,
+        agent_java_options: Vec::new(),
+        host: host.to_string(),
+        port,
+        username: user.to_string(),
+        password: password.to_string(),
+        database: Some(database.to_string()),
+        default_schema: None,
+        visible_databases: None,
+        visible_database_patterns: None,
+        visible_schemas: None,
+        attached_databases: Vec::new(),
+        init_script: None,
+        color: None,
+        transport_layers: Vec::new(),
+        connect_timeout_secs: 10,
+        query_timeout_secs: 30,
+        idle_timeout_secs: 60,
+        keepalive_interval_secs: 0,
+        ssl: false,
+        ca_cert_path: String::new(),
+        client_cert_path: String::new(),
+        client_key_path: String::new(),
+        sysdba: false,
+        oracle_connection_type: None,
+        connection_string: None,
+        redis_connection_mode: None,
+        redis_sentinel_master: String::new(),
+        redis_sentinel_nodes: String::new(),
+        redis_sentinel_username: String::new(),
+        redis_sentinel_password: String::new(),
+        redis_sentinel_tls: false,
+        redis_cluster_nodes: String::new(),
+        redis_key_separator: dbx_core::models::connection::default_redis_key_separator(),
+        redis_scan_page_size: None,
+        redis_database_aliases: Default::default(),
+        redis_key_templates: Vec::new(),
+        redis_key_grouping: None,
+        etcd_endpoints: String::new(),
+        gbase_server: String::new(),
+        informix_server: String::new(),
+        external_config: None,
+        plugin_id: None,
+        plugin_connection_provider: None,
+        plugin_connection_type: None,
+        connection_secrets: Default::default(),
+        jdbc_driver_class: None,
+        jdbc_driver_paths: Vec::new(),
+        one_time: false,
+        save_password: true,
+        read_only: false,
+        is_production: false,
+        production_databases: vec![],
+        show_system_schemas: false,
+        database_info: None,
+    }
+}