From c50323cd6c699392c7516d9da0f45ce1b9b059c3 Mon Sep 17 00:00:00 2001 From: t8y2 <1156263951@qq.com> Date: Wed, 13 May 2026 09:20:06 +0800 Subject: [PATCH] feat(agent): migrate Oracle to JDBC agent, add install progress bar and toast - Migrate Oracle from rust-oracle to JDBC agent (ojdbc11) - Delete oracle_driver.rs (567 lines) and rust-oracle dependency - Add streaming download progress bar for driver/JRE installation - Add toast notifications for install/uninstall success/failure - Add "update available" badge in driver manager --- Cargo.lock | 132 +----- crates/dbx-core/Cargo.toml | 1 - crates/dbx-core/src/agent_manager.rs | 1 + crates/dbx-core/src/connection.rs | 81 +--- crates/dbx-core/src/db/mod.rs | 1 - crates/dbx-core/src/db/oracle_driver.rs | 567 ------------------------ crates/dbx-core/src/query.rs | 23 - crates/dbx-core/src/schema.rs | 136 +----- crates/dbx-core/src/transfer.rs | 15 - src-tauri/Cargo.toml | 1 - src-tauri/src/commands/agents.rs | 61 ++- src-tauri/src/commands/connection.rs | 37 +- src/components/config/DriverManager.vue | 89 +++- src/lib/databaseCapabilities.ts | 2 +- 14 files changed, 175 insertions(+), 972 deletions(-) delete mode 100644 crates/dbx-core/src/db/oracle_driver.rs diff --git a/Cargo.lock b/Cargo.lock index 3e5e3436d..9df21d42d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1573,7 +1573,6 @@ dependencies = [ "redis", "reqwest 0.12.28", "russh", - "rust-oracle", "rust_decimal", "rustls 0.23.40", "serde", @@ -1615,7 +1614,6 @@ dependencies = [ "reqwest 0.12.28", "russh", "rust-gaussdb", - "rust-oracle", "rust_decimal", "rustls 0.23.40", "serde", @@ -2937,17 +2935,6 @@ dependencies = [ "windows-sys 0.61.2", ] -[[package]] -name = "hostname" -version = "0.4.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "617aaa3557aef3810a6369d0a99fac8a080891b68bd9f9812a1eeda0c0730cbd" -dependencies = [ - "cfg-if", - "libc", - "windows-link 0.2.1", -] - [[package]] name = "html5ever" version = "0.38.0" @@ -3092,7 +3079,7 @@ dependencies = [ "libc", "percent-encoding", "pin-project-lite", - "socket2 0.6.3", + "socket2", "system-configuration", "tokio", "tower-service", @@ -3348,7 +3335,7 @@ version = "0.3.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4d40460c0ce33d6ce4b0630ad68ff63d6661961c48b6dba35e5a4d81cfb48222" dependencies = [ - "socket2 0.6.3", + "socket2", "widestring", "windows-registry", "windows-result 0.4.1", @@ -4043,7 +4030,7 @@ dependencies = [ "serde_with", "sha1 0.10.6", "sha2 0.10.9", - "socket2 0.6.3", + "socket2", "stringprep", "strsim", "take_mut", @@ -4865,21 +4852,6 @@ dependencies = [ "spki 0.8.0-rc.4", ] -[[package]] -name = "pkcs5" -version = "0.7.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e847e2c91a18bfa887dd028ec33f2fe6f25db77db3619024764914affe8b69a6" -dependencies = [ - "aes 0.8.4", - "cbc 0.1.2", - "der 0.7.10", - "pbkdf2 0.12.2", - "scrypt 0.11.0", - "sha2 0.10.9", - "spki 0.7.3", -] - [[package]] name = "pkcs5" version = "0.8.0-rc.13" @@ -4892,7 +4864,7 @@ dependencies = [ "der 0.8.0", "pbkdf2 0.13.0", "rand_core 0.10.1", - "scrypt 0.12.0", + "scrypt", "sha2 0.11.0", "spki 0.8.0-rc.4", ] @@ -4904,8 +4876,6 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f950b2377845cebe5cf8b5165cb3cc1a5e0fa5cfa3e1f7f55707d8fd82e0a7b7" dependencies = [ "der 0.7.10", - "pkcs5 0.7.1", - "rand_core 0.6.4", "spki 0.7.3", ] @@ -4916,7 +4886,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "12922b6296c06eb741b02d7b5161e3aaa22864af38dfa025a1a3ba3f68c84577" dependencies = [ "der 0.8.0", - "pkcs5 0.8.0-rc.13", + "pkcs5", "rand_core 0.10.1", "spki 0.8.0-rc.4", ] @@ -5242,7 +5212,7 @@ dependencies = [ "quinn-udp", "rustc-hash", "rustls 0.23.40", - "socket2 0.6.3", + "socket2", "thiserror 2.0.18", "tokio", "tracing", @@ -5279,7 +5249,7 @@ dependencies = [ "cfg_aliases", "libc", "once_cell", - "socket2 0.6.3", + "socket2", "tracing", "windows-sys 0.60.2", ] @@ -5411,7 +5381,7 @@ dependencies = [ "rustls-native-certs 0.8.3", "ryu", "sha1_smol", - "socket2 0.6.3", + "socket2", "tokio", "tokio-rustls 0.26.4", "tokio-util", @@ -5769,7 +5739,7 @@ dependencies = [ "pbkdf2 0.12.2", "pbkdf2 0.13.0", "pkcs1 0.8.0-rc.4", - "pkcs5 0.8.0-rc.13", + "pkcs5", "pkcs8 0.11.0-rc.11", "polyval 0.7.1", "rand 0.10.1", @@ -5777,8 +5747,8 @@ dependencies = [ "rsa 0.10.0-rc.16", "russh-cryptovec", "russh-util", - "salsa20 0.11.0", - "scrypt 0.12.0", + "salsa20", + "scrypt", "sec1", "sha1 0.10.6", "sha1 0.11.0", @@ -5846,39 +5816,6 @@ dependencies = [ "tokio-util", ] -[[package]] -name = "rust-oracle" -version = "0.1.6" -source = "git+https://github.com/t8y2/rust-oracle?branch=main#72ceb94752c4c16daa4f449fc6d7a0b16ba9fd09" -dependencies = [ - "aes 0.8.4", - "async-trait", - "bytes", - "cbc 0.1.2", - "chrono", - "hex", - "hmac 0.12.1", - "hostname", - "indexmap 2.14.0", - "md-5 0.10.6", - "pbkdf2 0.12.2", - "pkcs8 0.10.2", - "rand 0.8.6", - "rustls 0.23.40", - "rustls-pemfile 2.2.0", - "rustls-pki-types", - "serde", - "serde_json", - "sha1 0.10.6", - "sha2 0.10.9", - "socket2 0.5.10", - "thiserror 1.0.69", - "tokio", - "tokio-rustls 0.26.4", - "tracing", - "webpki-roots 0.26.11", -] - [[package]] name = "rust_decimal" version = "1.42.0" @@ -6003,7 +5940,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a9aace74cb666635c918e9c12bc0d348266037aa8eb599b5cba565709a8dff00" dependencies = [ "openssl-probe 0.1.6", - "rustls-pemfile 1.0.4", + "rustls-pemfile", "schannel", "security-framework 2.11.1", ] @@ -6029,15 +5966,6 @@ 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" @@ -6109,15 +6037,6 @@ version = "1.0.23" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9774ba4a74de5f7b1c1451ed6cd5285a32eddb5cccb8cc655a4e50009e06477f" -[[package]] -name = "salsa20" -version = "0.10.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "97a22f5af31f73a954c10289c93e8a50cc23d971e80ee446f1f6f7137a088213" -dependencies = [ - "cipher 0.4.4", -] - [[package]] name = "salsa20" version = "0.11.0" @@ -6203,17 +6122,6 @@ version = "1.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" -[[package]] -name = "scrypt" -version = "0.11.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0516a385866c09368f0b5bcd1caff3366aace790fcd46e2bb032697bb172fd1f" -dependencies = [ - "pbkdf2 0.12.2", - "salsa20 0.10.2", - "sha2 0.10.9", -] - [[package]] name = "scrypt" version = "0.12.0" @@ -6222,7 +6130,7 @@ checksum = "d87af57419b594aa23fa95f09f0e06d80d84ba01c26148c43844cad6ff4485f0" dependencies = [ "cfg-if", "pbkdf2 0.13.0", - "salsa20 0.11.0", + "salsa20", "sha2 0.11.0", ] @@ -6702,16 +6610,6 @@ 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" @@ -7742,7 +7640,7 @@ dependencies = [ "pretty-hex", "rust_decimal", "rustls-native-certs 0.6.3", - "rustls-pemfile 1.0.4", + "rustls-pemfile", "thiserror 1.0.69", "tokio-rustls 0.24.1", "tokio-util", @@ -7829,7 +7727,7 @@ dependencies = [ "parking_lot", "pin-project-lite", "signal-hook-registry", - "socket2 0.6.3", + "socket2", "tokio-macros", "windows-sys 0.61.2", ] diff --git a/crates/dbx-core/Cargo.toml b/crates/dbx-core/Cargo.toml index 699f54397..9050a1d46 100644 --- a/crates/dbx-core/Cargo.toml +++ b/crates/dbx-core/Cargo.toml @@ -26,7 +26,6 @@ tiberius = { version = "0.12.3", default-features = false, features = ["tds73", reqwest = { version = "0.12", default-features = false, features = ["json", "stream", "rustls-tls", "socks"] } futures = "0.3" mongodb = "3.2.5" -rust-oracle = { git = "https://github.com/t8y2/rust-oracle", branch = "main" } russh = "0.60" portpicker = "0.1.1" csv = "1" diff --git a/crates/dbx-core/src/agent_manager.rs b/crates/dbx-core/src/agent_manager.rs index acf0859b5..a5b975fd0 100644 --- a/crates/dbx-core/src/agent_manager.rs +++ b/crates/dbx-core/src/agent_manager.rs @@ -121,6 +121,7 @@ impl AgentManager { DatabaseType::Kingbase => Some("kingbase"), DatabaseType::Vastbase => Some("vastbase"), DatabaseType::Goldendb => Some("goldendb"), + DatabaseType::Oracle => Some("oracle"), _ => None, } } diff --git a/crates/dbx-core/src/connection.rs b/crates/dbx-core/src/connection.rs index c8d498b2d..532382d5d 100644 --- a/crates/dbx-core/src/connection.rs +++ b/crates/dbx-core/src/connection.rs @@ -1,6 +1,5 @@ use std::collections::HashMap; use std::path::PathBuf; -use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; use tokio::sync::RwLock; @@ -43,7 +42,6 @@ pub enum PoolKind { MongoDb(mongodb::Client), ClickHouse(db::clickhouse_driver::ChClient), SqlServer(Arc>), - Oracle(Arc), Elasticsearch(db::elasticsearch_driver::EsClient), Agent(Arc>), Gaussdb(Arc>), @@ -51,46 +49,6 @@ pub enum PoolKind { ExternalDriver { driver_id: String, config: ConnectionConfig, session: Arc }, } -pub struct OraclePool { - clients: Vec>>, - next: AtomicUsize, -} - -impl OraclePool { - pub fn new(clients: Vec) -> Self { - Self { - clients: clients.into_iter().map(|client| Arc::new(tokio::sync::Mutex::new(client))).collect(), - next: AtomicUsize::new(0), - } - } - - pub fn client(&self) -> Arc> { - let index = self.next.fetch_add(1, Ordering::Relaxed) % self.clients.len(); - self.clients[index].clone() - } - - pub fn primary(&self) -> Arc> { - self.clients[0].clone() - } -} - -async fn connect_oracle_pool( - host: &str, - port: u16, - service: &str, - user: &str, - pass: &str, - sysdba: bool, -) -> Result { - let (first, second, third) = tokio::try_join!( - db::oracle_driver::connect(host, port, service, user, pass, sysdba), - db::oracle_driver::connect(host, port, service, user, pass, sysdba), - db::oracle_driver::connect(host, port, service, user, pass, sysdba), - )?; - let clients = vec![first, second, third]; - Ok(OraclePool::new(clients)) -} - pub struct AppState { pub connections: RwLock>, pub configs: RwLock>, @@ -187,20 +145,7 @@ impl AppState { let conns = self.connections.read().await; if conns.contains_key(&pool_key) { - if let Some(PoolKind::Oracle(pool)) = conns.get(&pool_key) { - let client = pool.primary(); - let conn = client.lock().await; - if conn.is_closed() { - drop(conn); - drop(conns); - log::info!("[oracle] connection closed, reconnecting..."); - self.connections.write().await.remove(&pool_key); - } else { - return Ok(pool_key); - } - } else { - return Ok(pool_key); - } + return Ok(pool_key); } else { drop(conns); } @@ -259,25 +204,17 @@ impl AppState { .await?; PoolKind::SqlServer(Arc::new(tokio::sync::Mutex::new(client))) } - DatabaseType::Oracle => { - let pool = connect_oracle_pool( - &host, - port, - db_config.database.as_deref().unwrap_or("ORCL"), - &db_config.username, - &db_config.password, - db_config.sysdba, - ) - .await?; - PoolKind::Oracle(Arc::new(pool)) - } DatabaseType::Elasticsearch => { let client = db::elasticsearch_driver::EsClient::new(&url, Some(&db_config.username), Some(&db_config.password)); db::elasticsearch_driver::test_connection(&client).await?; PoolKind::Elasticsearch(client) } - DatabaseType::Dameng | DatabaseType::Kingbase | DatabaseType::Vastbase | DatabaseType::Goldendb => { + DatabaseType::Dameng + | DatabaseType::Kingbase + | DatabaseType::Vastbase + | DatabaseType::Goldendb + | DatabaseType::Oracle => { let mut client = self.agent_manager.spawn(&db_config.db_type).await?; client .call::( @@ -465,7 +402,11 @@ pub async fn probe_connection_endpoint(config: &ConnectionConfig, host: &str, po DatabaseType::Sqlite | DatabaseType::DuckDb => Ok(()), DatabaseType::MongoDb if config.connection_string.as_deref().is_some_and(|value| !value.is_empty()) => Ok(()), DatabaseType::Jdbc => Ok(()), - DatabaseType::Dameng | DatabaseType::Kingbase | DatabaseType::Vastbase | DatabaseType::Goldendb => Ok(()), + DatabaseType::Dameng + | DatabaseType::Kingbase + | DatabaseType::Vastbase + | DatabaseType::Goldendb + | DatabaseType::Oracle => Ok(()), _ => db::probe_tcp_endpoint(&format!("{:?}", config.db_type), host, port).await, } } diff --git a/crates/dbx-core/src/db/mod.rs b/crates/dbx-core/src/db/mod.rs index 96c465ac0..fd66cc752 100644 --- a/crates/dbx-core/src/db/mod.rs +++ b/crates/dbx-core/src/db/mod.rs @@ -7,7 +7,6 @@ pub mod gaussdb_driver; pub mod mongo_driver; pub mod mysql; pub mod ob_oracle; -pub mod oracle_driver; pub mod postgres; pub mod proxy_tunnel; pub mod redis_driver; diff --git a/crates/dbx-core/src/db/oracle_driver.rs b/crates/dbx-core/src/db/oracle_driver.rs deleted file mode 100644 index 363843a5a..000000000 --- a/crates/dbx-core/src/db/oracle_driver.rs +++ /dev/null @@ -1,567 +0,0 @@ -use log; -use rust_oracle::{Config, Connection}; -use std::time::Instant; - -use super::{connection_timeout, CONNECTION_TIMEOUT_SECS}; -use crate::sql::starts_with_executable_sql_keyword; -use crate::types::{ColumnInfo, DatabaseInfo, ForeignKeyInfo, IndexInfo, QueryResult, TableInfo, TriggerInfo}; - -pub type OracleClient = Connection; -const ORACLE_QUERY_LIMIT: usize = crate::query::MAX_ROWS + 1; - -fn quote_literal(value: &str) -> String { - format!("'{}'", value.replace('\'', "''")) -} - -pub async fn connect( - host: &str, - port: u16, - service: &str, - user: &str, - pass: &str, - sysdba: bool, -) -> Result { - let config = Config::new(host, port, service, user, pass).with_statement_cache_size(0).sysdba_flag(sysdba); - let conn = tokio::time::timeout(connection_timeout(), Connection::connect_with_config(config)) - .await - .map_err(|_| format!("Oracle connection timed out ({CONNECTION_TIMEOUT_SECS}s)"))? - .map_err(|e| format!("Oracle connection failed: {e}"))?; - - Ok(conn) -} - -fn value_to_json(val: &rust_oracle::Value) -> serde_json::Value { - match val { - rust_oracle::Value::Null => serde_json::Value::Null, - rust_oracle::Value::String(s) => serde_json::Value::String(s.clone()), - rust_oracle::Value::Integer(n) => serde_json::Value::Number((*n).into()), - rust_oracle::Value::Float(f) => { - serde_json::Number::from_f64(*f).map(serde_json::Value::Number).unwrap_or(serde_json::Value::Null) - } - rust_oracle::Value::Date(d) => serde_json::Value::String(format!( - "{:04}-{:02}-{:02} {:02}:{:02}:{:02}", - d.year, d.month, d.day, d.hour, d.minute, d.second - )), - rust_oracle::Value::Timestamp(ts) => { - let base = format!( - "{:04}-{:02}-{:02} {:02}:{:02}:{:02}", - ts.year, ts.month, ts.day, ts.hour, ts.minute, ts.second - ); - let value = if ts.microsecond > 0 { format!("{base}.{:06}", ts.microsecond) } else { base }; - serde_json::Value::String(value) - } - rust_oracle::Value::Boolean(b) => serde_json::Value::Bool(*b), - rust_oracle::Value::Json(v) => v.clone(), - _ => serde_json::Value::String(format!("{val:?}")), - } -} - -pub async fn list_databases(conn: &OracleClient) -> Result, String> { - let result = conn - .query( - "WITH schema_names AS ( \ - SELECT SYS_CONTEXT('USERENV', 'CURRENT_SCHEMA') AS owner FROM DUAL \ - UNION \ - SELECT DISTINCT owner FROM all_tables \ - UNION \ - SELECT DISTINCT owner FROM all_views \ - ) \ - SELECT owner \ - FROM schema_names \ - WHERE owner IS NOT NULL \ - AND owner NOT IN (\ - 'SYS','SYSTEM','SYSMAN','DBSNMP','SYSBACKUP','SYSDG','SYSKM','OUTLN',\ - 'AUDSYS','LBACSYS','DVF','DVSYS','APPQOSSYS','CTXSYS','MDSYS','MDDATA',\ - 'ORDSYS','ORDDATA','ORDPLUGINS','XDB','ANONYMOUS','DIP','EXFSYS',\ - 'GSMADMIN_INTERNAL','GSMCATUSER','GSMUSER','OJVMSYS','OLAPSYS',\ - 'ORACLE_OCM','SI_INFORMTN_SCHEMA','WMSYS','XS$NULL','DBSFWUSER',\ - 'REMOTE_SCHEDULER_AGENT','PDBADMIN','DGPDB_INT','OPS$ORACLE',\ - 'GGSYS','FLOWS_FILES','APEX_PUBLIC_USER'\ - ) \ - AND owner NOT LIKE 'APEX_%' \ - AND owner NOT LIKE 'FLOWS_%' \ - AND owner NOT LIKE '%$%' \ - ORDER BY CASE \ - WHEN owner = SYS_CONTEXT('USERENV', 'CURRENT_SCHEMA') THEN 0 \ - WHEN owner = SYS_CONTEXT('USERENV', 'SESSION_USER') THEN 1 \ - ELSE 2 \ - END, owner", - &[], - ) - .await - .map_err(|e| { - log::error!("[oracle] list_databases failed: {e}"); - e.to_string() - })?; - - Ok(result.rows.iter().map(|row| DatabaseInfo { name: row.get_string(0).unwrap_or("").to_string() }).collect()) -} - -pub async fn list_schemas(conn: &OracleClient) -> Result, String> { - let dbs = list_databases(conn).await?; - Ok(dbs.into_iter().map(|d| d.name).collect()) -} - -pub async fn list_tables(conn: &OracleClient, schema: &str) -> Result, String> { - let s = quote_literal(schema); - let sql = format!( - "SELECT o.OBJECT_NAME, \ - CASE o.OBJECT_TYPE WHEN 'VIEW' THEN 'VIEW' ELSE 'TABLE' END AS TABLE_TYPE, \ - c.COMMENTS \ - FROM ALL_OBJECTS o \ - LEFT JOIN ALL_TAB_COMMENTS c ON c.OWNER = o.OWNER AND c.TABLE_NAME = o.OBJECT_NAME \ - WHERE o.OWNER = {s} AND o.OBJECT_TYPE IN ('TABLE','VIEW') \ - ORDER BY o.OBJECT_NAME" - ); - log::debug!("[oracle] list_tables: schema={schema}, sql={sql}"); - let result = conn.query(&sql, &[]).await.map_err(|e| { - log::error!("[oracle] list_tables failed: {e}"); - e.to_string() - })?; - Ok(result - .rows - .iter() - .map(|row| TableInfo { - name: row.get_string(0).unwrap_or("").to_string(), - table_type: row.get_string(1).unwrap_or("TABLE").to_string(), - comment: row.get_string(2).filter(|s| !s.is_empty()).map(|s| s.to_string()), - }) - .collect()) -} - -pub async fn list_objects(conn: &OracleClient, schema: &str) -> Result, String> { - let s = quote_literal(schema); - let sql = format!( - "SELECT o.OBJECT_NAME, \ - CASE o.OBJECT_TYPE \ - WHEN 'TABLE' THEN 'TABLE' \ - WHEN 'VIEW' THEN 'VIEW' \ - WHEN 'PROCEDURE' THEN 'PROCEDURE' \ - WHEN 'FUNCTION' THEN 'FUNCTION' \ - ELSE o.OBJECT_TYPE \ - END AS OBJECT_TYPE, \ - c.COMMENTS \ - FROM ALL_OBJECTS o \ - LEFT JOIN ALL_TAB_COMMENTS c ON c.OWNER = o.OWNER AND c.TABLE_NAME = o.OBJECT_NAME \ - WHERE o.OWNER = {s} \ - AND o.OBJECT_TYPE IN ('TABLE','VIEW','PROCEDURE','FUNCTION') \ - AND o.OBJECT_NAME NOT LIKE 'BIN$%' \ - ORDER BY CASE o.OBJECT_TYPE \ - WHEN 'TABLE' THEN 0 \ - WHEN 'VIEW' THEN 1 \ - WHEN 'PROCEDURE' THEN 2 \ - WHEN 'FUNCTION' THEN 3 \ - ELSE 4 \ - END, o.OBJECT_NAME" - ); - let result = conn.query(&sql, &[]).await.map_err(|e| e.to_string())?; - Ok(result - .rows - .iter() - .map(|row| crate::types::ObjectInfo { - name: row.get_string(0).unwrap_or("").to_string(), - object_type: row.get_string(1).unwrap_or("TABLE").to_string(), - schema: Some(schema.to_string()), - comment: row.get_string(2).filter(|s| !s.is_empty()).map(|s| s.to_string()), - }) - .collect()) -} - -pub async fn get_columns(conn: &OracleClient, schema: &str, table: &str) -> Result, String> { - log::debug!("[oracle] get_columns: schema={schema}, table={table}"); - let s = quote_literal(schema); - let t = quote_literal(table); - - let col_result = conn - .query( - &format!( - "SELECT c.COLUMN_NAME, c.DATA_TYPE, c.NULLABLE, c.DATA_PRECISION, c.DATA_SCALE, c.DATA_LENGTH, \ - c.CHAR_LENGTH, cc.COMMENTS, CASE WHEN pk.COLUMN_NAME IS NULL THEN 0 ELSE 1 END AS IS_PK \ - FROM ALL_TAB_COLUMNS c \ - LEFT JOIN ALL_COL_COMMENTS cc ON cc.OWNER = c.OWNER AND cc.TABLE_NAME = c.TABLE_NAME AND cc.COLUMN_NAME = c.COLUMN_NAME \ - LEFT JOIN ( \ - SELECT cols.COLUMN_NAME \ - FROM ALL_CONS_COLUMNS cols \ - JOIN ALL_CONSTRAINTS cons ON cols.CONSTRAINT_NAME = cons.CONSTRAINT_NAME AND cols.OWNER = cons.OWNER \ - WHERE cons.CONSTRAINT_TYPE = 'P' AND cons.OWNER = {s} AND cons.TABLE_NAME = {t} \ - ) pk ON pk.COLUMN_NAME = c.COLUMN_NAME \ - WHERE c.OWNER = {s} AND c.TABLE_NAME = {t} \ - ORDER BY c.COLUMN_ID" - ), - &[], - ) - .await - .map_err(|e| e.to_string())?; - - Ok(col_result - .rows - .iter() - .map(|row| { - let name = row.get_string(0).unwrap_or("").to_string(); - let base = row.get_string(1).unwrap_or("").to_string(); - let data_len = row.get_i64(5).map(|v| v as i32); - let char_len = row.get_i64(6).map(|v| v as i32); - let num_prec = row.get_i64(3).map(|v| v as i32); - let num_scale = row.get_i64(4).map(|v| v as i32); - let data_type = match base.to_uppercase().as_str() { - "VARCHAR2" | "NVARCHAR2" | "CHAR" | "NCHAR" => { - let len = char_len.or(data_len); - match len { - Some(n) => format!("{base}({n})"), - None => base, - } - } - "NUMBER" => match (num_prec, num_scale) { - (Some(p), Some(s)) if s > 0 => format!("NUMBER({p},{s})"), - (Some(p), _) if p > 0 => format!("NUMBER({p})"), - _ => "NUMBER".to_string(), - }, - "RAW" => match data_len { - Some(n) => format!("RAW({n})"), - None => "RAW".to_string(), - }, - _ => base, - }; - ColumnInfo { - is_primary_key: row.get_i64(8).unwrap_or(0) == 1, - name, - data_type, - is_nullable: row.get_string(2).unwrap_or("N") == "Y", - column_default: None, - extra: None, - comment: row.get_string(7).filter(|s| !s.is_empty()).map(|s| s.to_string()), - numeric_precision: num_prec, - numeric_scale: num_scale, - character_maximum_length: char_len, - } - }) - .collect()) -} - -pub async fn get_table_comment(conn: &OracleClient, schema: &str, table: &str) -> Result, String> { - let s = quote_literal(schema); - let t = quote_literal(table); - let result = conn - .query(&format!("SELECT COMMENTS FROM ALL_TAB_COMMENTS WHERE OWNER = {s} AND TABLE_NAME = {t}"), &[]) - .await - .map_err(|e| e.to_string())?; - Ok(result.rows.first().and_then(|row| row.get_string(0)).filter(|s| !s.is_empty()).map(|s| s.to_string())) -} - -pub async fn list_indexes(conn: &OracleClient, schema: &str, table: &str) -> Result, String> { - let sql = format!( - "SELECT i.INDEX_NAME, \ - LISTAGG(ic.COLUMN_NAME, ',') WITHIN GROUP (ORDER BY ic.COLUMN_POSITION) AS columns, \ - i.UNIQUENESS, \ - CASE WHEN c.CONSTRAINT_TYPE = 'P' THEN 1 ELSE 0 END AS IS_PK, \ - i.INDEX_TYPE \ - FROM ALL_INDEXES i \ - JOIN ALL_IND_COLUMNS ic ON i.INDEX_NAME = ic.INDEX_NAME AND i.OWNER = ic.INDEX_OWNER AND i.TABLE_OWNER = ic.TABLE_OWNER \ - LEFT JOIN ALL_CONSTRAINTS c ON i.INDEX_NAME = c.INDEX_NAME AND i.TABLE_OWNER = c.OWNER \ - AND c.CONSTRAINT_TYPE = 'P' \ - WHERE i.TABLE_OWNER = '{s}' AND i.TABLE_NAME = '{t}' \ - GROUP BY i.INDEX_NAME, i.UNIQUENESS, c.CONSTRAINT_TYPE, i.INDEX_TYPE \ - ORDER BY i.INDEX_NAME", - s = schema.replace('\'', "''"), t = table.replace('\'', "''") - ); - let result = conn.query(&sql, &[]).await.map_err(|e| e.to_string())?; - Ok(result - .rows - .iter() - .map(|row| { - let cols_str = row.get_string(1).unwrap_or(""); - IndexInfo { - name: row.get_string(0).unwrap_or("").to_string(), - columns: cols_str.split(',').filter(|s| !s.is_empty()).map(|s| s.to_string()).collect(), - is_unique: row.get_string(2).unwrap_or("") == "UNIQUE", - is_primary: row.get_i64(3).unwrap_or(0) == 1, - filter: None, - index_type: row.get_string(4).map(|s| s.to_string()), - included_columns: None, - comment: None, - } - }) - .collect()) -} - -pub async fn list_foreign_keys(conn: &OracleClient, schema: &str, table: &str) -> Result, String> { - let sql = format!( - "SELECT c.CONSTRAINT_NAME, cc.COLUMN_NAME, rc.TABLE_NAME, rcc.COLUMN_NAME \ - FROM ALL_CONSTRAINTS c \ - JOIN ALL_CONS_COLUMNS cc ON c.CONSTRAINT_NAME = cc.CONSTRAINT_NAME AND c.OWNER = cc.OWNER \ - JOIN ALL_CONSTRAINTS rc ON c.R_CONSTRAINT_NAME = rc.CONSTRAINT_NAME AND c.R_OWNER = rc.OWNER \ - JOIN ALL_CONS_COLUMNS rcc ON rc.CONSTRAINT_NAME = rcc.CONSTRAINT_NAME AND rc.OWNER = rcc.OWNER \ - WHERE c.CONSTRAINT_TYPE = 'R' AND c.OWNER = '{s}' AND c.TABLE_NAME = '{t}' \ - ORDER BY c.CONSTRAINT_NAME", - s = schema.replace('\'', "''"), - t = table.replace('\'', "''") - ); - let result = conn.query(&sql, &[]).await.map_err(|e| e.to_string())?; - Ok(result - .rows - .iter() - .map(|row| ForeignKeyInfo { - name: row.get_string(0).unwrap_or("").to_string(), - column: row.get_string(1).unwrap_or("").to_string(), - ref_table: row.get_string(2).unwrap_or("").to_string(), - ref_column: row.get_string(3).unwrap_or("").to_string(), - }) - .collect()) -} - -pub async fn list_triggers(conn: &OracleClient, schema: &str, table: &str) -> Result, String> { - let sql = format!( - "SELECT TRIGGER_NAME, TRIGGERING_EVENT, TRIGGER_TYPE \ - FROM ALL_TRIGGERS \ - WHERE OWNER = '{s}' AND TABLE_NAME = '{t}' \ - ORDER BY TRIGGER_NAME", - s = schema.replace('\'', "''"), - t = table.replace('\'', "''") - ); - let result = conn.query(&sql, &[]).await.map_err(|e| e.to_string())?; - Ok(result - .rows - .iter() - .map(|row| TriggerInfo { - name: row.get_string(0).unwrap_or("").to_string(), - event: row.get_string(1).unwrap_or("").to_string(), - timing: row.get_string(2).unwrap_or("").to_string(), - }) - .collect()) -} - -pub async fn execute_query_with_schema(conn: &OracleClient, schema: &str, sql: &str) -> Result { - let set_schema = format!("ALTER SESSION SET CURRENT_SCHEMA = \"{}\"", schema); - log::info!("[oracle][set-schema:start] schema={schema}"); - conn.execute(&set_schema, &[]).await.map_err(|e| { - log::error!("[oracle] set current_schema failed: {e}"); - e.to_string() - })?; - log::info!("[oracle][set-schema:done] schema={schema}"); - execute_query(conn, sql).await -} - -pub async fn execute_query(conn: &OracleClient, sql: &str) -> Result { - let start = Instant::now(); - let sql = sql.trim().trim_end_matches(';'); - let explicit_limit = explicit_select_row_limit(sql); - log::info!("[oracle][execute:start] explicit_limit={:?} sql={}", explicit_limit, sql); - - // Rewrite FETCH FIRST N ROWS ONLY → ROWNUM for Oracle 11g compatibility. - let sql = rewrite_fetch_first(sql); - log::info!("[oracle][execute:rewritten] sql={}", sql.as_ref()); - - if starts_with_executable_sql_keyword(sql.as_ref(), &["SELECT", "WITH", "SHOW", "DESCRIBE", "EXPLAIN"]) { - let capped_sql = cap_select_rows(sql.as_ref()); - let query_limit = explicit_limit.unwrap_or(ORACLE_QUERY_LIMIT).min(ORACLE_QUERY_LIMIT); - log::info!( - "[oracle][query_with_limit:start] query_limit={} fetch_size=500 sql={}", - query_limit, - capped_sql.as_ref() - ); - let result = conn.query_with_limit(capped_sql.as_ref(), &[], query_limit, 500).await.map_err(|e| { - log::error!("[oracle] execute_query SELECT failed: {e}"); - e.to_string() - })?; - log::info!( - "[oracle][query_with_limit:done] column_count={} row_count={} has_more_rows={} elapsed_ms={}", - result.columns.len(), - result.rows.len(), - result.has_more_rows, - start.elapsed().as_millis() - ); - let columns: Vec = result.columns.iter().map(|c| c.name.clone()).collect(); - let mut rows: Vec> = result - .rows - .iter() - .map(|row| { - (0..columns.len()) - .map(|i| row.get(i).map(|v| value_to_json(v)).unwrap_or(serde_json::Value::Null)) - .collect() - }) - .collect(); - let truncated = rows.len() > crate::query::MAX_ROWS || result.has_more_rows; - if rows.len() > crate::query::MAX_ROWS { - rows.truncate(crate::query::MAX_ROWS); - } - - log::info!( - "[oracle][execute:done] column_count={} row_count={} truncated={} elapsed_ms={}", - columns.len(), - rows.len(), - truncated, - start.elapsed().as_millis() - ); - Ok(QueryResult { columns, rows, affected_rows: 0, execution_time_ms: start.elapsed().as_millis(), truncated }) - } else { - log::info!("[oracle][execute-non-select:start] sql={}", sql.as_ref()); - match conn.execute(sql.as_ref(), &[]).await { - Ok(result) => { - let _ = conn.commit().await; - log::info!( - "[oracle][execute-non-select:done] affected_rows={} elapsed_ms={}", - result.rows_affected, - start.elapsed().as_millis() - ); - Ok(QueryResult { - columns: vec![], - rows: vec![], - affected_rows: result.rows_affected, - execution_time_ms: start.elapsed().as_millis(), - truncated: false, - }) - } - Err(e) => { - let msg = e.to_string(); - log::error!("[oracle][execute-non-select:error] {msg}"); - if msg.contains("Server rejected") || msg.contains("closed the connection") { - Err(format!("Operation failed (connection closed). Original driver error: {msg}")) - } else { - Err(msg) - } - } - } - } -} - -fn cap_select_rows(sql: &str) -> std::borrow::Cow<'_, str> { - if !starts_with_executable_sql_keyword(sql, &["SELECT", "WITH"]) || has_for_update_clause(sql) { - return std::borrow::Cow::Borrowed(sql); - } - - std::borrow::Cow::Owned(format!("SELECT * FROM ({sql}) WHERE ROWNUM <= {ORACLE_QUERY_LIMIT}")) -} - -fn has_for_update_clause(sql: &str) -> bool { - sql.to_uppercase().contains(" FOR UPDATE") -} - -fn explicit_select_row_limit(sql: &str) -> Option { - fetch_first_row_limit(sql).or_else(|| rownum_row_limit(sql)) -} - -fn fetch_first_row_limit(sql: &str) -> Option { - let upper = sql.to_uppercase(); - let fetch_pos = upper.find("FETCH FIRST").or_else(|| upper.find("FETCH NEXT"))?; - let after_fetch = &upper[fetch_pos..]; - let end = after_fetch.find("ROWS ONLY")?; - let keyword_len = if after_fetch.starts_with("FETCH FIRST") { 11 } else { 10 }; - sql[fetch_pos + keyword_len..fetch_pos + end].trim().parse::().ok() -} - -fn rownum_row_limit(sql: &str) -> Option { - let upper = sql.to_uppercase(); - let mut rest = upper.as_str(); - let mut best: Option = None; - - while let Some(pos) = rest.find("ROWNUM") { - rest = &rest[pos + "ROWNUM".len()..]; - let trimmed = rest.trim_start(); - let value_start = if let Some(after) = trimmed.strip_prefix("<=") { - after.trim_start() - } else if let Some(after) = trimmed.strip_prefix('<') { - if let Some(n) = parse_leading_usize(after.trim_start()) { - let exclusive = n.saturating_sub(1); - best = Some(best.map_or(exclusive, |current| current.min(exclusive))); - } - continue; - } else { - continue; - }; - - if let Some(n) = parse_leading_usize(value_start) { - best = Some(best.map_or(n, |current| current.min(n))); - } - } - - best -} - -fn parse_leading_usize(value: &str) -> Option { - let digits: String = value.chars().take_while(|ch| ch.is_ascii_digit()).collect(); - if digits.is_empty() { - return None; - } - digits.parse().ok() -} - -fn rewrite_fetch_first(sql: &str) -> std::borrow::Cow<'_, str> { - let upper = sql.to_uppercase(); - // Match: ... [OFFSET M ROWS] FETCH FIRST|NEXT N ROWS ONLY - let fetch_pos = upper.find("FETCH FIRST").or_else(|| upper.find("FETCH NEXT")); - let Some(fpos) = fetch_pos else { return std::borrow::Cow::Borrowed(sql) }; - let after_fetch = &upper[fpos..]; - let Some(end) = after_fetch.find("ROWS ONLY") else { return std::borrow::Cow::Borrowed(sql) }; - let keyword_len = if after_fetch.starts_with("FETCH FIRST") { 11 } else { 10 }; - let between = sql[fpos + keyword_len..fpos + end].trim(); - let Ok(n) = between.parse::() else { return std::borrow::Cow::Borrowed(sql) }; - - // Check for OFFSET M ROWS before FETCH - let mut base = &sql[..fpos]; - let base_upper = base.to_uppercase(); - if let Some(opos) = base_upper.rfind("OFFSET ") { - let after_offset = base_upper[opos + 7..].trim(); - if let Some(rpos) = after_offset.find(" ROWS") { - let offset_str = after_offset[..rpos].trim(); - if let Ok(offset) = offset_str.parse::() { - let inner = sql[..opos].trim_end(); - return std::borrow::Cow::Owned(format!( - "SELECT * FROM (SELECT a.*, ROWNUM rn__ FROM ({inner}) a WHERE ROWNUM <= {}) WHERE rn__ > {offset}", - offset + n - )); - } - } - base = sql[..opos].trim_end(); - } else { - base = base.trim_end(); - } - - std::borrow::Cow::Owned(format!("SELECT * FROM ({base}) WHERE ROWNUM <= {n}")) -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn cap_select_rows_wraps_selects() { - let sql = "SELECT * FROM users ORDER BY id"; - assert_eq!(cap_select_rows(sql), format!("SELECT * FROM ({sql}) WHERE ROWNUM <= {ORACLE_QUERY_LIMIT}")); - } - - #[test] - fn cap_select_rows_wraps_ctes() { - let sql = "WITH recent AS (SELECT * FROM users) SELECT * FROM recent"; - assert_eq!(cap_select_rows(sql), format!("SELECT * FROM ({sql}) WHERE ROWNUM <= {ORACLE_QUERY_LIMIT}")); - } - - #[test] - fn cap_select_rows_keeps_for_update_queries() { - let sql = "SELECT * FROM users FOR UPDATE"; - assert_eq!(cap_select_rows(sql), sql); - } - - #[test] - fn rewrite_fetch_first_to_rownum() { - assert_eq!( - rewrite_fetch_first("SELECT * FROM users FETCH FIRST 20 ROWS ONLY"), - "SELECT * FROM (SELECT * FROM users) WHERE ROWNUM <= 20" - ); - } - - #[test] - fn explicit_select_row_limit_reads_fetch_first() { - assert_eq!(explicit_select_row_limit("SELECT * FROM users FETCH FIRST 100 ROWS ONLY"), Some(100)); - assert_eq!(explicit_select_row_limit("SELECT * FROM users OFFSET 20 ROWS FETCH NEXT 50 ROWS ONLY"), Some(50)); - } - - #[test] - fn explicit_select_row_limit_reads_rownum() { - assert_eq!(explicit_select_row_limit("SELECT * FROM users WHERE ROWNUM <= 100"), Some(100)); - assert_eq!(explicit_select_row_limit("SELECT * FROM users WHERE ROWNUM < 101"), Some(100)); - assert_eq!( - explicit_select_row_limit("SELECT * FROM (SELECT * FROM users WHERE ROWNUM <= 500) WHERE ROWNUM <= 100"), - Some(100) - ); - } -} diff --git a/crates/dbx-core/src/query.rs b/crates/dbx-core/src/query.rs index 1ae1c93f8..d22467f46 100644 --- a/crates/dbx-core/src/query.rs +++ b/crates/dbx-core/src/query.rs @@ -201,28 +201,6 @@ pub async fn do_execute( }; wait_for_query(cancel_token, db::sqlserver::execute_query(&mut client, sql)).await.map(truncate_result) } - PoolKind::Oracle(pool) => { - let client = pool.client(); - let schema = schema.map(|s| s.to_string()); - drop(connections); - log::info!("[query][oracle:lock:start] schema={:?} sql={}", schema, sql); - let client = match cancel_token.as_ref() { - Some(token) => tokio::select! { - biased; - _ = token.cancelled() => return Err(canceled_error()), - guard = client.lock() => guard, - }, - None => client.lock().await, - }; - log::info!("[query][oracle:lock:done] schema={:?}", schema); - if let Some(schema) = schema { - wait_for_query(cancel_token, db::oracle_driver::execute_query_with_schema(&*client, &schema, sql)) - .await - .map(truncate_result) - } else { - wait_for_query(cancel_token, db::oracle_driver::execute_query(&*client, sql)).await.map(truncate_result) - } - } PoolKind::Elasticsearch(client) => { let client = client.clone(); let sql = sql.to_string(); @@ -531,7 +509,6 @@ pub async fn execute_statements_in_transaction( PoolKind::DuckDb(_) | PoolKind::Redis(_) | PoolKind::MongoDb(_) - | PoolKind::Oracle(_) | PoolKind::Elasticsearch(_) | PoolKind::ExternalTabular(_) | PoolKind::ExternalDriver { .. } => TxPath::None, diff --git a/crates/dbx-core/src/schema.rs b/crates/dbx-core/src/schema.rs index 10c7d6779..f2c1553ae 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, MysqlMode, OraclePool, PoolKind}; +use crate::connection::{AppState, MysqlMode, PoolKind}; use crate::db; pub fn duckdb_query_tables(con: &duckdb::Connection) -> Result, String> { @@ -102,13 +102,6 @@ pub fn extract_clickhouse( } } -pub fn extract_oracle(connections: &HashMap, key: &str) -> Option> { - match connections.get(key)? { - PoolKind::Oracle(pool) => Some(pool.clone()), - _ => None, - } -} - pub fn extract_agent( connections: &HashMap, key: &str, @@ -152,12 +145,6 @@ pub async fn list_databases_core(state: &AppState, connection_id: &str) -> Resul let mut client = client.lock().await; return db::sqlserver::list_databases(&mut client).await; } - if let Some(pool) = extract_oracle(&connections, connection_id) { - drop(connections); - let client = pool.client(); - let client = client.lock().await; - return db::oracle_driver::list_databases(&*client).await; - } if let Some(client) = extract_agent(&connections, connection_id) { drop(connections); let mut client = client.lock().await; @@ -206,12 +193,6 @@ pub async fn list_schemas_core(state: &AppState, connection_id: &str, database: let mut client = client.lock().await; return db::sqlserver::list_schemas(&mut client).await; } - if let Some(pool) = extract_oracle(&connections, &pool_key) { - drop(connections); - let client = pool.client(); - let client = client.lock().await; - return db::oracle_driver::list_schemas(&*client).await; - } if let Some(client) = extract_agent(&connections, &pool_key) { drop(connections); let mut client = client.lock().await; @@ -278,12 +259,6 @@ pub async fn list_tables_core( let mut client = client.lock().await; return db::sqlserver::list_tables(&mut client, schema).await; } - if let Some(pool) = extract_oracle(&connections, &pool_key) { - drop(connections); - let client = pool.client(); - let client = client.lock().await; - return db::oracle_driver::list_tables(&*client, schema).await; - } if let Some(client) = extract_agent(&connections, &pool_key) { drop(connections); let mut client = client.lock().await; @@ -346,12 +321,6 @@ pub async fn list_objects_core( let mut client = client.lock().await; return db::sqlserver::list_objects(&mut client, schema).await; } - if let Some(pool) = extract_oracle(&connections, &pool_key) { - drop(connections); - let client = pool.client(); - let client = client.lock().await; - return db::oracle_driver::list_objects(&*client, schema).await; - } } Ok(list_tables_core(state, connection_id, database, schema) @@ -418,12 +387,6 @@ pub async fn get_columns_core( let mut client = client.lock().await; return db::sqlserver::get_columns(&mut client, schema, table).await; } - if let Some(pool) = extract_oracle(&connections, &pool_key) { - drop(connections); - let client = pool.client(); - let client = client.lock().await; - return db::oracle_driver::get_columns(&*client, schema, table).await; - } if let Some(client) = extract_agent(&connections, &pool_key) { drop(connections); let mut client = client.lock().await; @@ -469,12 +432,6 @@ pub async fn list_indexes_core( let mut client = client.lock().await; return db::sqlserver::list_indexes(&mut client, schema, table).await; } - if let Some(pool) = extract_oracle(&connections, &pool_key) { - drop(connections); - let client = pool.client(); - let client = client.lock().await; - return db::oracle_driver::list_indexes(&*client, schema, table).await; - } if let Some(client) = extract_agent(&connections, &pool_key) { drop(connections); let mut client = client.lock().await; @@ -520,12 +477,6 @@ pub async fn list_foreign_keys_core( let mut client = client.lock().await; return db::sqlserver::list_foreign_keys(&mut client, schema, table).await; } - if let Some(pool) = extract_oracle(&connections, &pool_key) { - drop(connections); - let client = pool.client(); - let client = client.lock().await; - return db::oracle_driver::list_foreign_keys(&*client, schema, table).await; - } if let Some(client) = extract_agent(&connections, &pool_key) { drop(connections); let mut client = client.lock().await; @@ -571,12 +522,6 @@ pub async fn list_triggers_core( let mut client = client.lock().await; return db::sqlserver::list_triggers(&mut client, schema, table).await; } - if let Some(pool) = extract_oracle(&connections, &pool_key) { - drop(connections); - let client = pool.client(); - let client = client.lock().await; - return db::oracle_driver::list_triggers(&*client, schema, table).await; - } if let Some(client) = extract_agent(&connections, &pool_key) { drop(connections); let mut client = client.lock().await; @@ -648,12 +593,6 @@ pub async fn get_table_ddl_core( let mut client = client.lock().await; return build_sqlserver_ddl(&mut client, schema, table).await; } - if let Some(pool) = extract_oracle(&connections, &pool_key) { - drop(connections); - let client = pool.client(); - let client = client.lock().await; - return build_oracle_ddl(&*client, schema, table).await; - } if let Some(client) = extract_agent(&connections, &pool_key) { drop(connections); let mut client = client.lock().await; @@ -808,14 +747,6 @@ pub async fn get_object_source_core( db::sqlserver::execute_query(&mut client, &sqlserver_object_source_sql(schema, name, &object_type)) .await?, )? - } else if let Some(pool) = extract_oracle(&connections, &pool_key) { - drop(connections); - let client = pool.client(); - let client = client.lock().await; - first_string_cell( - db::oracle_driver::execute_query(&*client, &oracle_object_source_sql(schema, name, &object_type)) - .await?, - )? } else if let Some(client) = extract_gaussdb(&connections, &pool_key) { drop(connections); let mut client = client.lock().await; @@ -1034,71 +965,6 @@ pub async fn build_sqlserver_ddl( Ok(ddl) } -pub async fn build_oracle_ddl( - client: &db::oracle_driver::OracleClient, - schema: &str, - table: &str, -) -> Result { - let columns = db::oracle_driver::get_columns(client, schema, table).await?; - let indexes = db::oracle_driver::list_indexes(client, schema, table).await?; - let fkeys = db::oracle_driver::list_foreign_keys(client, schema, table).await?; - let table_comment = db::oracle_driver::get_table_comment(client, schema, table).await?; - - let mut ddl = format!("CREATE TABLE \"{schema}\".\"{table}\" (\n"); - let col_lines: Vec = columns - .iter() - .map(|c| { - let mut line = format!(" \"{}\" {}", c.name, c.data_type); - if !c.is_nullable { - line.push_str(" NOT NULL"); - } - if let Some(ref def) = c.column_default { - line.push_str(&format!(" DEFAULT {def}")); - } - line - }) - .collect(); - ddl.push_str(&col_lines.join(",\n")); - - let pks: Vec<&str> = columns.iter().filter(|c| c.is_primary_key).map(|c| c.name.as_str()).collect(); - if !pks.is_empty() { - ddl.push_str(&format!( - ",\n PRIMARY KEY ({})", - pks.iter().map(|k| format!("\"{k}\"")).collect::>().join(", ") - )); - } - for fk in &fkeys { - ddl.push_str(&format!( - ",\n CONSTRAINT \"{}\" FOREIGN KEY (\"{}\") REFERENCES \"{}\"(\"{}\")", - fk.name, fk.column, fk.ref_table, fk.ref_column - )); - } - ddl.push_str("\n);\n"); - - if let Some(comment) = table_comment.as_deref().filter(|comment| !comment.trim().is_empty()) { - ddl.push_str(&format!("\nCOMMENT ON TABLE \"{schema}\".\"{table}\" IS '{}';", comment.replace('\'', "''"))); - } - for col in &columns { - if let Some(comment) = col.comment.as_deref().filter(|comment| !comment.trim().is_empty()) { - ddl.push_str(&format!( - "\nCOMMENT ON COLUMN \"{schema}\".\"{table}\".\"{}\" IS '{}';", - col.name, - comment.replace('\'', "''") - )); - } - } - - for idx in &indexes { - if idx.is_primary { - continue; - } - let unique = if idx.is_unique { "UNIQUE " } else { "" }; - let cols = idx.columns.iter().map(|c| format!("\"{c}\"")).collect::>().join(", "); - ddl.push_str(&format!("\nCREATE {unique}INDEX \"{}\" ON \"{schema}\".\"{table}\" ({cols});", idx.name)); - } - Ok(ddl) -} - pub async fn build_gaussdb_ddl( client: &mut db::gaussdb_driver::GaussdbClient, schema: &str, diff --git a/crates/dbx-core/src/transfer.rs b/crates/dbx-core/src/transfer.rs index bce5efc60..969b53a7f 100644 --- a/crates/dbx-core/src/transfer.rs +++ b/crates/dbx-core/src/transfer.rs @@ -501,12 +501,6 @@ pub async fn execute_on_pool(state: &AppState, pool_key: &str, sql: &str) -> Res let mut client = client.lock().await; db::sqlserver::execute_query(&mut client, sql).await } - PoolKind::Oracle(pool) => { - let client = pool.client(); - drop(connections); - let client = client.lock().await; - db::oracle_driver::execute_query(&*client, sql).await - } PoolKind::DuckDb(con) => { let con = con.clone(); let sql = sql.to_string(); @@ -641,15 +635,6 @@ pub async fn get_columns_for_transfer( let mut client = client.lock().await; return db::sqlserver::get_columns(&mut client, &schema, &table).await; } - if let Some(PoolKind::Oracle(pool)) = connections.get(pool_key) { - let client = pool.client(); - let schema = schema.to_string(); - let table = table.to_string(); - drop(connections); - let client = client.lock().await; - return db::oracle_driver::get_columns(&*client, &schema, &table).await; - } - let pool = connections.get(pool_key).ok_or("Pool not found")?; let schema = schema.to_string(); let table = table.to_string(); diff --git a/src-tauri/Cargo.toml b/src-tauri/Cargo.toml index 08b00aa33..b1d010930 100644 --- a/src-tauri/Cargo.toml +++ b/src-tauri/Cargo.toml @@ -42,7 +42,6 @@ tokio-util = { version = "0.7", features = ["compat"] } reqwest = { version = "0.12", features = ["json", "stream"] } futures = "0.3" mongodb = "3.2.5" -rust-oracle = { git = "https://github.com/t8y2/rust-oracle", branch = "main" } percent-encoding = "2" russh = "0.60" csv = "1.4.0" diff --git a/src-tauri/src/commands/agents.rs b/src-tauri/src/commands/agents.rs index cb76dc83f..df5244924 100644 --- a/src-tauri/src/commands/agents.rs +++ b/src-tauri/src/commands/agents.rs @@ -1,6 +1,6 @@ use std::sync::Arc; -use tauri::State; +use tauri::{Emitter, State}; use dbx_core::agent_manager::{AgentDriverInfo, AgentManager, AgentRegistry, InstalledDriver}; use dbx_core::connection::AppState; @@ -43,23 +43,46 @@ pub async fn list_installed_agents(state: State<'_, Arc>) -> Result>, db_type: String) -> Result<(), String> { +pub async fn install_agent( + app: tauri::AppHandle, + state: State<'_, Arc>, + db_type: String, +) -> Result<(), String> { let am = &state.agent_manager; let registry = fetch_registry().await?; + let needs_jre = !am.is_jre_installed(); - if !am.is_jre_installed() { + if needs_jre { let platform = AgentManager::current_platform(); let jre_info = registry.jre.platforms.get(platform).ok_or_else(|| format!("No JRE available for platform: {platform}"))?; let jre_archive = am.base_dir().join("jre-download.tar.gz"); - download_with_proxy(&jre_info.url, &jre_archive).await?; + let _ = app.emit( + "agent-install-progress", + serde_json::json!({ + "step": "jre", "downloaded": 0u64, "total": jre_info.size, + }), + ); + download_with_progress(&app, "jre", &jre_info.url, &jre_archive, jre_info.size).await?; + let _ = app.emit( + "agent-install-progress", + serde_json::json!({ + "step": "jre-extract", "downloaded": 0u64, "total": 0u64, + }), + ); extract_archive(&jre_archive, &am.base_dir().join("jre"))?; std::fs::remove_file(&jre_archive).ok(); } let driver = registry.drivers.get(&db_type).ok_or_else(|| format!("Unknown driver type: {db_type}"))?; let jar_path = am.driver_jar_path(&db_type); - download_with_proxy(&driver.jar.url, &jar_path).await?; + let _ = app.emit( + "agent-install-progress", + serde_json::json!({ + "step": "driver", "downloaded": 0u64, "total": driver.jar.size, + }), + ); + download_with_progress(&app, "driver", &driver.jar.url, &jar_path, driver.jar.size).await?; let mut local_state = am.load_state(); local_state.jre_version = Some(registry.jre.version.clone()); @@ -68,6 +91,7 @@ pub async fn install_agent(state: State<'_, Arc>, db_type: String) -> InstalledDriver { version: driver.version.clone(), installed_at: chrono::Utc::now().to_rfc3339() }, ); am.save_state(&local_state)?; + let _ = app.emit("agent-install-progress", serde_json::json!({ "step": "done" })); Ok(()) } @@ -94,7 +118,7 @@ pub async fn check_jre_installed(state: State<'_, Arc>) -> Result>) -> Result<(), String> { +pub async fn reinstall_jre(app: tauri::AppHandle, state: State<'_, Arc>) -> Result<(), String> { let am = &state.agent_manager; let jre_dir = am.base_dir().join("jre"); if jre_dir.exists() { @@ -105,12 +129,13 @@ pub async fn reinstall_jre(state: State<'_, Arc>) -> Result<(), String let jre_info = registry.jre.platforms.get(platform).ok_or_else(|| format!("No JRE available for platform: {platform}"))?; let jre_archive = am.base_dir().join("jre-download.tar.gz"); - download_with_proxy(&jre_info.url, &jre_archive).await?; + download_with_progress(&app, "jre", &jre_info.url, &jre_archive, jre_info.size).await?; extract_archive(&jre_archive, &jre_dir)?; std::fs::remove_file(&jre_archive).ok(); let mut local_state = am.load_state(); local_state.jre_version = Some(registry.jre.version.clone()); am.save_state(&local_state)?; + let _ = app.emit("agent-install-progress", serde_json::json!({ "step": "done" })); Ok(()) } @@ -140,7 +165,13 @@ async fn fetch_registry() -> Result { Err(format!("Failed to fetch agent registry: {last_err}")) } -async fn download_with_proxy(url: &str, dest: &std::path::Path) -> Result<(), String> { +async fn download_with_progress( + app: &tauri::AppHandle, + step: &str, + url: &str, + dest: &std::path::Path, + total_size: u64, +) -> Result<(), String> { if let Some(parent) = dest.parent() { std::fs::create_dir_all(parent).map_err(|e| e.to_string())?; } @@ -160,8 +191,18 @@ async fn download_with_proxy(url: &str, dest: &std::path::Path) -> Result<(), St .and_then(|r| r.error_for_status()) { Ok(resp) => { - let bytes = resp.bytes().await.map_err(|e| format!("Download read failed: {e}"))?; - std::fs::write(dest, &bytes).map_err(|e| format!("Failed to write file: {e}"))?; + let content_length = resp.content_length().unwrap_or(total_size); + let mut file = std::fs::File::create(dest).map_err(|e| format!("Failed to create file: {e}"))?; + let mut downloaded: u64 = 0; + let mut bytes = resp; + while let Some(chunk) = bytes.chunk().await.map_err(|e| format!("Download stream error: {e}"))? { + std::io::Write::write_all(&mut file, &chunk).map_err(|e| format!("Failed to write chunk: {e}"))?; + downloaded += chunk.len() as u64; + let _ = app.emit( + "agent-install-progress", + serde_json::json!({ "step": step, "downloaded": downloaded, "total": content_length }), + ); + } return Ok(()); } Err(e) => { diff --git a/src-tauri/src/commands/connection.rs b/src-tauri/src/commands/connection.rs index c38edd95c..22e32d77d 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, metadata_connection_config, probe_connection_endpoint, - redacted_connection_url_for_endpoint, AppState, MysqlMode, OraclePool, PoolKind, + redacted_connection_url_for_endpoint, AppState, MysqlMode, PoolKind, }; use dbx_core::db; use dbx_core::models::connection::{rewrite_jdbc_url_host, ConnectionConfig, DatabaseType}; @@ -97,22 +97,16 @@ pub async fn test_connection(state: State<'_, Arc>, config: Connection .await .map(|_| "Connection successful".to_string()) } - DatabaseType::Oracle => db::oracle_driver::connect( - &host, - port, - config.database.as_deref().unwrap_or("ORCL"), - &config.username, - &config.password, - config.sysdba, - ) - .await - .map(|_| "Connection successful".to_string()), DatabaseType::Elasticsearch => { let client = db::elasticsearch_driver::EsClient::new(&url, Some(&config.username), Some(&config.password)); db::elasticsearch_driver::test_connection(&client).await.map(|_| "Connection successful".to_string()) } - DatabaseType::Dameng | DatabaseType::Kingbase | DatabaseType::Vastbase | DatabaseType::Goldendb => { + DatabaseType::Dameng + | DatabaseType::Kingbase + | DatabaseType::Vastbase + | DatabaseType::Goldendb + | DatabaseType::Oracle => { state .agent_manager .call_daemon::( @@ -210,25 +204,17 @@ pub async fn connect_db(state: State<'_, Arc>, config: ConnectionConfi .await?; PoolKind::SqlServer(std::sync::Arc::new(tokio::sync::Mutex::new(client))) } - DatabaseType::Oracle => { - let client = db::oracle_driver::connect( - &host, - port, - db_config.database.as_deref().unwrap_or("ORCL"), - &db_config.username, - &db_config.password, - db_config.sysdba, - ) - .await?; - PoolKind::Oracle(std::sync::Arc::new(OraclePool::new(vec![client]))) - } DatabaseType::Elasticsearch => { let client = db::elasticsearch_driver::EsClient::new(&url, Some(&db_config.username), Some(&db_config.password)); db::elasticsearch_driver::test_connection(&client).await?; PoolKind::Elasticsearch(client) } - DatabaseType::Dameng | DatabaseType::Kingbase | DatabaseType::Vastbase | DatabaseType::Goldendb => { + DatabaseType::Dameng + | DatabaseType::Kingbase + | DatabaseType::Vastbase + | DatabaseType::Goldendb + | DatabaseType::Oracle => { let mut client = state.agent_manager.spawn(&db_config.db_type).await?; client .call::( @@ -280,7 +266,6 @@ pub async fn disconnect_db(state: State<'_, Arc>, connection_id: Strin PoolKind::MongoDb(_) => {} PoolKind::ClickHouse(_) => {} PoolKind::SqlServer(_) => {} - PoolKind::Oracle(_) => {} PoolKind::Elasticsearch(_) => {} PoolKind::Agent(_) => {} PoolKind::Gaussdb(_) => {} diff --git a/src/components/config/DriverManager.vue b/src/components/config/DriverManager.vue index 6afa67543..6320f1522 100644 --- a/src/components/config/DriverManager.vue +++ b/src/components/config/DriverManager.vue @@ -1,8 +1,13 @@ @@ -96,6 +149,17 @@ function formatSize(bytes: number): string { + +
+
{{ progressText }}
+
+
+
+
+
加载中...
@@ -106,6 +170,7 @@ function formatSize(bytes: number): string { v{{ driver.installed_version }} · {{ formatSize(driver.size) }} + 有更新
@@ -113,7 +178,21 @@ function formatSize(bytes: number): string { - +
diff --git a/src/lib/databaseCapabilities.ts b/src/lib/databaseCapabilities.ts index 47e36d5d5..25f3991cc 100644 --- a/src/lib/databaseCapabilities.ts +++ b/src/lib/databaseCapabilities.ts @@ -96,7 +96,7 @@ export const TREE_SCHEMA_TYPES = new Set(["postgres", "redshift", export const PG_LIKE_STRUCTURE_TYPES = new Set(["postgres", "redshift", "gaussdb"]); -export const AGENT_DRIVER_TYPES = new Set(["dameng", "kingbase", "vastbase", "goldendb"]); +export const AGENT_DRIVER_TYPES = new Set(["dameng", "kingbase", "vastbase", "goldendb", "oracle"]); export const TRANSFER_SQL_TYPES = new Set([ "mysql",