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(_) => {}