From 3e051d61a76377175281b0412d7c104949fc80fd Mon Sep 17 00:00:00 2001 From: t8y2 <1156263951@qq.com> Date: Sun, 24 May 2026 00:53:27 +0800 Subject: [PATCH] fix(dbx-core): migrate mysql driver to mysql_async Add mysql_async with client_ed25519 support for MySQL and MariaDB connections.\nSwitch MySQL transaction control and schema metadata reads away from sqlx MySQL APIs.\nPreserve connection compatibility by stripping unsupported URL params and handling nullable metadata safely. --- Cargo.lock | 281 +++++++++++++++- crates/dbx-core/Cargo.toml | 3 +- crates/dbx-core/src/connection.rs | 35 +- crates/dbx-core/src/db/mysql.rs | 479 +++++++++++++-------------- crates/dbx-core/src/db/ob_oracle.rs | 95 +++--- crates/dbx-core/src/query.rs | 37 +-- crates/dbx-core/src/schema.rs | 40 ++- src-tauri/src/commands/connection.rs | 10 +- 8 files changed, 625 insertions(+), 355 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 3fad569e4d..afea57811c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -893,6 +893,15 @@ dependencies = [ "uuid", ] +[[package]] +name = "btoi" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9dd6407f73a9b8b6162d8a2ef999fe6afd7cc15902ebf42c5cd296addf17e0ad" +dependencies = [ + "num-traits", +] + [[package]] name = "bumpalo" version = "3.20.2" @@ -1655,6 +1664,21 @@ dependencies = [ "subtle", ] +[[package]] +name = "curve25519-dalek" +version = "4.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "97fb8b7c4503de7d6ae7b42ab72a5a59857b4c937ec27a3d4539dba95b5ab2be" +dependencies = [ + "cfg-if", + "cpufeatures 0.2.17", + "curve25519-dalek-derive", + "fiat-crypto 0.2.9", + "rustc_version", + "subtle", + "zeroize", +] + [[package]] name = "curve25519-dalek" version = "5.0.0-pre.6" @@ -1665,7 +1689,7 @@ dependencies = [ "cpufeatures 0.2.17", "curve25519-dalek-derive", "digest 0.11.3", - "fiat-crypto", + "fiat-crypto 0.3.0", "rustc_version", "subtle", "zeroize", @@ -1682,14 +1706,38 @@ dependencies = [ "syn 2.0.117", ] +[[package]] +name = "darling" +version = "0.20.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc7f46116c46ff9ab3eb1597a45688b6715c6e628b5c133e288e709a29bcb4ee" +dependencies = [ + "darling_core 0.20.11", + "darling_macro 0.20.11", +] + [[package]] name = "darling" version = "0.23.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "25ae13da2f202d56bd7f91c25fba009e7717a1e4a1cc98a76d844b65ae912e9d" dependencies = [ - "darling_core", - "darling_macro", + "darling_core 0.23.0", + "darling_macro 0.23.0", +] + +[[package]] +name = "darling_core" +version = "0.20.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0d00b9596d185e565c2207a0b01f8bd1a135483d02d9b7b0a54b11da8d53412e" +dependencies = [ + "fnv", + "ident_case", + "proc-macro2", + "quote", + "strsim", + "syn 2.0.117", ] [[package]] @@ -1705,13 +1753,24 @@ dependencies = [ "syn 2.0.117", ] +[[package]] +name = "darling_macro" +version = "0.20.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc34b93ccb385b40dc71c6fceac4b2ad23662c7eeb248cf10d529b7e055b6ead" +dependencies = [ + "darling_core 0.20.11", + "quote", + "syn 2.0.117", +] + [[package]] name = "darling_macro" version = "0.23.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ac3984ec7bd6cfa798e62b4a642426a5be0e68f9401cfc2a01e3fa9ea2fcdb8d" dependencies = [ - "darling_core", + "darling_core 0.23.0", "quote", "syn 2.0.117", ] @@ -1790,6 +1849,7 @@ dependencies = [ "iana-time-zone", "log", "mongodb", + "mysql_async", "percent-encoding", "portpicker", "redis", @@ -2185,7 +2245,7 @@ version = "3.0.0-pre.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "053618a4c3d3bc24f188aa660ae75a46eeab74ef07fb415c61431e5e7cd4749b" dependencies = [ - "curve25519-dalek", + "curve25519-dalek 5.0.0-pre.6", "ed25519", "rand_core 0.10.1", "serde", @@ -2418,6 +2478,12 @@ dependencies = [ "log", ] +[[package]] +name = "fiat-crypto" +version = "0.2.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "28dea519a9695b9977216879a3ebfddf92f1c08c05d984f8996aecd6ecdc811d" + [[package]] name = "fiat-crypto" version = "0.3.0" @@ -3093,6 +3159,17 @@ dependencies = [ "foldhash 0.1.5", ] +[[package]] +name = "hashbrown" +version = "0.16.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "841d1cc9bed7f9236f321df977030373f4a4163ae1a7dbfe1a51a2c1a51d9100" +dependencies = [ + "allocator-api2", + "equivalent", + "foldhash 0.2.0", +] + [[package]] name = "hashbrown" version = "0.17.0" @@ -3373,7 +3450,7 @@ dependencies = [ "libc", "percent-encoding", "pin-project-lite", - "socket2", + "socket2 0.6.3", "system-configuration", "tokio", "tower-service", @@ -3629,7 +3706,7 @@ version = "0.3.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4d40460c0ce33d6ce4b0630ad68ff63d6661961c48b6dba35e5a4d81cfb48222" dependencies = [ - "socket2", + "socket2 0.6.3", "widestring", "windows-registry 0.6.1", "windows-result 0.4.1", @@ -3839,6 +3916,15 @@ dependencies = [ "unicode-segmentation", ] +[[package]] +name = "keyed_priority_queue" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4ee7893dab2e44ae5f9d0173f26ff4aa327c10b01b06a72b52dd9405b628640d" +dependencies = [ + "indexmap 2.14.0", +] + [[package]] name = "lazy_static" version = "1.5.0" @@ -4068,6 +4154,15 @@ dependencies = [ "value-bag", ] +[[package]] +name = "lru" +version = "0.16.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7f66e8d5d03f609abc3a39e6f08e4164ebf1447a732906d39eb9b99b7919ef39" +dependencies = [ + "hashbrown 0.16.1", +] + [[package]] name = "lru-slab" version = "0.1.2" @@ -4314,7 +4409,7 @@ dependencies = [ "serde_with", "sha1 0.10.6", "sha2 0.10.9", - "socket2", + "socket2 0.6.3", "stringprep", "strsim", "take_mut", @@ -4377,6 +4472,86 @@ dependencies = [ "version_check", ] +[[package]] +name = "mysql-common-derive" +version = "0.32.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "66f62cad7623a9cb6f8f64037f0c4f69c8db8e82914334a83c9788201c2c1bfa" +dependencies = [ + "darling 0.20.11", + "heck 0.5.0", + "num-bigint", + "proc-macro-crate 3.5.0", + "proc-macro-error2", + "proc-macro2", + "quote", + "syn 2.0.117", + "termcolor", + "thiserror 2.0.18", +] + +[[package]] +name = "mysql_async" +version = "0.36.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d1d9585dc9058886ff3a1f48a23024dd1d054264dee7c5ae0e4bd640c953bee5" +dependencies = [ + "bytes", + "crossbeam-queue", + "flate2", + "futures-core", + "futures-sink", + "futures-util", + "keyed_priority_queue", + "lru", + "mysql_common", + "pem", + "percent-encoding", + "rand 0.9.4", + "rustls 0.23.40", + "rustls-pemfile 2.2.0", + "serde", + "serde_json", + "socket2 0.5.10", + "thiserror 2.0.18", + "tokio", + "tokio-rustls 0.26.4", + "tokio-util", + "twox-hash", + "url", + "webpki-roots 0.26.11", +] + +[[package]] +name = "mysql_common" +version = "0.35.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fbb9f371618ce723f095c61fbcdc36e8936956d2b62832f9c7648689b338e052" +dependencies = [ + "base64 0.22.1", + "bitflags 2.11.1", + "btoi", + "byteorder", + "bytes", + "chrono", + "crc32fast", + "curve25519-dalek 4.1.3", + "flate2", + "getrandom 0.3.4", + "mysql-common-derive", + "num-bigint", + "num-traits", + "regex", + "rust_decimal", + "saturating", + "serde", + "serde_json", + "sha1 0.10.6", + "sha2 0.10.9", + "thiserror 2.0.18", + "uuid", +] + [[package]] name = "native-tls" version = "0.2.18" @@ -4913,7 +5088,7 @@ checksum = "8c91df688211f5957dbe2ab599dcbcaade8d6d3cdc15c5b350d350d7d07ce423" dependencies = [ "ecdsa", "elliptic-curve", - "fiat-crypto", + "fiat-crypto 0.3.0", "primefield", "primeorder", "sha2 0.11.0", @@ -5062,6 +5237,16 @@ dependencies = [ "hmac 0.13.0", ] +[[package]] +name = "pem" +version = "3.0.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1d30c53c26bc5b31a98cd02d20f25a7c8567146caf63ed593a9d87b2775291be" +dependencies = [ + "base64 0.22.1", + "serde_core", +] + [[package]] name = "pem-rfc7468" version = "0.7.0" @@ -5458,6 +5643,28 @@ dependencies = [ "version_check", ] +[[package]] +name = "proc-macro-error-attr2" +version = "2.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "96de42df36bb9bba5542fe9f1a054b8cc87e172759a1868aa05c1f3acc89dfc5" +dependencies = [ + "proc-macro2", + "quote", +] + +[[package]] +name = "proc-macro-error2" +version = "2.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "11ec05c52be0a07b08061f7dd003e7d7092e0472bc731b4af7bb1ef876109802" +dependencies = [ + "proc-macro-error-attr2", + "proc-macro2", + "quote", + "syn 2.0.117", +] + [[package]] name = "proc-macro2" version = "1.0.106" @@ -5529,7 +5736,7 @@ dependencies = [ "quinn-udp", "rustc-hash", "rustls 0.23.40", - "socket2", + "socket2 0.6.3", "thiserror 2.0.18", "tokio", "tracing", @@ -5566,7 +5773,7 @@ dependencies = [ "cfg_aliases", "libc", "once_cell", - "socket2", + "socket2 0.6.3", "tracing", "windows-sys 0.60.2", ] @@ -5718,7 +5925,7 @@ dependencies = [ "rustls-native-certs 0.8.3", "ryu", "sha1_smol", - "socket2", + "socket2 0.6.3", "tokio", "tokio-rustls 0.26.4", "tokio-util", @@ -6042,7 +6249,7 @@ dependencies = [ "crypto-bigint", "ctr 0.10.0", "ctr 0.9.2", - "curve25519-dalek", + "curve25519-dalek 5.0.0-pre.6", "data-encoding", "delegate", "der 0.8.0", @@ -6261,7 +6468,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a9aace74cb666635c918e9c12bc0d348266037aa8eb599b5cba565709a8dff00" dependencies = [ "openssl-probe 0.1.6", - "rustls-pemfile", + "rustls-pemfile 1.0.4", "schannel", "security-framework 2.11.1", ] @@ -6287,6 +6494,15 @@ dependencies = [ "base64 0.21.7", ] +[[package]] +name = "rustls-pemfile" +version = "2.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dce314e5fee3f39953d46bb63bb8a46d40c2f8fb7cc5a3b6cab2bde9721d6e50" +dependencies = [ + "rustls-pki-types", +] + [[package]] name = "rustls-pki-types" version = "1.14.1" @@ -6377,6 +6593,12 @@ dependencies = [ "winapi-util", ] +[[package]] +name = "saturating" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ece8e78b2f38ec51c51f5d475df0a7187ba5111b2a28bdc761ee05b075d40a71" + [[package]] name = "schannel" version = "0.1.29" @@ -6704,7 +6926,7 @@ version = "3.19.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cf2ebbe86054f9b45bc3881e865683ccfaccce97b9b4cb53f3039d67f355a334" dependencies = [ - "darling", + "darling 0.23.0", "proc-macro2", "quote", "syn 2.0.117", @@ -6931,6 +7153,16 @@ dependencies = [ "serde", ] +[[package]] +name = "socket2" +version = "0.5.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e22376abed350d73dd1cd119b57ffccad95b4e585a7cda43e286245ce23c0678" +dependencies = [ + "libc", + "windows-sys 0.52.0", +] + [[package]] name = "socket2" version = "0.6.3" @@ -7950,6 +8182,15 @@ dependencies = [ "utf-8", ] +[[package]] +name = "termcolor" +version = "1.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "06794f8f6c5c898b3275aebefa6b8a1cb24cd2c6c79397ab15774837a0bc5755" +dependencies = [ + "winapi-util", +] + [[package]] name = "thiserror" version = "1.0.69" @@ -8020,7 +8261,7 @@ dependencies = [ "pretty-hex", "rust_decimal", "rustls-native-certs 0.6.3", - "rustls-pemfile", + "rustls-pemfile 1.0.4", "thiserror 1.0.69", "tokio", "tokio-rustls 0.24.1", @@ -8108,7 +8349,7 @@ dependencies = [ "parking_lot", "pin-project-lite", "signal-hook-registry", - "socket2", + "socket2 0.6.3", "tokio-macros", "windows-sys 0.61.2", ] @@ -8446,6 +8687,12 @@ version = "0.2.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" +[[package]] +name = "twox-hash" +version = "2.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9ea3136b675547379c4bd395ca6b938e5ad3c3d20fad76e7fe85f9e0d011419c" + [[package]] name = "typed-builder" version = "0.22.0" diff --git a/crates/dbx-core/Cargo.toml b/crates/dbx-core/Cargo.toml index 8ad537a24d..43e1373e32 100644 --- a/crates/dbx-core/Cargo.toml +++ b/crates/dbx-core/Cargo.toml @@ -19,7 +19,8 @@ chrono = { version = "0.4", features = ["serde"] } rust_decimal = { version = "1", features = ["serde"] } anyhow = "1" uuid = { version = "1", features = ["v4", "serde"] } -sqlx = { version = "0.8", features = ["runtime-tokio", "tls-rustls", "mysql", "postgres", "sqlite", "json", "chrono", "uuid", "rust_decimal"] } +sqlx = { version = "0.8", features = ["runtime-tokio", "tls-rustls", "postgres", "sqlite", "json", "chrono", "uuid", "rust_decimal"] } +mysql_async = { version = "0.36", default-features = false, features = ["default-rustls", "client_ed25519", "chrono", "rust_decimal"] } sqlparser = "0.62.0" redis = { version = "0.32.2", features = ["tokio-comp", "tls-rustls", "tokio-rustls-comp"] } rustls = { version = "0.23", features = ["aws-lc-rs"] } diff --git a/crates/dbx-core/src/connection.rs b/crates/dbx-core/src/connection.rs index e617d01d45..0b61aa0b50 100644 --- a/crates/dbx-core/src/connection.rs +++ b/crates/dbx-core/src/connection.rs @@ -4,6 +4,9 @@ use std::path::PathBuf; use std::sync::Arc; use tokio::sync::RwLock; +use mysql_async::prelude::Queryable; +use mysql_async::Row as MysqlRow; + use crate::database_capabilities; use crate::db; use crate::db::agent_driver::AgentMethod; @@ -37,7 +40,7 @@ pub enum MysqlMode { } pub enum PoolKind { - Mysql(sqlx::mysql::MySqlPool, MysqlMode), + Mysql(db::mysql::MySqlPool, MysqlMode), Postgres(sqlx::postgres::PgPool), Sqlite(sqlx::sqlite::SqlitePool), Redis(tokio::sync::Mutex), @@ -715,17 +718,33 @@ fn is_original_hostname_endpoint(config: &ConnectionConfig, host: &str, port: u1 host == config.host && port == config.port && host.parse::().is_err() } -async fn detect_ob_oracle_mode(config: &ConnectionConfig, pool: &sqlx::mysql::MySqlPool) -> MysqlMode { +async fn detect_ob_oracle_mode(config: &ConnectionConfig, pool: &db::mysql::MySqlPool) -> MysqlMode { let profile = config.driver_profile.as_deref().unwrap_or("").to_lowercase(); if !profile.contains("oceanbase") { return MysqlMode::Normal; } - match sqlx::query_as::<_, (String, String)>("SHOW VARIABLES LIKE 'ob_compatibility_mode'") - .fetch_optional(pool) - .await - { - Ok(Some((_, val))) if val.to_lowercase() == "oracle" => MysqlMode::OceanBaseOracle, - _ => MysqlMode::Normal, + let mut conn = match pool.get_conn().await { + Ok(c) => c, + Err(_) => return MysqlMode::Normal, + }; + let result = conn.query_iter("SHOW VARIABLES LIKE 'ob_compatibility_mode'").await; + let rows: Vec = match result { + Ok(r) => match r.collect_and_drop().await { + Ok(rows) => rows, + Err(_) => return MysqlMode::Normal, + }, + Err(_) => return MysqlMode::Normal, + }; + match rows.first() { + Some(row) => { + let val: String = row.get(1).unwrap_or_default(); + if val.to_lowercase() == "oracle" { + MysqlMode::OceanBaseOracle + } else { + MysqlMode::Normal + } + } + None => MysqlMode::Normal, } } diff --git a/crates/dbx-core/src/db/mysql.rs b/crates/dbx-core/src/db/mysql.rs index 13dc9dda0c..2a82ae953d 100644 --- a/crates/dbx-core/src/db/mysql.rs +++ b/crates/dbx-core/src/db/mysql.rs @@ -1,41 +1,47 @@ -use chrono::{DateTime, NaiveDate, NaiveDateTime, NaiveTime, Utc}; +use chrono::{NaiveDate, NaiveDateTime, NaiveTime}; use futures::StreamExt; +use mysql_async::consts::ColumnType; +use mysql_async::prelude::*; use rust_decimal::Decimal; -use sqlx::mysql::{MySqlPool, MySqlPoolOptions, MySqlRow}; -use sqlx::{Column, Executor, Row, TypeInfo, ValueRef}; +use std::borrow::Cow; use std::collections::HashSet; -use std::str::FromStr; -use std::time::{Duration, Instant}; +use std::time::Duration; +use std::time::Instant; use crate::sql::starts_with_executable_sql_keyword; use crate::types::{ ColumnInfo, DatabaseInfo, ForeignKeyInfo, IndexInfo, ObjectInfo, QueryResult, TableInfo, TriggerInfo, }; +pub type MySqlPool = mysql_async::Pool; + fn quote_value(s: &str) -> String { format!("'{}'", s.replace('\\', "\\\\").replace('\'', "\\'")) } -fn get_str(row: &MySqlRow, idx: usize) -> String { - row.try_get::(idx) - .or_else(|_| row.try_get::, _>(idx).map(|b| String::from_utf8_lossy(&b).to_string())) +fn row_get(row: &mysql_async::Row, index: I) -> Option +where + T: mysql_async::prelude::FromValue, + I: mysql_async::prelude::ColumnIndex, +{ + row.get_opt::(index).and_then(|result| result.ok()) +} + +fn get_str(row: &mysql_async::Row, idx: usize) -> String { + row_get::(row, idx) + .or_else(|| row_get::, _>(row, idx).map(|b| String::from_utf8_lossy(&b).to_string())) .unwrap_or_default() } -fn get_str_by_name(row: &MySqlRow, name: &str) -> String { - row.try_get::(name) - .or_else(|_| row.try_get::, _>(name).map(|b| String::from_utf8_lossy(&b).to_string())) +fn get_str_by_name(row: &mysql_async::Row, name: &str) -> String { + row_get::(row, name) + .or_else(|| row_get::, _>(row, name).map(|b| String::from_utf8_lossy(&b).to_string())) .unwrap_or_default() } -fn get_opt_str(row: &MySqlRow, name: &str) -> Option { - row.try_get::, _>(name) - .ok() - .flatten() - .or_else(|| row.try_get::, _>(name).ok().flatten().map(|d| d.to_string())) - .or_else(|| { - row.try_get::>, _>(name).ok().flatten().map(|b| String::from_utf8_lossy(&b).to_string()) - }) +fn get_opt_str(row: &mysql_async::Row, name: &str) -> Option { + row_get::(row, name) + .or_else(|| row_get::, _>(row, name).map(|b| String::from_utf8_lossy(&b).to_string())) } fn numeric_metadata_u64_to_i32(value: Option) -> Option { @@ -50,173 +56,149 @@ fn numeric_metadata_str_to_i32(value: Option) -> Option { value.and_then(|v| v.parse::().ok()).and_then(|v| i32::try_from(v).ok()) } -fn get_opt_i32(row: &MySqlRow, name: &str) -> Option { - if row.try_get_raw(name).map(|v| v.is_null()).unwrap_or(true) { - return None; - } - - row.try_get::, _>(name) - .ok() - .flatten() - .or_else(|| numeric_metadata_i64_to_i32(row.try_get::, _>(name).ok().flatten())) - .or_else(|| numeric_metadata_u64_to_i32(row.try_get::, _>(name).ok().flatten())) - .or_else(|| numeric_metadata_str_to_i32(row.try_get::, _>(name).ok().flatten())) +fn get_opt_i32(row: &mysql_async::Row, name: &str) -> Option { + row_get::(row, name) + .or_else(|| numeric_metadata_i64_to_i32(row_get::(row, name))) + .or_else(|| numeric_metadata_u64_to_i32(row_get::(row, name))) + .or_else(|| numeric_metadata_str_to_i32(row_get::(row, name))) .or_else(|| { - row.try_get::>, _>(name) - .ok() - .flatten() + row_get::, _>(row, name) .and_then(|b| String::from_utf8(b).ok()) .and_then(|v| numeric_metadata_str_to_i32(Some(v))) }) } -fn mysql_temporal_to_json_value(row: &MySqlRow, idx: usize) -> Option { - if let Ok(v) = row.try_get::(idx) { - return Some(serde_json::Value::String(v.to_string())); - } - if let Ok(v) = row.try_get::, _>(idx) { - return Some(serde_json::Value::String(mysql_datetime_to_string(v))); - } - if let Ok(v) = row.try_get::(idx) { - return Some(serde_json::Value::String(v.to_string())); - } - if let Ok(v) = row.try_get::(idx) { - return Some(serde_json::Value::String(v.to_string())); - } - None -} - -fn mysql_datetime_to_string(value: DateTime) -> String { - value.naive_utc().to_string() +#[cfg(test)] +fn mysql_datetime_to_string(value: NaiveDateTime) -> String { + value.to_string() } +#[cfg(test)] fn is_mysql_lossless_integer_type(type_name: &str) -> bool { let upper_type = type_name.to_uppercase(); upper_type.contains("BIGINT") || upper_type.contains("LARGEINT") } -fn mysql_lossless_integer_to_json(row: &MySqlRow, idx: usize) -> serde_json::Value { - row.try_get::(idx) - .map(serde_json::Value::String) - .or_else(|_| row.try_get::(idx).map(|v: Decimal| serde_json::Value::String(v.to_string()))) - .or_else(|_| row.try_get::(idx).map(|v| serde_json::Value::String(v.to_string()))) - .or_else(|_| row.try_get::(idx).map(|v| serde_json::Value::String(v.to_string()))) - .or_else(|_| { - row.try_get::, _>(idx).map(|b| serde_json::Value::String(String::from_utf8_lossy(&b).to_string())) - }) - .or_else(|_| row.try_get_unchecked::(idx).map(serde_json::Value::String)) - .or_else(|_| { - row.try_get_unchecked::, _>(idx) - .map(|b| serde_json::Value::String(String::from_utf8_lossy(&b).to_string())) - }) - .unwrap_or(serde_json::Value::Null) +fn is_lossless_integer_column(column: &mysql_async::Column) -> bool { + matches!(column.column_type(), ColumnType::MYSQL_TYPE_LONGLONG | ColumnType::MYSQL_TYPE_NEWDECIMAL) } -fn mysql_value_to_json(row: &MySqlRow, idx: usize, type_name: &str) -> serde_json::Value { - if row.try_get_raw(idx).map(|v| v.is_null()).unwrap_or(true) { +fn mysql_value_to_json(row: &mysql_async::Row, idx: usize) -> serde_json::Value { + let Some(column) = row.columns_ref().get(idx) else { + return serde_json::Value::Null; + }; + + let Some(value) = row.as_ref(idx) else { + return serde_json::Value::Null; + }; + if matches!(value, mysql_async::Value::NULL) { return serde_json::Value::Null; } - let upper_type = type_name.to_uppercase(); - - if upper_type == "JSON" { - if let Ok(v) = row.try_get::(idx) { - return serde_json::Value::String(v.to_string()); + match column.column_type() { + ColumnType::MYSQL_TYPE_JSON => { + if let Some(v) = row_get::(row, idx) { + return serde_json::Value::String(v); + } } - if let Ok(v) = row.try_get::(idx) { - return serde_json::Value::String(v); + ColumnType::MYSQL_TYPE_DECIMAL | ColumnType::MYSQL_TYPE_NEWDECIMAL | ColumnType::MYSQL_TYPE_LONGLONG => { + if is_lossless_integer_column(column) { + return row + .get_opt::(idx) + .and_then(|result| result.ok()) + .map(serde_json::Value::String) + .or_else(|| { + row_get::(row, idx).map(|v: Decimal| serde_json::Value::String(v.to_string())) + }) + .or_else(|| row_get::(row, idx).map(|v| serde_json::Value::String(v.to_string()))) + .or_else(|| row_get::(row, idx).map(|v| serde_json::Value::String(v.to_string()))) + .or_else(|| { + row_get::, _>(row, idx) + .map(|b| serde_json::Value::String(String::from_utf8_lossy(&b).to_string())) + }) + .unwrap_or(serde_json::Value::Null); + } + return row + .get_opt::(idx) + .and_then(|result| result.ok()) + .map(|v: Decimal| serde_json::Value::String(v.to_string())) + .unwrap_or(serde_json::Value::Null); } - return serde_json::Value::Null; - } - - if upper_type == "BOOLEAN" { - // MySQL BOOLEAN is an alias for TINYINT(1); display as integer - return row - .try_get::(idx) - .map(|v| serde_json::Value::Number((v as i64).into())) - .or_else(|_| row.try_get::(idx).map(|v| serde_json::Value::Number((v as i64).into()))) - .unwrap_or(serde_json::Value::Null); - } - - if is_mysql_lossless_integer_type(&upper_type) { - return mysql_lossless_integer_to_json(row, idx); - } - - if upper_type == "DECIMAL" { - return row - .try_get::(idx) - .map(|v: Decimal| serde_json::Value::String(v.to_string())) - .unwrap_or(serde_json::Value::Null); - } - - if upper_type.starts_with("DATETIME") - || upper_type.starts_with("TIMESTAMP") - || upper_type == "DATE" - || upper_type == "TIME" - || upper_type.starts_with("TIME(") - { - if let Some(v) = mysql_temporal_to_json_value(row, idx) { - return v; + ColumnType::MYSQL_TYPE_TIMESTAMP + | ColumnType::MYSQL_TYPE_TIMESTAMP2 + | ColumnType::MYSQL_TYPE_DATETIME + | ColumnType::MYSQL_TYPE_DATETIME2 + | ColumnType::MYSQL_TYPE_DATE + | ColumnType::MYSQL_TYPE_TIME + | ColumnType::MYSQL_TYPE_TIME2 + | ColumnType::MYSQL_TYPE_NEWDATE => { + if let Some(v) = row_get::(row, idx) { + return serde_json::Value::String(v.to_string()); + } + if let Some(v) = row_get::(row, idx) { + return serde_json::Value::String(v.to_string()); + } + if let Some(v) = row_get::(row, idx) { + return serde_json::Value::String(v.to_string()); + } } + _ => {} } - row.try_get::(idx) + row_get::(row, idx) .map(serde_json::Value::String) - .or_else(|_| row.try_get::(idx).map(super::safe_i64_to_json)) - .or_else(|_| row.try_get::(idx).map(super::safe_u64_to_json)) - .or_else(|_| row.try_get::(idx).map(|v| serde_json::Value::Number(v.into()))) - .or_else(|_| row.try_get::(idx).map(|v| serde_json::Value::Number(v.into()))) - .or_else(|_| { - row.try_get::(idx).map(|v| { + .or_else(|| row_get::(row, idx).map(super::safe_i64_to_json)) + .or_else(|| row_get::(row, idx).map(super::safe_u64_to_json)) + .or_else(|| row_get::(row, idx).map(|v| serde_json::Value::Number(v.into()))) + .or_else(|| row_get::(row, idx).map(|v| serde_json::Value::Number(v.into()))) + .or_else(|| { + row_get::(row, idx).map(|v| { serde_json::Number::from_f64(v).map(serde_json::Value::Number).unwrap_or(serde_json::Value::Null) }) }) - .or_else(|_| row.try_get::(idx).map(serde_json::Value::Bool)) - .or_else(|_| { - row.try_get::, _>(idx).map(|b| serde_json::Value::String(String::from_utf8_lossy(&b).to_string())) - }) - .or_else(|e| mysql_temporal_to_json_value(row, idx).ok_or(e)) - .or_else(|_| row.try_get_unchecked::(idx).map(serde_json::Value::String)) - .or_else(|_| { - row.try_get_unchecked::, _>(idx) - .map(|b| serde_json::Value::String(String::from_utf8_lossy(&b).to_string())) + .or_else(|| row_get::(row, idx).map(serde_json::Value::Bool)) + .or_else(|| { + row_get::, _>(row, idx).map(|b| serde_json::Value::String(String::from_utf8_lossy(&b).to_string())) }) .unwrap_or(serde_json::Value::Null) } pub async fn connect(url: &str) -> Result { - let options = mysql_connect_options(url, true)?; - let result = super::with_connection_timeout("MySQL", async { - MySqlPoolOptions::new() - .max_connections(5) - .acquire_timeout(super::connection_timeout()) - .idle_timeout(Duration::from_secs(300)) - .connect_with(options) - .await - .map_err(|e| format!("MySQL connection failed: {e}")) - }) - .await; + let pool = create_pool(url)?; + let result = verify_pool_connection(&pool).await; if let Err(ref e) = result { if mysql_error_should_retry_without_ssl(e) { if let Some(fallback_url) = ssl_fallback_url(url) { - let fallback_options = mysql_connect_options(&fallback_url, true)?; log::info!("SSL handshake failed, retrying with ssl-mode=disabled"); - return super::with_connection_timeout("MySQL", async { - MySqlPoolOptions::new() - .max_connections(5) - .acquire_timeout(super::connection_timeout()) - .idle_timeout(Duration::from_secs(300)) - .connect_with(fallback_options) - .await - .map_err(|e| format!("MySQL connection failed: {e}")) - }) - .await; + let fallback_pool = create_pool(&fallback_url)?; + return match verify_pool_connection(&fallback_pool).await { + Ok(()) => Ok(fallback_pool), + Err(e) => Err(e), + }; } } } - result + result.map(|_| pool) +} + +fn create_pool(url: &str) -> Result { + let opts = mysql_async::Opts::from_url(&mysql_async_url(url)).map_err(|e| format!("Invalid MySQL URL: {e}"))?; + let pool_opts = mysql_async::PoolOpts::new() + .with_constraints(mysql_async::PoolConstraints::new(1, 5).unwrap()) + .with_inactive_connection_ttl(Duration::from_secs(300)); + let builder = mysql_async::OptsBuilder::from_opts(opts).stmt_cache_size(0).pool_opts(Some(pool_opts)); + Ok(MySqlPool::new(builder)) +} + +async fn verify_pool_connection(pool: &MySqlPool) -> Result<(), String> { + super::with_connection_timeout("MySQL", async { + let mut conn = pool.get_conn().await.map_err(|e| format!("MySQL connection failed: {e}"))?; + conn.ping().await.map_err(|e| format!("MySQL ping failed: {e}"))?; + Ok(()) + }) + .await } fn mysql_error_should_retry_without_ssl(error: &str) -> bool { @@ -245,52 +227,48 @@ fn ssl_fallback_url(url: &str) -> Option { } } -pub async fn connect_bare(url: &str) -> Result { - let options = mysql_connect_options(url, true)?; - super::with_connection_timeout("MySQL", async { - MySqlPoolOptions::new() - .max_connections(5) - .acquire_timeout(super::connection_timeout()) - .idle_timeout(Duration::from_secs(300)) - .connect_with(options) - .await - .map_err(|e| format!("MySQL connection failed: {e}")) - }) - .await -} - -fn mysql_connect_options( - url: &str, - preserve_server_timezone: bool, -) -> Result { - let mut options = sqlx::mysql::MySqlConnectOptions::from_str(url).map_err(|e| format!("Invalid MySQL URL: {e}"))?; - options = options.no_engine_substitution(false).set_names(false).pipes_as_concat(false); - if preserve_server_timezone && !mysql_url_has_timezone_param(url) { - options = options.timezone(None); - } - Ok(options) -} - -fn mysql_url_has_timezone_param(url: &str) -> bool { - let Some((_, query)) = url.split_once('?') else { - return false; +fn mysql_async_url(url: &str) -> Cow<'_, str> { + let Some((base, query)) = url.split_once('?') else { + return Cow::Borrowed(url); }; - query.split('&').filter(|segment| !segment.is_empty()).any(|segment| { - let key = segment.split('=').next().unwrap_or("").trim().to_ascii_lowercase(); - key == "timezone" || key == "time-zone" - }) + let filtered: Vec<&str> = query + .split('&') + .filter(|segment| { + let segment = segment.trim(); + !segment.is_empty() + && !segment.starts_with("ssl-mode=") + && !segment.starts_with("charset=") + && !segment.starts_with("time_zone=") + && !segment.starts_with("time-zone=") + }) + .collect(); + + if filtered.len() == query.split('&').filter(|segment| !segment.trim().is_empty()).count() { + Cow::Borrowed(url) + } else if filtered.is_empty() { + Cow::Owned(base.to_string()) + } else { + Cow::Owned(format!("{base}?{}", filtered.join("&"))) + } +} + +pub async fn connect_bare(url: &str) -> Result { + let pool = create_pool(url)?; + verify_pool_connection(&pool).await.map(|_| pool) } pub async fn list_databases(pool: &MySqlPool) -> Result, String> { - let rows: Vec = sqlx::raw_sql( - "SELECT SCHEMA_NAME FROM information_schema.SCHEMATA \ - WHERE SCHEMA_NAME NOT IN ('information_schema', 'mysql', 'performance_schema', 'sys') \ - ORDER BY SCHEMA_NAME", - ) - .fetch_all(pool) - .await - .map_err(|e| e.to_string())?; + let mut conn = pool.get_conn().await.map_err(|e| e.to_string())?; + let result = conn + .query_iter( + "SELECT SCHEMA_NAME FROM information_schema.SCHEMATA \ + WHERE SCHEMA_NAME NOT IN ('information_schema', 'mysql', 'performance_schema', 'sys') \ + ORDER BY SCHEMA_NAME", + ) + .await + .map_err(|e| e.to_string())?; + let rows: Vec = result.collect_and_drop().await.map_err(|e| e.to_string())?; Ok(rows.iter().map(|row| DatabaseInfo { name: get_str(row, 0) }).collect()) } @@ -300,14 +278,16 @@ pub async fn list_tables(pool: &MySqlPool, database: &str) -> Result = sqlx::raw_sql(&sql).fetch_all(pool).await.map_err(|e| e.to_string())?; + let mut conn = pool.get_conn().await.map_err(|e| e.to_string())?; + let result = conn.query_iter(&sql).await.map_err(|e| e.to_string())?; + let rows: Vec = result.collect_and_drop().await.map_err(|e| e.to_string())?; Ok(rows .iter() .map(|row| TableInfo { name: get_str_by_name(row, "TABLE_NAME"), table_type: get_str_by_name(row, "TABLE_TYPE"), - comment: row.try_get::("TABLE_COMMENT").ok().filter(|s| !s.is_empty()), + comment: get_opt_str(row, "TABLE_COMMENT").filter(|s| !s.is_empty()), }) .collect()) } @@ -335,7 +315,9 @@ fn list_objects_sql(database: &str) -> String { pub async fn list_objects(pool: &MySqlPool, database: &str) -> Result, String> { let sql = list_objects_sql(database); - let rows: Vec = sqlx::raw_sql(&sql).fetch_all(pool).await.map_err(|e| e.to_string())?; + let mut conn = pool.get_conn().await.map_err(|e| e.to_string())?; + let result = conn.query_iter(&sql).await.map_err(|e| e.to_string())?; + let rows: Vec = result.collect_and_drop().await.map_err(|e| e.to_string())?; Ok(rows .iter() @@ -379,11 +361,16 @@ fn is_primary_key_column(primary_key_columns: &HashSet, name: &str, colu pub async fn get_columns(pool: &MySqlPool, database: &str, table: &str) -> Result, String> { let pk_sql = primary_key_columns_sql(database, table); - let pk_rows: Vec = sqlx::raw_sql(&pk_sql).fetch_all(pool).await.map_err(|e| e.to_string())?; + let mut conn = pool.get_conn().await.map_err(|e| e.to_string())?; + let result = conn.query_iter(&pk_sql).await.map_err(|e| e.to_string())?; + let pk_rows: Vec = result.collect_and_drop().await.map_err(|e| e.to_string())?; let primary_key_columns: HashSet = pk_rows.iter().map(|row| get_str_by_name(row, "COLUMN_NAME")).collect(); + drop(conn); let sql = columns_sql(database, table); - let rows: Vec = sqlx::raw_sql(&sql).fetch_all(pool).await.map_err(|e| e.to_string())?; + let mut conn = pool.get_conn().await.map_err(|e| e.to_string())?; + let result = conn.query_iter(&sql).await.map_err(|e| e.to_string())?; + let rows: Vec = result.collect_and_drop().await.map_err(|e| e.to_string())?; Ok(rows .iter() @@ -416,22 +403,21 @@ async fn execute_result_set_with_text_protocol( row_limit: usize, start: Instant, ) -> Result { - let mut stream = sqlx::raw_sql(sql).fetch(&*pool); - let mut columns: Vec = vec![]; - let mut column_types: Vec = vec![]; + let mut conn = pool.get_conn().await.map_err(|e| e.to_string())?; + let mut result = conn.query_iter(sql).await.map_err(|e| e.to_string())?; + let columns: Vec = result.columns_ref().iter().map(|c| c.name_str().to_string()).collect(); + let mut result_rows: Vec> = Vec::new(); + let mut stream = result + .stream::() + .await + .map_err(|e| e.to_string())? + .ok_or_else(|| "Empty result set stream".to_string())?; while let Some(row) = stream.next().await { - let row: MySqlRow = row.map_err(|e| e.to_string())?; - if columns.is_empty() { - columns = row.columns().iter().map(|c| c.name().to_string()).collect(); - column_types = row.columns().iter().map(|c| c.type_info().name().to_string()).collect(); - } - result_rows.push( - (0..row.len()) - .map(|i| mysql_value_to_json(&row, i, column_types.get(i).map(String::as_str).unwrap_or(""))) - .collect(), - ); + let row = row.map_err(|e| e.to_string())?; + let values: Vec = (0..row.len()).map(|i| mysql_value_to_json(&row, i)).collect(); + result_rows.push(values); if result_rows.len() > row_limit { break; } @@ -459,20 +445,21 @@ async fn execute_result_set_with_prepared_protocol( row_limit: usize, start: Instant, ) -> Result { - let desc = pool.describe(sql).await.map_err(|e| e.to_string())?; - let columns: Vec = desc.columns().iter().map(|c| c.name().to_string()).collect(); - let column_types: Vec = desc.columns().iter().map(|c| c.type_info().name().to_string()).collect(); + let mut conn = pool.get_conn().await.map_err(|e| e.to_string())?; + let mut result = conn.exec_iter(sql, ()).await.map_err(|e| e.to_string())?; + let columns: Vec = result.columns_ref().iter().map(|c| c.name_str().to_string()).collect(); - let mut stream = sqlx::query(sql).fetch(&*pool); let mut result_rows: Vec> = Vec::new(); + let mut stream = result + .stream::() + .await + .map_err(|e| e.to_string())? + .ok_or_else(|| "Empty result set stream".to_string())?; while let Some(row) = stream.next().await { let row = row.map_err(|e| e.to_string())?; - result_rows.push( - (0..row.len()) - .map(|i| mysql_value_to_json(&row, i, column_types.get(i).map(String::as_str).unwrap_or(""))) - .collect(), - ); + let values: Vec = (0..row.len()).map(|i| mysql_value_to_json(&row, i)).collect(); + result_rows.push(values); if result_rows.len() > row_limit { break; } @@ -520,12 +507,15 @@ pub async fn execute_query_with_max_rows( } } } else { - let result = sqlx::raw_sql(sql).execute(pool).await.map_err(|e| e.to_string())?; + let mut conn = pool.get_conn().await.map_err(|e| e.to_string())?; + let result = conn.query_iter(sql).await.map_err(|e| e.to_string())?; + let affected_rows = result.affected_rows(); + result.drop_result().await.map_err(|e| e.to_string())?; Ok(QueryResult { columns: vec![], rows: vec![], - affected_rows: result.rows_affected(), + affected_rows, execution_time_ms: start.elapsed().as_millis(), truncated: false, session_id: None, @@ -570,7 +560,9 @@ pub async fn list_indexes(pool: &MySqlPool, database: &str, table: &str) -> Resu quote_value(database), quote_value(table), ); - let rows: Vec = sqlx::raw_sql(&sql).fetch_all(pool).await.map_err(|e| e.to_string())?; + let mut conn = pool.get_conn().await.map_err(|e| e.to_string())?; + let result = conn.query_iter(&sql).await.map_err(|e| e.to_string())?; + let rows: Vec = result.collect_and_drop().await.map_err(|e| e.to_string())?; Ok(rows .iter() @@ -579,8 +571,8 @@ pub async fn list_indexes(pool: &MySqlPool, database: &str, table: &str) -> Resu IndexInfo { name: get_str_by_name(row, "INDEX_NAME"), columns: cols_str.split(',').filter(|s| !s.is_empty()).map(|s| s.to_string()).collect(), - is_unique: row.get::("is_unique"), - is_primary: row.get::("is_primary"), + is_unique: row.get::("is_unique").unwrap_or(false), + is_primary: row.get::("is_primary").unwrap_or(false), filter: None, index_type: Some(get_str_by_name(row, "INDEX_TYPE")), included_columns: None, @@ -601,7 +593,9 @@ pub async fn list_foreign_keys(pool: &MySqlPool, database: &str, table: &str) -> quote_value(database), quote_value(table), ); - let rows: Vec = sqlx::raw_sql(&sql).fetch_all(pool).await.map_err(|e| e.to_string())?; + let mut conn = pool.get_conn().await.map_err(|e| e.to_string())?; + let result = conn.query_iter(&sql).await.map_err(|e| e.to_string())?; + let rows: Vec = result.collect_and_drop().await.map_err(|e| e.to_string())?; Ok(rows .iter() @@ -623,7 +617,9 @@ pub async fn list_triggers(pool: &MySqlPool, database: &str, table: &str) -> Res quote_value(database), quote_value(table), ); - let rows: Vec = sqlx::raw_sql(&sql).fetch_all(pool).await.map_err(|e| e.to_string())?; + let mut conn = pool.get_conn().await.map_err(|e| e.to_string())?; + let result = conn.query_iter(&sql).await.map_err(|e| e.to_string())?; + let rows: Vec = result.collect_and_drop().await.map_err(|e| e.to_string())?; Ok(rows .iter() @@ -642,7 +638,6 @@ mod tests { #[test] fn mysql_with_queries_are_treated_as_result_sets() { let sql = "WITH RECURSIVE org_tree AS (SELECT 1 AS id) SELECT id FROM org_tree"; - assert!(is_result_set_query(sql)); } @@ -736,36 +731,24 @@ mod tests { assert!(mysql_error_should_retry_with_text_protocol(error)); } - #[test] - fn mysql_connect_options_preserve_server_timezone_by_default() { - let options = mysql_connect_options("mysql://root:secret@127.0.0.1:3306/app", true).expect("parse mysql url"); - let debug = format!("{options:?}"); - - assert!(debug.contains("timezone: None"), "{debug}"); - assert!(debug.contains("set_names: false"), "{debug}"); - assert!(debug.contains("no_engine_substitution: false"), "{debug}"); - } - - #[test] - fn mysql_connect_options_keep_explicit_timezone_param() { - let options = mysql_connect_options("mysql://root:secret@127.0.0.1:3306/app?timezone=%2B08:00", true) - .expect("parse mysql url"); - let debug = format!("{options:?}"); - - assert!(debug.contains("timezone: Some(\"+08:00\")"), "{debug}"); - } - - #[test] - fn mysql_timezone_param_detection_matches_supported_aliases() { - assert!(mysql_url_has_timezone_param("mysql://root@localhost/app?timezone=%2B08:00")); - assert!(mysql_url_has_timezone_param("mysql://root@localhost/app?charset=utf8mb4&time-zone=%2B08:00")); - assert!(!mysql_url_has_timezone_param("mysql://root@localhost/app?charset=utf8mb4")); - } - #[test] fn mysql_datetime_utc_values_display_without_rfc3339_offset() { - let value = DateTime::from_timestamp(1_778_544_000, 0).expect("valid timestamp"); + let value = NaiveDateTime::new( + NaiveDate::from_ymd_opt(2026, 5, 12).unwrap(), + NaiveTime::from_hms_opt(0, 0, 0).unwrap(), + ); assert_eq!(mysql_datetime_to_string(value), "2026-05-12 00:00:00"); } + + #[tokio::test] + #[ignore = "requires remote MariaDB with ed25519 user"] + async fn test_ed25519_auth() { + let url = "mysql://edtest:test123@172.26.128.159:20026/testdb"; + let pool = super::connect(url).await.expect("connect with ed25519"); + let mut conn = pool.get_conn().await.expect("get connection"); + conn.ping().await.expect("ping"); + let _ = conn.disconnect().await; + let _ = pool.disconnect().await; + } } diff --git a/crates/dbx-core/src/db/ob_oracle.rs b/crates/dbx-core/src/db/ob_oracle.rs index 77971989ee..c8ff550a17 100644 --- a/crates/dbx-core/src/db/ob_oracle.rs +++ b/crates/dbx-core/src/db/ob_oracle.rs @@ -1,45 +1,50 @@ -use sqlx::mysql::{MySqlPool, MySqlRow}; -use sqlx::Row; +use mysql_async::prelude::*; use crate::types::{ColumnInfo, DatabaseInfo, ForeignKeyInfo, IndexInfo, ObjectInfo, TableInfo, TriggerInfo}; +fn row_get(row: &mysql_async::Row, index: I) -> Option +where + T: mysql_async::prelude::FromValue, + I: mysql_async::prelude::ColumnIndex, +{ + row.get_opt::(index).and_then(|result| result.ok()) +} + fn quote_value(s: &str) -> String { format!("'{}'", s.replace('\\', "\\\\").replace('\'', "\\'")) } -fn get_str(row: &MySqlRow, idx: usize) -> String { - row.try_get::(idx) - .or_else(|_| row.try_get::, _>(idx).map(|b| String::from_utf8_lossy(&b).to_string())) +fn get_str(row: &mysql_async::Row, idx: usize) -> String { + row_get::(row, idx) + .or_else(|| row_get::, _>(row, idx).map(|b| String::from_utf8_lossy(&b).to_string())) .unwrap_or_default() } -fn get_opt_str(row: &MySqlRow, idx: usize) -> Option { - row.try_get::, _>(idx).ok().flatten().or_else(|| { - row.try_get::>, _>(idx).ok().flatten().map(|b| String::from_utf8_lossy(&b).to_string()) - }) +fn get_opt_str(row: &mysql_async::Row, idx: usize) -> Option { + row_get::(row, idx) + .or_else(|| row_get::, _>(row, idx).map(|b| String::from_utf8_lossy(&b).to_string())) } -fn get_opt_i32(row: &MySqlRow, idx: usize) -> Option { - row.try_get::, _>(idx) - .ok() - .flatten() - .or_else(|| row.try_get::, _>(idx).ok().flatten().and_then(|v| i32::try_from(v).ok())) +fn get_opt_i32(row: &mysql_async::Row, idx: usize) -> Option { + row_get::(row, idx).or_else(|| row_get::(row, idx).and_then(|v| i32::try_from(v).ok())) } -pub async fn list_databases(pool: &MySqlPool) -> Result, String> { - let rows: Vec = sqlx::raw_sql( - "SELECT USERNAME FROM ALL_USERS \ - WHERE USERNAME NOT IN ('SYS','LBACSYS','ORAAUDITOR','__public') \ - ORDER BY USERNAME", - ) - .fetch_all(pool) - .await - .map_err(|e| e.to_string())?; +pub async fn list_databases(pool: &mysql_async::Pool) -> Result, String> { + let mut conn = pool.get_conn().await.map_err(|e| e.to_string())?; + let result = conn + .query_iter( + "SELECT USERNAME FROM ALL_USERS \ + WHERE USERNAME NOT IN ('SYS','LBACSYS','ORAAUDITOR','__public') \ + ORDER BY USERNAME", + ) + .await + .map_err(|e| e.to_string())?; + let rows: Vec = result.collect_and_drop().await.map_err(|e| e.to_string())?; Ok(rows.iter().map(|row| DatabaseInfo { name: get_str(row, 0) }).collect()) } -pub async fn list_tables(pool: &MySqlPool, schema: &str) -> Result, String> { +pub async fn list_tables(pool: &mysql_async::Pool, schema: &str) -> Result, String> { let sql = format!( "SELECT TABLE_NAME, 'TABLE' AS TABLE_TYPE FROM ALL_TABLES WHERE OWNER = {s} \ UNION ALL \ @@ -47,7 +52,9 @@ pub async fn list_tables(pool: &MySqlPool, schema: &str) -> Result = sqlx::raw_sql(&sql).fetch_all(pool).await.map_err(|e| e.to_string())?; + let mut conn = pool.get_conn().await.map_err(|e| e.to_string())?; + let result = conn.query_iter(&sql).await.map_err(|e| e.to_string())?; + let rows: Vec = result.collect_and_drop().await.map_err(|e| e.to_string())?; Ok(rows.iter().map(|row| TableInfo { name: get_str(row, 0), table_type: get_str(row, 1), comment: None }).collect()) } @@ -68,9 +75,11 @@ fn list_objects_sql(schema: &str) -> String { ) } -pub async fn list_objects(pool: &MySqlPool, schema: &str) -> Result, String> { +pub async fn list_objects(pool: &mysql_async::Pool, schema: &str) -> Result, String> { let sql = list_objects_sql(schema); - let rows: Vec = sqlx::raw_sql(&sql).fetch_all(pool).await.map_err(|e| e.to_string())?; + let mut conn = pool.get_conn().await.map_err(|e| e.to_string())?; + let result = conn.query_iter(&sql).await.map_err(|e| e.to_string())?; + let rows: Vec = result.collect_and_drop().await.map_err(|e| e.to_string())?; Ok(rows .iter() @@ -85,7 +94,7 @@ pub async fn list_objects(pool: &MySqlPool, schema: &str) -> Result Result, String> { +pub async fn get_columns(pool: &mysql_async::Pool, schema: &str, table: &str) -> Result, String> { let sql = format!( "SELECT c.COLUMN_NAME, c.DATA_TYPE, c.NULLABLE, c.DATA_DEFAULT, \ c.DATA_LENGTH, c.DATA_PRECISION, c.DATA_SCALE, c.COLUMN_ID, \ @@ -102,7 +111,9 @@ pub async fn get_columns(pool: &MySqlPool, schema: &str, table: &str) -> Result< s = quote_value(schema), t = quote_value(table), ); - let rows: Vec = sqlx::raw_sql(&sql).fetch_all(pool).await.map_err(|e| e.to_string())?; + let mut conn = pool.get_conn().await.map_err(|e| e.to_string())?; + let result = conn.query_iter(&sql).await.map_err(|e| e.to_string())?; + let rows: Vec = result.collect_and_drop().await.map_err(|e| e.to_string())?; Ok(rows .iter() @@ -117,7 +128,7 @@ pub async fn get_columns(pool: &MySqlPool, schema: &str, table: &str) -> Result< data_type: display_type, is_nullable: get_str(row, 2) == "Y", column_default: get_opt_str(row, 3).map(|s| s.trim().to_string()).filter(|s| !s.is_empty()), - is_primary_key: row.try_get::(8).unwrap_or(0) == 1, + is_primary_key: row_get::(row, 8).unwrap_or(0) == 1, extra: None, comment: None, numeric_precision: precision, @@ -143,7 +154,7 @@ fn format_oracle_type(data_type: &str, precision: Option, scale: Option Result, String> { +pub async fn list_indexes(pool: &mysql_async::Pool, schema: &str, table: &str) -> Result, String> { let sql = format!( "SELECT ai.INDEX_NAME, \ LISTAGG(aic.COLUMN_NAME, ',') WITHIN GROUP (ORDER BY aic.COLUMN_POSITION) AS COLUMNS, \ @@ -158,7 +169,9 @@ pub async fn list_indexes(pool: &MySqlPool, schema: &str, table: &str) -> Result s = quote_value(schema), t = quote_value(table), ); - let rows: Vec = sqlx::raw_sql(&sql).fetch_all(pool).await.map_err(|e| e.to_string())?; + let mut conn = pool.get_conn().await.map_err(|e| e.to_string())?; + let result = conn.query_iter(&sql).await.map_err(|e| e.to_string())?; + let rows: Vec = result.collect_and_drop().await.map_err(|e| e.to_string())?; Ok(rows .iter() @@ -168,7 +181,7 @@ pub async fn list_indexes(pool: &MySqlPool, schema: &str, table: &str) -> Result name: get_str(row, 0), columns: cols_str.split(',').filter(|s| !s.is_empty()).map(|s| s.to_string()).collect(), is_unique: get_str(row, 2) == "UNIQUE", - is_primary: row.try_get::(3).unwrap_or(0) == 1, + is_primary: row_get::(row, 3).unwrap_or(0) == 1, filter: None, index_type: None, included_columns: None, @@ -178,7 +191,11 @@ pub async fn list_indexes(pool: &MySqlPool, schema: &str, table: &str) -> Result .collect()) } -pub async fn list_foreign_keys(pool: &MySqlPool, schema: &str, table: &str) -> Result, String> { +pub async fn list_foreign_keys( + pool: &mysql_async::Pool, + schema: &str, + table: &str, +) -> Result, String> { let sql = format!( "SELECT ac.CONSTRAINT_NAME, acc.COLUMN_NAME, \ ac2.TABLE_NAME AS R_TABLE, acc2.COLUMN_NAME AS R_COLUMN \ @@ -192,7 +209,9 @@ pub async fn list_foreign_keys(pool: &MySqlPool, schema: &str, table: &str) -> R s = quote_value(schema), t = quote_value(table), ); - let rows: Vec = sqlx::raw_sql(&sql).fetch_all(pool).await.map_err(|e| e.to_string())?; + let mut conn = pool.get_conn().await.map_err(|e| e.to_string())?; + let result = conn.query_iter(&sql).await.map_err(|e| e.to_string())?; + let rows: Vec = result.collect_and_drop().await.map_err(|e| e.to_string())?; Ok(rows .iter() @@ -205,7 +224,7 @@ pub async fn list_foreign_keys(pool: &MySqlPool, schema: &str, table: &str) -> R .collect()) } -pub async fn list_triggers(pool: &MySqlPool, schema: &str, table: &str) -> Result, String> { +pub async fn list_triggers(pool: &mysql_async::Pool, schema: &str, table: &str) -> Result, String> { let sql = format!( "SELECT TRIGGER_NAME, TRIGGERING_EVENT, TRIGGER_TYPE \ FROM ALL_TRIGGERS \ @@ -214,7 +233,9 @@ pub async fn list_triggers(pool: &MySqlPool, schema: &str, table: &str) -> Resul s = quote_value(schema), t = quote_value(table), ); - let rows: Vec = sqlx::raw_sql(&sql).fetch_all(pool).await.map_err(|e| e.to_string())?; + let mut conn = pool.get_conn().await.map_err(|e| e.to_string())?; + let result = conn.query_iter(&sql).await.map_err(|e| e.to_string())?; + let rows: Vec = result.collect_and_drop().await.map_err(|e| e.to_string())?; Ok(rows .iter() diff --git a/crates/dbx-core/src/query.rs b/crates/dbx-core/src/query.rs index 638084dd2a..1c0b7c429f 100644 --- a/crates/dbx-core/src/query.rs +++ b/crates/dbx-core/src/query.rs @@ -1,5 +1,6 @@ use chrono::{DateTime, Duration as ChronoDuration, NaiveDate, NaiveDateTime, NaiveTime, Utc}; use duckdb::types::{TimeUnit, ValueRef}; +use mysql_async::prelude::Queryable; use std::future::Future; use std::time::Duration; use tokio::time::timeout; @@ -904,7 +905,7 @@ pub async fn execute_statements_in_transaction( /// Owned pool variants for safe dispatch across async boundaries. enum TxPath { Pg(sqlx::postgres::PgPool), - Mysql(sqlx::mysql::MySqlPool, bool), + Mysql(mysql_async::Pool, bool), Sqlite(sqlx::sqlite::SqlitePool), Explicit, None, @@ -950,43 +951,23 @@ async fn exec_tx_pg_inner( } async fn exec_tx_mysql_inner( - pool: sqlx::mysql::MySqlPool, + pool: mysql_async::Pool, statements: &[String], start: std::time::Instant, ) -> Result { - let statements = statements.to_vec(); - tokio::task::spawn_blocking(move || { - tokio::runtime::Builder::new_current_thread() - .enable_all() - .build() - .map_err(|e| format!("Failed to start MySQL transaction runtime: {}", e))? - .block_on(exec_tx_mysql_raw_inner(pool, statements, start)) - }) - .await - .map_err(|e| format!("MySQL transaction task failed: {}", e))? -} - -async fn exec_tx_mysql_raw_inner( - pool: sqlx::mysql::MySqlPool, - statements: Vec, - start: std::time::Instant, -) -> Result { - let mut conn = pool.acquire().await.map_err(|e| format!("Failed to acquire connection: {}", e))?; - sqlx::raw_sql("START TRANSACTION") - .execute(&mut *conn) - .await - .map_err(|e| format!("Failed to begin transaction: {}", e))?; + let mut conn = pool.get_conn().await.map_err(|e| format!("Failed to acquire connection: {}", e))?; + conn.query_drop("START TRANSACTION").await.map_err(|e| format!("Failed to begin transaction: {}", e))?; let mut total_affected: u64 = 0; for (i, sql) in statements.iter().enumerate() { - match sqlx::raw_sql(sql).execute(&mut *conn).await { - Ok(r) => total_affected += r.rows_affected(), + match conn.query_iter(sql).await { + Ok(result) => total_affected += result.affected_rows(), Err(e) => { - let _ = sqlx::raw_sql("ROLLBACK").execute(&mut *conn).await; + let _ = conn.query_drop("ROLLBACK").await; return Err(format!("Statement {} failed: {}", i + 1, e)); } } } - sqlx::raw_sql("COMMIT").execute(&mut *conn).await.map_err(|e| format!("COMMIT failed: {}", e))?; + conn.query_drop("COMMIT").await.map_err(|e| format!("COMMIT failed: {}", e))?; Ok(db::QueryResult { columns: vec![], rows: vec![], diff --git a/crates/dbx-core/src/schema.rs b/crates/dbx-core/src/schema.rs index ef23d3b945..f065c006a1 100644 --- a/crates/dbx-core/src/schema.rs +++ b/crates/dbx-core/src/schema.rs @@ -908,17 +908,25 @@ fn first_string_cell(result: db::QueryResult) -> Result { } async fn mysql_object_source( - pool: &sqlx::mysql::MySqlPool, + pool: &db::mysql::MySqlPool, name: &str, kind: &db::ObjectSourceKind, ) -> Result { - use sqlx::Row; + use mysql_async::prelude::*; let sql = mysql_object_source_sql(name, kind); - let row: sqlx::mysql::MySqlRow = sqlx::raw_sql(&sql).fetch_one(pool).await.map_err(|e| e.to_string())?; + let mut conn = pool.get_conn().await.map_err(|e| e.to_string())?; + let result = conn.query_iter(&sql).await.map_err(|e| e.to_string())?; + let rows: Vec = result.collect_and_drop().await.map_err(|e| e.to_string())?; + let row = rows.first().ok_or("Object source not found")?; let index = if matches!(kind, db::ObjectSourceKind::View) { 1 } else { 2 }; - row.try_get::(index) - .or_else(|_| row.try_get::, _>(index).map(|b| String::from_utf8_lossy(&b).to_string())) - .map_err(|e| e.to_string()) + row.get_opt::(index) + .and_then(|result| result.ok()) + .or_else(|| { + row.get_opt::, usize>(index) + .and_then(|result| result.ok()) + .map(|b| String::from_utf8_lossy(&b).to_string()) + }) + .ok_or_else(|| "Failed to read object source".to_string()) } pub async fn get_object_source_core( @@ -1040,13 +1048,21 @@ mod ddl_tests { } } -pub async fn mysql_ddl(pool: &sqlx::mysql::MySqlPool, table: &str) -> Result { - use sqlx::Row; +pub async fn mysql_ddl(pool: &db::mysql::MySqlPool, table: &str) -> Result { + use mysql_async::prelude::*; let sql = format!("SHOW CREATE TABLE `{}`", table.replace('`', "``")); - let row: sqlx::mysql::MySqlRow = sqlx::raw_sql(&sql).fetch_one(pool).await.map_err(|e| e.to_string())?; - row.try_get::(1) - .or_else(|_| row.try_get::, _>(1).map(|b| String::from_utf8_lossy(&b).to_string())) - .map_err(|e| e.to_string()) + let mut conn = pool.get_conn().await.map_err(|e| e.to_string())?; + let result = conn.query_iter(&sql).await.map_err(|e| e.to_string())?; + let rows: Vec = result.collect_and_drop().await.map_err(|e| e.to_string())?; + let row = rows.first().ok_or("DDL not found")?; + row.get_opt::(1) + .and_then(|result| result.ok()) + .or_else(|| { + row.get_opt::, usize>(1) + .and_then(|result| result.ok()) + .map(|b| String::from_utf8_lossy(&b).to_string()) + }) + .ok_or_else(|| "Failed to read DDL".to_string()) } pub async fn sqlite_ddl(pool: &sqlx::sqlite::SqlitePool, table: &str) -> Result { diff --git a/src-tauri/src/commands/connection.rs b/src-tauri/src/commands/connection.rs index 5433ad3abd..2d25e62d75 100644 --- a/src-tauri/src/commands/connection.rs +++ b/src-tauri/src/commands/connection.rs @@ -217,21 +217,21 @@ pub async fn test_connection(state: State<'_, Arc>, config: Connection Ok(()) => match config.db_type { DatabaseType::Mysql if config.needs_bare_mysql() => match db::mysql::connect_bare(&url).await { Ok(pool) => { - pool.close().await; + let _ = pool.disconnect().await; Ok("Connection successful".to_string()) } Err(e) => Err(e), }, DatabaseType::Mysql => match db::mysql::connect(&url).await { Ok(pool) => { - pool.close().await; + let _ = pool.disconnect().await; Ok("Connection successful".to_string()) } Err(e) => Err(e), }, DatabaseType::Doris | DatabaseType::StarRocks => match db::mysql::connect_bare(&url).await { Ok(pool) => { - pool.close().await; + let _ = pool.disconnect().await; Ok("Connection successful".to_string()) } Err(e) => Err(e), @@ -435,7 +435,9 @@ pub async fn disconnect_db(state: State<'_, Arc>, connection_id: Strin for key in keys_to_remove { if let Some(pool) = conns.remove(&key) { match pool { - PoolKind::Mysql(p, _) => p.close().await, + PoolKind::Mysql(p, _) => { + let _ = p.disconnect().await; + } PoolKind::Postgres(p) => p.close().await, PoolKind::Sqlite(p) => p.close().await, PoolKind::Redis(_) => {}