fix(mysql): prefer text protocol for result sets

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

@@ -724,6 +724,8 @@ fn mysql_error_should_retry_with_text_protocol(error: &str) -> bool {
     (lower.contains("1105") && lower.contains("hy000"))
         || (lower.contains("1615") && lower.contains("re-prepared"))
         || lower.contains("com_stmt_prepare")
+        || lower.contains("can't parse")
+        || lower.contains("buf doesn't have enough data")
         || lower.contains("prepared statement protocol")
         || lower.contains("this command is not supported in the prepared statement protocol yet")
 }
@@ -1556,7 +1558,7 @@ pub async fn execute_query_on_conn_with_max_rows(
     let row_limit = query_result_row_limit(max_rows);
 
     if is_result_set_query(sql, dialect) {
-        if bare || requires_text_protocol_query(sql, dialect) {
+        if bare || prefers_text_protocol_query(sql, dialect) {
             execute_result_set_with_text_protocol_on_conn(conn, sql, row_limit, start).await
         } else {
             match execute_result_set_with_prepared_protocol_on_conn(conn, sql, row_limit, start).await {
@@ -1595,6 +1597,12 @@ pub async fn execute_query_on_conn_with_max_rows(
     }
 }
 
+fn prefers_text_protocol_query(sql: &str, dialect: MySqlQueryDialect) -> bool {
+    // User-entered result-set queries are not parameterized in DBX. Text protocol
+    // avoids binary result decoding bugs in MySQL-compatible servers and proxies.
+    is_result_set_query(sql, dialect) || requires_text_protocol_query(sql, dialect)
+}
+
 fn is_result_set_query(sql: &str, dialect: MySqlQueryDialect) -> bool {
     starts_with_executable_sql_keyword(sql, &["SELECT", "SHOW", "DESCRIBE", "EXPLAIN", "WITH"])
         || dialect.supports_admin_show_results && is_admin_show_query(sql)
@@ -2018,6 +2026,23 @@ mod tests {
         assert!(!requires_text_protocol_query("SELECT * FROM users", MySqlQueryDialect::default()));
     }
 
+    #[test]
+    fn mysql_user_result_sets_prefer_text_protocol() {
+        let dialect = MySqlQueryDialect::default();
+
+        assert!(prefers_text_protocol_query("SELECT * FROM users", dialect));
+        assert!(prefers_text_protocol_query("WITH recent AS (SELECT 1 AS id) SELECT id FROM recent", dialect));
+        assert!(prefers_text_protocol_query("SHOW TABLES", dialect));
+        assert!(!prefers_text_protocol_query("UPDATE users SET name = 'Ada' WHERE id = 1", dialect));
+    }
+
+    #[test]
+    fn mysql_binary_decode_parse_errors_retry_with_text_protocol() {
+        assert!(mysql_error_should_retry_with_text_protocol(
+            "Input/output error: can't parse: buf doesn't have enough data"
+        ));
+    }
+
     #[test]
     fn mysql_timestamp_default_null_ddl_enables_explicit_defaults() {
         let create_sql = r#"

crates/dbx-core/tests/live_mysql57.rs

@@ -0,0 +1,19 @@
+#[tokio::test]
+#[ignore = "requires the remote DBX MySQL 5.7 smoke-test container"]
+async fn live_mysql57_text_protocol_select_succeeds() {
+    let url = std::env::var("DBX_LIVE_MYSQL57_URL").expect("DBX_LIVE_MYSQL57_URL");
+
+    let pool = dbx_core::db::mysql::connect(&url, std::time::Duration::from_secs(5)).await.unwrap();
+    let result = dbx_core::db::mysql::execute_query_with_max_rows(
+        &pool,
+        "SELECT 1 AS id, CAST('mysql57' AS CHAR) AS label",
+        false,
+        Some(10),
+        Default::default(),
+    )
+    .await
+    .unwrap();
+
+    assert_eq!(result.columns, vec!["id", "label"]);
+    assert_eq!(result.rows, vec![vec![serde_json::json!("1"), serde_json::json!("mysql57")]]);
+}