feat: add ability to set or unset the CLIENT_FOUND_ROWS flag

sqlx-mysql/src/connection/stream.rs

@@ -35,7 +35,6 @@ impl<S: Socket> MySqlStream<S> {
         let mut capabilities = Capabilities::PROTOCOL_41
             | Capabilities::IGNORE_SPACE
             | Capabilities::DEPRECATE_EOF
-            | Capabilities::FOUND_ROWS
             | Capabilities::TRANSACTIONS
             | Capabilities::SECURE_CONNECTION
             | Capabilities::PLUGIN_AUTH_LENENC_DATA
@@ -49,6 +48,10 @@ impl<S: Socket> MySqlStream<S> {
             capabilities |= Capabilities::CONNECT_WITH_DB;
         }
 
+        if options.found_rows {
+            capabilities |= Capabilities::FOUND_ROWS;
+        }
+
         Self {
             waiting: VecDeque::new(),
             capabilities,

sqlx-mysql/src/options/mod.rs

@@ -80,6 +80,7 @@ pub struct MySqlConnectOptions {
     pub(crate) no_engine_substitution: bool,
     pub(crate) timezone: Option<String>,
     pub(crate) set_names: bool,
+    pub(crate) found_rows: bool,
 }
 
 impl Default for MySqlConnectOptions {
@@ -111,6 +112,7 @@ impl MySqlConnectOptions {
             no_engine_substitution: true,
             timezone: Some(String::from("+00:00")),
             set_names: true,
+            found_rows: true,
         }
     }
 
@@ -414,6 +416,15 @@ impl MySqlConnectOptions {
         self.set_names = flag_val;
         self
     }
+
+    /// Sets the flag that enables or disables CLIENT_FOUND_ROWS,
+    /// to return the number of found (matched) rows, and not the number of changed rows.
+    ///
+    /// The default value is set to true.
+    pub fn found_rows(mut self, flag_val: bool) -> Self {
+        self.found_rows = flag_val;
+        self
+    }
 }
 
 impl MySqlConnectOptions {

tests/mysql/mysql.rs

@@ -727,3 +727,55 @@ async fn any_blob_conversions() -> anyhow::Result<()> {
 
     Ok(())
 }
+
+#[sqlx_macros::test]
+async fn test_client_found_rows() -> anyhow::Result<()> {
+    setup_if_needed();
+
+    let url = url::Url::parse(&env::var("DATABASE_URL")?)?;
+
+    // CLIENT_AFFECTED_ROWS unspecified = true by default.
+    let mut conn = MySqlConnectOptions::from_url(&url)?.connect().await?;
+    let mut tx = conn.begin().await?;
+
+    tx.execute(sqlx::query(
+        "CREATE TEMPORARY TABLE found_rows_testing (id INT PRIMARY KEY, field INT NOT NULL)",
+    ))
+    .await?;
+
+    let result = tx.execute(sqlx::query("INSERT INTO found_rows_testing VALUES (0, 10) ON DUPLICATE KEY UPDATE field = VALUES(field)")).await?;
+    assert_eq!(result.rows_affected(), 1);
+
+    let result = tx.execute(sqlx::query("INSERT INTO found_rows_testing VALUES (0, 10) ON DUPLICATE KEY UPDATE field = VALUES(field)")).await?;
+    assert_eq!(result.rows_affected(), 1);
+
+    let result = tx.execute(sqlx::query("INSERT INTO found_rows_testing VALUES (0, 20) ON DUPLICATE KEY UPDATE field = VALUES(field)")).await?;
+    assert_eq!(result.rows_affected(), 2);
+
+    tx.rollback().await?;
+
+    // Explicitly unset CLIENT_AFFECTED_ROWS.
+    let mut conn = MySqlConnectOptions::from_url(&url)?
+        .found_rows(false)
+        .connect()
+        .await?;
+    let mut tx = conn.begin().await?;
+
+    tx.execute(sqlx::query(
+        "CREATE TEMPORARY TABLE found_rows_testing (id INT PRIMARY KEY, field INT NOT NULL)",
+    ))
+    .await?;
+
+    let result = tx.execute(sqlx::query("INSERT INTO found_rows_testing VALUES (0, 10) ON DUPLICATE KEY UPDATE field = VALUES(field)")).await?;
+    assert_eq!(result.rows_affected(), 1);
+
+    let result = tx.execute(sqlx::query("INSERT INTO found_rows_testing VALUES (0, 10) ON DUPLICATE KEY UPDATE field = VALUES(field)")).await?;
+    assert_eq!(result.rows_affected(), 0);
+
+    let result = tx.execute(sqlx::query("INSERT INTO found_rows_testing VALUES (0, 20) ON DUPLICATE KEY UPDATE field = VALUES(field)")).await?;
+    assert_eq!(result.rows_affected(), 2);
+
+    tx.rollback().await?;
+
+    Ok(())
+}