diff --git a/crates/dbx-core/src/connection.rs b/crates/dbx-core/src/connection.rs index 4f9de1721..3feb93768 100644 --- a/crates/dbx-core/src/connection.rs +++ b/crates/dbx-core/src/connection.rs @@ -17,8 +17,15 @@ pub fn expand_tilde(path: &str) -> String { path.to_string() } +#[derive(Debug, Clone, Copy, PartialEq)] +pub enum MysqlMode { + Normal, + Bare, + OceanBaseOracle, +} + pub enum PoolKind { - Mysql(sqlx::mysql::MySqlPool, bool), + Mysql(sqlx::mysql::MySqlPool, MysqlMode), Postgres(sqlx::postgres::PgPool), Sqlite(sqlx::sqlite::SqlitePool), Redis(tokio::sync::Mutex), @@ -111,11 +118,15 @@ impl AppState { let url = connection_url_for_endpoint(&db_config, &host, port); let pool = match db_config.db_type { DatabaseType::Mysql if db_config.needs_bare_mysql() => { - PoolKind::Mysql(db::mysql::connect_bare(&url).await?, true) + PoolKind::Mysql(db::mysql::connect_bare(&url).await?, MysqlMode::Bare) + } + DatabaseType::Mysql => { + let pool = db::mysql::connect(&url).await?; + let mode = detect_ob_oracle_mode(&db_config, &pool).await; + PoolKind::Mysql(pool, mode) } - DatabaseType::Mysql => PoolKind::Mysql(db::mysql::connect(&url).await?, false), DatabaseType::Doris | DatabaseType::StarRocks => { - PoolKind::Mysql(db::mysql::connect_bare(&url).await?, true) + PoolKind::Mysql(db::mysql::connect_bare(&url).await?, MysqlMode::Bare) } DatabaseType::Postgres | DatabaseType::Redshift => PoolKind::Postgres(db::postgres::connect(&url).await?), DatabaseType::Sqlite => PoolKind::Sqlite(db::sqlite::connect_path(&expand_tilde(&db_config.host)).await?), @@ -288,3 +299,17 @@ pub async fn probe_connection_endpoint(config: &ConnectionConfig, host: &str, po _ => db::probe_tcp_endpoint(&format!("{:?}", config.db_type), host, port).await, } } + +async fn detect_ob_oracle_mode(config: &ConnectionConfig, pool: &sqlx::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, + } +} diff --git a/crates/dbx-core/src/db/mod.rs b/crates/dbx-core/src/db/mod.rs index d54f5e8e1..c7dbdc574 100644 --- a/crates/dbx-core/src/db/mod.rs +++ b/crates/dbx-core/src/db/mod.rs @@ -6,6 +6,7 @@ pub mod file_validator; pub mod gaussdb_driver; pub mod mongo_driver; pub mod mysql; +pub mod ob_oracle; pub mod oracle_driver; pub mod postgres; pub mod redis_driver; diff --git a/crates/dbx-core/src/db/ob_oracle.rs b/crates/dbx-core/src/db/ob_oracle.rs new file mode 100644 index 000000000..340298f6f --- /dev/null +++ b/crates/dbx-core/src/db/ob_oracle.rs @@ -0,0 +1,200 @@ +use sqlx::mysql::{MySqlPool, MySqlRow}; +use sqlx::Row; + +use crate::types::{ColumnInfo, DatabaseInfo, ForeignKeyInfo, IndexInfo, TableInfo, TriggerInfo}; + +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())) + .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_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())) +} + +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())?; + + Ok(rows.iter().map(|row| DatabaseInfo { name: get_str(row, 0) }).collect()) +} + +pub async fn list_tables(pool: &MySqlPool, schema: &str) -> Result, String> { + let sql = format!( + "SELECT TABLE_NAME, 'TABLE' AS TABLE_TYPE FROM ALL_TABLES WHERE OWNER = {s} \ + UNION ALL \ + SELECT VIEW_NAME, 'VIEW' AS TABLE_TYPE FROM ALL_VIEWS WHERE OWNER = {s} \ + ORDER BY 1", + s = quote_value(schema), + ); + let rows: Vec = sqlx::raw_sql(&sql).fetch_all(pool).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()) +} + +pub async fn get_columns(pool: &MySqlPool, 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, \ + CASE WHEN cc.COLUMN_NAME IS NOT NULL THEN 1 ELSE 0 END AS IS_PK \ + FROM ALL_TAB_COLUMNS c \ + LEFT JOIN ( \ + SELECT cols.OWNER, cols.TABLE_NAME, cols.COLUMN_NAME \ + FROM ALL_CONS_COLUMNS cols \ + JOIN ALL_CONSTRAINTS con ON con.CONSTRAINT_NAME = cols.CONSTRAINT_NAME AND con.OWNER = cols.OWNER \ + WHERE con.CONSTRAINT_TYPE = 'P' \ + ) cc ON cc.OWNER = c.OWNER AND cc.TABLE_NAME = c.TABLE_NAME AND cc.COLUMN_NAME = c.COLUMN_NAME \ + WHERE c.OWNER = {s} AND c.TABLE_NAME = {t} \ + ORDER BY c.COLUMN_ID", + 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())?; + + Ok(rows + .iter() + .map(|row| { + let data_type = get_str(row, 1); + let precision = get_opt_i32(row, 5); + let scale = get_opt_i32(row, 6); + let length = get_opt_i32(row, 4); + let display_type = format_oracle_type(&data_type, precision, scale, length); + ColumnInfo { + name: get_str(row, 0), + 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, + extra: None, + comment: None, + numeric_precision: precision, + numeric_scale: scale, + character_maximum_length: length, + } + }) + .collect()) +} + +fn format_oracle_type(data_type: &str, precision: Option, scale: Option, length: Option) -> String { + match data_type.to_uppercase().as_str() { + "NUMBER" => match (precision, scale) { + (Some(p), Some(s)) if s > 0 => format!("NUMBER({p},{s})"), + (Some(p), _) => format!("NUMBER({p})"), + _ => "NUMBER".to_string(), + }, + "VARCHAR2" | "NVARCHAR2" | "CHAR" | "NCHAR" | "RAW" => match length { + Some(l) => format!("{data_type}({l})"), + None => data_type.to_string(), + }, + _ => data_type.to_string(), + } +} + +pub async fn list_indexes(pool: &MySqlPool, 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, \ + ai.UNIQUENESS, \ + CASE WHEN ac.CONSTRAINT_TYPE = 'P' THEN 1 ELSE 0 END AS IS_PRIMARY \ + FROM ALL_INDEXES ai \ + JOIN ALL_IND_COLUMNS aic ON ai.INDEX_NAME = aic.INDEX_NAME AND ai.TABLE_OWNER = aic.TABLE_OWNER \ + LEFT JOIN ALL_CONSTRAINTS ac ON ac.INDEX_NAME = ai.INDEX_NAME AND ac.OWNER = ai.TABLE_OWNER AND ac.CONSTRAINT_TYPE = 'P' \ + WHERE ai.TABLE_OWNER = {s} AND ai.TABLE_NAME = {t} \ + GROUP BY ai.INDEX_NAME, ai.UNIQUENESS, ac.CONSTRAINT_TYPE \ + ORDER BY ai.INDEX_NAME", + 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())?; + + Ok(rows + .iter() + .map(|row| { + let cols_str = get_str(row, 1); + IndexInfo { + 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, + filter: None, + index_type: None, + included_columns: None, + comment: None, + } + }) + .collect()) +} + +pub async fn list_foreign_keys(pool: &MySqlPool, 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 \ + FROM ALL_CONSTRAINTS ac \ + JOIN ALL_CONS_COLUMNS acc ON ac.CONSTRAINT_NAME = acc.CONSTRAINT_NAME AND ac.OWNER = acc.OWNER \ + JOIN ALL_CONSTRAINTS ac2 ON ac.R_CONSTRAINT_NAME = ac2.CONSTRAINT_NAME AND ac.R_OWNER = ac2.OWNER \ + JOIN ALL_CONS_COLUMNS acc2 ON ac2.CONSTRAINT_NAME = acc2.CONSTRAINT_NAME AND ac2.OWNER = acc2.OWNER \ + AND acc.POSITION = acc2.POSITION \ + WHERE ac.CONSTRAINT_TYPE = 'R' AND ac.OWNER = {s} AND ac.TABLE_NAME = {t} \ + ORDER BY ac.CONSTRAINT_NAME, acc.POSITION", + 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())?; + + Ok(rows + .iter() + .map(|row| ForeignKeyInfo { + name: get_str(row, 0), + column: get_str(row, 1), + ref_table: get_str(row, 2), + ref_column: get_str(row, 3), + }) + .collect()) +} + +pub async fn list_triggers(pool: &MySqlPool, schema: &str, table: &str) -> Result, String> { + let sql = format!( + "SELECT TRIGGER_NAME, TRIGGERING_EVENT, TRIGGER_TYPE \ + FROM ALL_TRIGGERS \ + WHERE TABLE_OWNER = {s} AND TABLE_NAME = {t} \ + ORDER BY TRIGGER_NAME", + 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())?; + + Ok(rows + .iter() + .map(|row| { + let trigger_type = get_str(row, 2); + let timing = if trigger_type.contains("BEFORE") { + "BEFORE" + } else if trigger_type.contains("AFTER") { + "AFTER" + } else { + "INSTEAD OF" + }; + TriggerInfo { name: get_str(row, 0), event: get_str(row, 1), timing: timing.to_string() } + }) + .collect()) +} diff --git a/crates/dbx-core/src/query.rs b/crates/dbx-core/src/query.rs index 564fa1fbb..6378abd76 100644 --- a/crates/dbx-core/src/query.rs +++ b/crates/dbx-core/src/query.rs @@ -154,9 +154,9 @@ pub async fn do_execute( }) .await } - PoolKind::Mysql(p, bare) => { + PoolKind::Mysql(p, mode) => { let p = p.clone(); - let bare = *bare; + let bare = *mode == crate::connection::MysqlMode::Bare; drop(connections); wait_for_query(cancel_token, db::mysql::execute_query(&p, sql, bare)).await.map(truncate_result) } @@ -396,7 +396,7 @@ pub async fn execute_statements_in_transaction( let conns = state.connections.lock().await; conns.get(&pool_key).map(|p| match p { PoolKind::Postgres(pg) => TxPath::Pg(pg.clone()), - PoolKind::Mysql(mp, bare) => TxPath::Mysql(mp.clone(), *bare), + PoolKind::Mysql(mp, _mode) => TxPath::Mysql(mp.clone(), false), PoolKind::Sqlite(sq) => TxPath::Sqlite(sq.clone()), PoolKind::ClickHouse(_) | PoolKind::SqlServer(_) | PoolKind::Dameng(_) | PoolKind::Gaussdb(_) => { TxPath::Explicit diff --git a/crates/dbx-core/src/schema.rs b/crates/dbx-core/src/schema.rs index 72fb0d5f2..59ab32ac9 100644 --- a/crates/dbx-core/src/schema.rs +++ b/crates/dbx-core/src/schema.rs @@ -1,7 +1,7 @@ use std::collections::HashMap; use std::sync::Arc; -use crate::connection::{AppState, PoolKind}; +use crate::connection::{AppState, MysqlMode, PoolKind}; use crate::db; pub fn duckdb_query_tables(con: &duckdb::Connection) -> Result, String> { @@ -155,7 +155,13 @@ pub async fn list_databases_core(state: &AppState, connection_id: &str) -> Resul let pool = connections.get(connection_id).ok_or("Connection not found")?; match pool { - PoolKind::Mysql(p, _) => db::mysql::list_databases(p).await, + PoolKind::Mysql(p, mode) => { + if *mode == MysqlMode::OceanBaseOracle { + db::ob_oracle::list_databases(p).await + } else { + db::mysql::list_databases(p).await + } + } PoolKind::Postgres(p) => db::postgres::list_databases(p).await, PoolKind::Sqlite(p) => db::sqlite::list_databases(p).await, PoolKind::DuckDb(_) => Ok(vec![db::DatabaseInfo { name: "main".to_string() }]), @@ -244,7 +250,13 @@ pub async fn list_tables_core( let pool = connections.get(&pool_key).ok_or("Pool not found")?; match pool { - PoolKind::Mysql(p, _) => db::mysql::list_tables(p, schema).await, + PoolKind::Mysql(p, mode) => { + if *mode == MysqlMode::OceanBaseOracle { + db::ob_oracle::list_tables(p, schema).await + } else { + db::mysql::list_tables(p, schema).await + } + } PoolKind::Postgres(p) => db::postgres::list_tables(p, schema).await, PoolKind::Sqlite(p) => db::sqlite::list_tables(p, schema).await, _ => Ok(vec![]), @@ -297,7 +309,13 @@ pub async fn get_columns_core( let pool = connections.get(&pool_key).ok_or("Pool not found")?; match pool { - PoolKind::Mysql(p, _) => db::mysql::get_columns(p, database, table).await, + PoolKind::Mysql(p, mode) => { + if *mode == MysqlMode::OceanBaseOracle { + db::ob_oracle::get_columns(p, database, table).await + } else { + db::mysql::get_columns(p, database, table).await + } + } PoolKind::Postgres(p) => db::postgres::get_columns(p, schema, table).await, PoolKind::Sqlite(p) => db::sqlite::get_columns(p, schema, table).await, _ => Ok(vec![]), @@ -341,7 +359,13 @@ pub async fn list_indexes_core( let pool = connections.get(&pool_key).ok_or("Pool not found")?; match pool { - PoolKind::Mysql(p, _) => db::mysql::list_indexes(p, schema, table).await, + PoolKind::Mysql(p, mode) => { + if *mode == MysqlMode::OceanBaseOracle { + db::ob_oracle::list_indexes(p, schema, table).await + } else { + db::mysql::list_indexes(p, schema, table).await + } + } PoolKind::Postgres(p) => db::postgres::list_indexes(p, schema, table).await, PoolKind::Sqlite(p) => db::sqlite::list_indexes(p, schema, table).await, _ => Ok(vec![]), @@ -385,7 +409,13 @@ pub async fn list_foreign_keys_core( let pool = connections.get(&pool_key).ok_or("Pool not found")?; match pool { - PoolKind::Mysql(p, _) => db::mysql::list_foreign_keys(p, schema, table).await, + PoolKind::Mysql(p, mode) => { + if *mode == MysqlMode::OceanBaseOracle { + db::ob_oracle::list_foreign_keys(p, schema, table).await + } else { + db::mysql::list_foreign_keys(p, schema, table).await + } + } PoolKind::Postgres(p) => db::postgres::list_foreign_keys(p, schema, table).await, PoolKind::Sqlite(p) => db::sqlite::list_foreign_keys(p, schema, table).await, _ => Ok(vec![]), @@ -429,7 +459,13 @@ pub async fn list_triggers_core( let pool = connections.get(&pool_key).ok_or("Pool not found")?; match pool { - PoolKind::Mysql(p, _) => db::mysql::list_triggers(p, schema, table).await, + PoolKind::Mysql(p, mode) => { + if *mode == MysqlMode::OceanBaseOracle { + db::ob_oracle::list_triggers(p, schema, table).await + } else { + db::mysql::list_triggers(p, schema, table).await + } + } PoolKind::Postgres(p) => db::postgres::list_triggers(p, schema, table).await, PoolKind::Sqlite(p) => db::sqlite::list_triggers(p, schema, table).await, _ => Ok(vec![]), diff --git a/crates/dbx-core/src/transfer.rs b/crates/dbx-core/src/transfer.rs index 68b70e221..be6f3c561 100644 --- a/crates/dbx-core/src/transfer.rs +++ b/crates/dbx-core/src/transfer.rs @@ -473,9 +473,9 @@ pub async fn execute_on_pool(state: &AppState, pool_key: &str, sql: &str) -> Res let pool = connections.get(pool_key).ok_or("Connection not found")?; match pool { - PoolKind::Mysql(p, bare) => { + PoolKind::Mysql(p, mode) => { let p = p.clone(); - let bare = *bare; + let bare = *mode == crate::connection::MysqlMode::Bare; drop(connections); db::mysql::execute_query(&p, sql, bare).await } diff --git a/src-tauri/src/commands/connection.rs b/src-tauri/src/commands/connection.rs index 4668328c1..8ba0da919 100644 --- a/src-tauri/src/commands/connection.rs +++ b/src-tauri/src/commands/connection.rs @@ -3,7 +3,7 @@ use tauri::State; pub use dbx_core::connection::{ connection_url_for_endpoint, expand_tilde, probe_connection_endpoint, redacted_connection_url_for_endpoint, - AppState, PoolKind, + AppState, MysqlMode, PoolKind, }; use dbx_core::db; use dbx_core::models::connection::{ConnectionConfig, DatabaseType}; @@ -149,9 +149,13 @@ pub async fn connect_db(state: State<'_, Arc>, config: ConnectionConfi let url = connection_url_for_endpoint(&config, &host, port); let pool = match config.db_type { - DatabaseType::Mysql if config.needs_bare_mysql() => PoolKind::Mysql(db::mysql::connect_bare(&url).await?, true), - DatabaseType::Mysql => PoolKind::Mysql(db::mysql::connect(&url).await?, false), - DatabaseType::Doris | DatabaseType::StarRocks => PoolKind::Mysql(db::mysql::connect_bare(&url).await?, true), + DatabaseType::Mysql if config.needs_bare_mysql() => { + PoolKind::Mysql(db::mysql::connect_bare(&url).await?, MysqlMode::Bare) + } + DatabaseType::Mysql => PoolKind::Mysql(db::mysql::connect(&url).await?, MysqlMode::Normal), + DatabaseType::Doris | DatabaseType::StarRocks => { + PoolKind::Mysql(db::mysql::connect_bare(&url).await?, MysqlMode::Bare) + } DatabaseType::Postgres | DatabaseType::Redshift => PoolKind::Postgres(db::postgres::connect(&url).await?), DatabaseType::Sqlite => PoolKind::Sqlite(db::sqlite::connect_path(&expand_tilde(&config.host)).await?), DatabaseType::Redis => {