diff --git a/crates/dbx-core/src/query/mod.rs b/crates/dbx-core/src/query/mod.rs
index 50a25d7df3..c9dc8f3e7d 100644
--- a/crates/dbx-core/src/query/mod.rs
+++ b/crates/dbx-core/src/query/mod.rs
@@ -6612,6 +6612,29 @@ async fn execute_manual_txn_mysql_statement(
         let mut result = conn.query_iter(sql).await.map_err(|e| format!("Query failed: {e}"))?;
         let columns: Vec<String> = result.columns_ref().iter().map(|c| c.name_str().to_string()).collect();
         let column_types: Vec<String> = result.columns_ref().iter().map(db::mysql::mysql_column_type_name).collect();
+        // Some statements only *may* return rows — `EXECUTE` of a prepared DML statement
+        // finishes with a plain OK packet instead — so report the affected rows rather than
+        // failing on the missing result set.
+        if columns.is_empty() {
+            let affected_rows = result.affected_rows();
+            result.drop_result().await.map_err(|e| format!("Query failed: {e}"))?;
+            return Ok(db::QueryResult {
+                columns: Vec::new(),
+                column_types: Vec::new(),
+                column_sortables: vec![],
+                spatial_columns: vec![],
+                spatial_values: vec![],
+                rows: vec![],
+                affected_rows,
+                execution_time_ms: start.elapsed().as_millis(),
+                server_execute_time_us: None,
+                truncated: false,
+                session_id: None,
+                has_more: false,
+                elasticsearch_raw_body: None,
+                messages: Vec::new(),
+            });
+        }
         let mut data: Vec<Vec<serde_json::Value>> = Vec::with_capacity(row_limit.min(1024));
         let mut stream = result
             .stream::<mysql_async::Row>()
diff --git a/crates/dbx-drivers/src/db/mysql.rs b/crates/dbx-drivers/src/db/mysql.rs
index 4bcb867548..2c97fb3308 100644
--- a/crates/dbx-drivers/src/db/mysql.rs
+++ b/crates/dbx-drivers/src/db/mysql.rs
@@ -5878,11 +5878,15 @@ fn prefers_text_protocol_query(sql: &str, dialect: MySqlQueryDialect) -> bool {
 pub fn is_result_set_query(sql: &str, dialect: MySqlQueryDialect) -> bool {
     // MySQL 的表维护语句虽然不是 SELECT，但服务器会返回包含表名和执行结果的表格。
     // 如果把它们当成普通写入语句，后续 drop_result 会直接丢弃这些返回行。
+    //
+    // `EXECUTE` 同理：它执行的是运行时才确定的动态语句，`PREPARE` 的是查询时服务器已经
+    // 把结果集发回来了，当成普通写入语句处理就会把这些行丢掉（#10005）。而且 MySQL 不
+    // 允许在预处理协议里执行 `EXECUTE`（ERROR 1295），所以它必须走文本协议。
     starts_with_executable_sql_keyword_for_database(
         sql,
         &[
             "SELECT", "SHOW", "DESCRIBE", "EXPLAIN", "WITH", "CALL", "CHECKSUM", "ANALYZE", "CHECK", "OPTIMIZE",
-            "REPAIR",
+            "REPAIR", "EXECUTE",
         ],
         DatabaseType::Mysql,
     ) || mysql_statement_returns_rows(sql)
@@ -6958,6 +6962,20 @@ mod tests {
         assert!(prefers_text_protocol_query("CALL proc_test1()", dialect));
     }
 
+    #[test]
+    fn mysql_execute_statements_are_treated_as_text_result_sets_per_issue_10005() {
+        let dialect = MySqlQueryDialect::default();
+
+        // `EXECUTE` runs a dynamic statement whose result set the server already sent, and it
+        // is rejected by the prepared-statement protocol, so it must take the text-protocol
+        // result-set path instead of the write path that drops rows.
+        for sql in ["EXECUTE stmt", "execute stmt;", "-- run dynamic sql\nEXECUTE stmt", "EXECUTE IMMEDIATE 'SELECT 1'"]
+        {
+            assert!(is_result_set_query(sql, dialect), "{sql}");
+            assert!(prefers_text_protocol_query(sql, dialect), "{sql}");
+        }
+    }
+
     #[test]
     fn mysql_xa_recover_queries_are_treated_as_text_result_sets() {
         let dialect = MySqlQueryDialect::default();