fix(mysql): Work around for Issue #2206 (ColumnNotFound error when querying) (#4086)

* Re-read mysql column info during execute

MariaDB may change the column info between PREPARE and EXECUTE if the schema changes
(https://jira.mariadb.org/browse/MDEV-27013). Therefore, always read column info from
the execute metadata and use it for the row column_names field. Fixes: #2206, #1530

* doc: caution that ColumnIndex may differ between Statement and Row

Author: duelafn

sqlx-core/src/column.rs

@@ -84,6 +84,11 @@ impl ColumnOrigin {
 /// This trait is implemented for strings which are used to look up a column by name, and for
 /// `usize` which is used as a positional index into the row.
 ///
+/// *Caution*: The column index may differ between a [`Statement`] and a [`Row`] returned by the
+/// statement. This can happen with some databases if, for example, the schema changes between
+/// prepare and execute or if the database does not provide column information when the statement
+/// is prepared.
+///
 /// [`Row`]: crate::row::Row
 /// [`Statement`]: crate::statement::Statement
 /// [`get`]: crate::row::Row::get

sqlx-mysql/src/connection/executor.rs

@@ -120,7 +120,7 @@ impl MySqlConnection {
             // to re-use this memory freely between result sets
             let mut columns = Arc::new(Vec::new());
 
-            let (mut column_names, format, mut needs_metadata) = if let Some(arguments) = arguments {
+            let format = if let Some(arguments) = arguments {
                 if persistent && self.inner.cache_statement.is_enabled() {
                     let (id, metadata) = self
                         .get_or_prepare_statement(sql)
@@ -144,7 +144,7 @@ impl MySqlConnection {
                         })
                         .await?;
 
-                    (metadata.column_names, MySqlValueFormat::Binary, false)
+                    MySqlValueFormat::Binary
                 } else {
                     let (id, metadata) = self
                         .prepare_statement(sql)
@@ -170,13 +170,13 @@ impl MySqlConnection {
 
                     self.inner.stream.send_packet(StmtClose { statement: id }).await?;
 
-                    (metadata.column_names, MySqlValueFormat::Binary, false)
+                    MySqlValueFormat::Binary
                 }
             } else {
                 // https://dev.mysql.com/doc/internals/en/com-query.html
                 self.inner.stream.send_packet(Query(sql)).await?;
 
-                (Arc::default(), MySqlValueFormat::Text, true)
+                MySqlValueFormat::Text
             };
 
             loop {
@@ -216,15 +216,9 @@ impl MySqlConnection {
                 let num_columns = usize::try_from(num_columns)
                     .map_err(|_| err_protocol!("column count overflows usize: {num_columns}"))?;
 
-                if needs_metadata {
-                    column_names = Arc::new(recv_result_metadata(&mut self.inner.stream, num_columns, Arc::make_mut(&mut columns)).await?);
-                } else {
-                    // next time we hit here, it'll be a new result set and we'll need the
-                    // full metadata
-                    needs_metadata = true;
-
-                    recv_result_columns(&mut self.inner.stream, num_columns, Arc::make_mut(&mut columns)).await?;
-                }
+                // Always reload column names, even for prepared statements (the schema
+                // may change between PREPARE and EXECUTE).
+                let column_names = Arc::new(recv_result_metadata(&mut self.inner.stream, num_columns, Arc::make_mut(&mut columns)).await?);
 
                 // finally, there will be none or many result-rows
                 loop {
@@ -395,25 +389,6 @@ impl<'c> Executor<'c> for &'c mut MySqlConnection {
     }
 }
 
-async fn recv_result_columns(
-    stream: &mut MySqlStream,
-    num_columns: usize,
-    columns: &mut Vec<MySqlColumn>,
-) -> Result<(), Error> {
-    columns.clear();
-    columns.reserve(num_columns);
-
-    for ordinal in 0..num_columns {
-        columns.push(recv_next_result_column(&stream.recv().await?, ordinal)?);
-    }
-
-    if num_columns > 0 {
-        stream.maybe_recv_eof().await?;
-    }
-
-    Ok(())
-}
-
 fn recv_next_result_column(def: &ColumnDefinition, ordinal: usize) -> Result<MySqlColumn, Error> {
     // if the alias is empty, use the alias
     // only then use the name

tests/mysql/mysql.rs

@@ -636,3 +636,33 @@ async fn issue_3200() -> anyhow::Result<()> {
 
     Ok(())
 }
+
+#[cfg(mariadb)]
+#[sqlx_macros::test]
+async fn it_can_name_columns_issue_2206() -> anyhow::Result<()> {
+    let mut conn = new::<MySql>().await?;
+
+    sqlx::raw_sql(
+        "\
+        CREATE TABLE IF NOT EXISTS issue_2206
+        (
+            `id`                     BIGINT AUTO_INCREMENT,
+            `name`                   VARCHAR(128) NOT NULL,
+            PRIMARY KEY (id)
+        );
+    ",
+    )
+    .execute(&mut conn)
+    .await?;
+
+    let row = sqlx::query("INSERT INTO issue_2206 (name) VALUES (?) RETURNING *")
+        .bind("Alice")
+        .fetch_one(&mut conn)
+        .await?;
+    let _id: i64 = row.get("id");
+    let name: String = row.get("name");
+
+    assert_eq!(&name, "Alice");
+
+    Ok(())
+}