(BREAKING) add useful information to database/migration errors

sqlx-core/src/error.rs

@@ -193,6 +193,26 @@ pub trait DatabaseError: 'static + Send + Sync + StdError {
         None
     }
 
+    /// The line and column in the executed SQL where the error occurred,
+    /// if applicable and supported by the database.
+    ///
+    /// ### Note
+    /// This may return an incorrect result if the database server disagrees with Rust
+    /// on the definition of a "character", i.e. a Unicode scalar value. This position should not
+    /// be considered authoritative.
+    ///
+    /// This also may not be returned or made readily available by every database flavor.
+    ///
+    /// For example, MySQL and MariaDB do not include the error position as a specific field
+    /// in the `ERR_PACKET` structure; the line number that appears in the error message is part
+    /// of the message string generated by the database server.
+    ///
+    /// SQLx does not attempt to parse the line number from the message string,
+    /// as we cannot assume that the exact message format is a stable part of the API contract.
+    fn position(&self) -> Option<ErrorPosition> {
+        None
+    }
+
     #[doc(hidden)]
     fn as_error(&self) -> &(dyn StdError + Send + Sync + 'static);
 
@@ -320,3 +340,131 @@ macro_rules! err_protocol {
         $crate::error::Error::Protocol(format!($fmt, $($arg)*))
     };
 }
+
+/// Details the position in an SQL string where the server says an error occurred.
+#[derive(Debug, Copy, Clone, PartialEq, Eq)]
+pub struct ErrorPosition {
+    /// The byte offset where the error occurred.
+    pub byte_offset: usize,
+    /// The character (Unicode scalar value) offset where the error occurred.
+    pub char_offset: usize,
+    /// The line number (1-based) in the string.
+    pub line: usize,
+    /// The column position (1-based) in the string.
+    pub column: usize,
+}
+
+/// The character basis for an error position. Used with [`ErrorPosition`].
+#[derive(Debug)]
+pub enum PositionBasis {
+    /// A zero-based byte offset.
+    ByteOffset(usize),
+    /// A zero-based character index.
+    CharIndex(usize),
+    /// A 1-based character position.
+    CharPos(usize),
+}
+
+impl ErrorPosition {
+    /// Given a query string and a character basis (byte offset, 0-based index or 1-based position),
+    /// return the line and column.
+    ///
+    /// Returns `None` if the character basis is out-of-bounds,
+    /// does not lie on a character boundary (byte offsets only),
+    /// or overflows `usize`.
+    ///
+    /// ### Note
+    /// This assumes that Rust and the database server agree on the definition of "character",
+    /// i.e. a Unicode scalar value.
+    pub fn find(query: &str, basis: PositionBasis) -> Option<ErrorPosition> {
+        let mut pos = ErrorPosition { byte_offset: 0, char_offset: 0, line: 1, column: 1 };
+
+        for (char_idx, (byte_idx, ch)) in query.char_indices().enumerate() {
+            pos.byte_offset = byte_idx;
+            pos.char_offset = char_idx;
+
+            // Note: since line and column are 1-based,
+            // we technically don't want to advance until the top of the next loop.
+            if pos.basis_reached(&basis) {
+                return Some(pos);
+            }
+
+            if ch == '\n' {
+                pos.line = pos.line.checked_add(1)?;
+                pos.column = 1;
+            } else {
+                pos.column = pos.column.checked_add(1)?;
+            }
+        }
+
+        // Check if the end of the string matches our basis.
+        pos.byte_offset = query.len();
+        pos.char_offset = pos.char_offset.checked_add(1)?;
+
+        pos.basis_reached(&basis).then_some(pos)
+    }
+
+    fn basis_reached(&self, basis: &PositionBasis) -> bool {
+        match *basis {
+            PositionBasis::ByteOffset(offset) => {
+                self.byte_offset == offset
+            }
+            PositionBasis::CharIndex(char_idx) => {
+                self.char_offset == char_idx
+            }
+            PositionBasis::CharPos(char_pos) => {
+                self.char_offset.checked_add(1) == Some(char_pos)
+            }
+        }
+    }
+}
+
+#[test]
+fn test_error_position() {
+    assert_eq!(
+        ErrorPosition::find(
+            "SELECT foo",
+            PositionBasis::CharPos(8),
+        ),
+        Some(ErrorPosition {
+            byte_offset: 7,
+            char_offset: 7,
+            line: 1,
+            column: 8
+        })
+    );
+
+    assert_eq!(
+        ErrorPosition::find(
+            "SELECT foo\nbar FROM baz",
+            PositionBasis::CharPos(16),
+        ),
+        Some(ErrorPosition {
+            byte_offset: 16,
+            char_offset: 16,
+            line: 2,
+            column: 5
+        })
+    );
+
+    assert_eq!(
+        ErrorPosition::find(
+            "SELECT foo\r\nbar FROM baz",
+            PositionBasis::CharPos(17)
+        ),
+        Some(ErrorPosition {
+            byte_offset: 16,
+            char_offset: 16,
+            line: 2,
+            column: 5
+        })
+    );
+
+    assert_eq!(
+        ErrorPosition::find(
+            "SELECT foo\r\nbar FROM baz",
+            PositionBasis::CharPos(27)
+        ),
+        None
+    );
+}

sqlx-postgres/src/connection/executor.rs

@@ -1,5 +1,5 @@
 use crate::describe::Describe;
-use crate::error::Error;
+use crate::error::{Error, PgResultExt};
 use crate::executor::{Execute, Executor};
 use crate::logger::QueryLogger;
 use crate::message::{
@@ -168,7 +168,9 @@ impl PgConnection {
             return Ok((*statement).clone());
         }
 
-        let statement = prepare(self, sql, parameters, metadata).await?;
+        let statement = prepare(self, sql, parameters, metadata)
+            .await
+            .pg_find_error_pos(sql)?;
 
         if store_to_cache && self.cache_statement.is_enabled() {
             if let Some((id, _)) = self.cache_statement.insert(sql, statement.clone()) {
@@ -267,7 +269,9 @@ impl PgConnection {
 
         Ok(try_stream! {
             loop {
-                let message = self.stream.recv().await?;
+                let message = self.stream.recv()
+                    .await
+                    .pg_find_error_pos(query)?;
 
                 match message.format {
                     MessageFormat::BindComplete

sqlx-postgres/src/connection/stream.rs

@@ -104,7 +104,7 @@ impl PgStream {
             match message.format {
                 MessageFormat::ErrorResponse => {
                     // An error returned from the database server.
-                    return Err(PgDatabaseError(message.decode()?).into());
+                    return Err(PgDatabaseError::new(message.decode()?).into());
                 }
 
                 MessageFormat::NotificationResponse => {