From c840ab4e4502f8c60b5bb7ad26f212d992637ac9 Mon Sep 17 00:00:00 2001 From: t8y2 <1156263951@qq.com> Date: Fri, 1 May 2026 00:00:59 +0800 Subject: [PATCH] feat: implement data transfer functionality with progress tracking --- src-tauri/Cargo.lock | 2 +- src-tauri/src/commands/mod.rs | 1 + src-tauri/src/commands/transfer.rs | 721 ++++++++++++++++++ src-tauri/src/lib.rs | 2 + src/App.vue | 18 + src/components/sidebar/TreeItem.vue | 53 +- .../transfer/DataTransferDialog.vue | 512 +++++++++++++ src/i18n/locales/en.ts | 33 + src/i18n/locales/zh-CN.ts | 33 + src/lib/tauri.ts | 52 ++ src/stores/connectionStore.ts | 2 + src/styles/globals.css | 2 + 12 files changed, 1412 insertions(+), 19 deletions(-) create mode 100644 src-tauri/src/commands/transfer.rs create mode 100644 src/components/transfer/DataTransferDialog.vue diff --git a/src-tauri/Cargo.lock b/src-tauri/Cargo.lock index baa77ad145..c8ca776481 100644 --- a/src-tauri/Cargo.lock +++ b/src-tauri/Cargo.lock @@ -1111,7 +1111,7 @@ checksum = "a4ae5f15dda3c708c0ade84bfee31ccab44a3da4f88015ed22f63732abe300c8" [[package]] name = "dbx" -version = "0.2.1" +version = "0.3.0" dependencies = [ "anyhow", "chrono", diff --git a/src-tauri/src/commands/mod.rs b/src-tauri/src/commands/mod.rs index 4c62e8d2bd..091d063d1f 100644 --- a/src-tauri/src/commands/mod.rs +++ b/src-tauri/src/commands/mod.rs @@ -5,4 +5,5 @@ pub mod mongo_cmd; pub mod query; pub mod redis_cmd; pub mod schema; +pub mod transfer; pub mod update; diff --git a/src-tauri/src/commands/transfer.rs b/src-tauri/src/commands/transfer.rs new file mode 100644 index 0000000000..e2f94e0dfd --- /dev/null +++ b/src-tauri/src/commands/transfer.rs @@ -0,0 +1,721 @@ +use std::collections::HashSet; +use std::sync::Arc; +use serde::{Deserialize, Serialize}; +use tauri::{AppHandle, Emitter, State}; +use tokio::sync::RwLock; + +use crate::commands::connection::{AppState, PoolKind}; +use crate::db; +use crate::models::connection::DatabaseType; + +static CANCELLED: std::sync::LazyLock>> = + std::sync::LazyLock::new(|| RwLock::new(HashSet::new())); + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct TransferRequest { + pub transfer_id: String, + pub source_connection_id: String, + pub source_database: String, + pub source_schema: String, + pub target_connection_id: String, + pub target_database: String, + pub target_schema: String, + pub tables: Vec, + pub create_table: bool, + pub truncate_before: bool, + pub batch_size: usize, +} + +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct TransferProgress { + pub transfer_id: String, + pub table: String, + pub table_index: usize, + pub total_tables: usize, + pub rows_transferred: u64, + pub total_rows: Option, + pub status: TransferStatus, + pub error: Option, +} + +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase")] +pub enum TransferStatus { + Running, + TableDone, + Done, + Error, + Cancelled, +} + +fn quote_identifier(name: &str, db_type: &DatabaseType) -> String { + match db_type { + DatabaseType::Mysql | DatabaseType::ClickHouse => format!("`{}`", name.replace('`', "``")), + DatabaseType::SqlServer => format!("[{}]", name.replace(']', "]]")), + _ => format!("\"{}\"", name.replace('"', "\"\"")), + } +} + +fn qualified_table(table: &str, schema: &str, db_type: &DatabaseType) -> String { + let qt = quote_identifier(table, db_type); + if schema.is_empty() { + qt + } else { + format!("{}.{}", quote_identifier(schema, db_type), qt) + } +} + +fn escape_value(val: &serde_json::Value, db_type: &DatabaseType) -> String { + match val { + serde_json::Value::Null => "NULL".to_string(), + serde_json::Value::Bool(b) => match db_type { + DatabaseType::Mysql | DatabaseType::Sqlite | DatabaseType::DuckDb => { + if *b { "1".to_string() } else { "0".to_string() } + } + _ => if *b { "TRUE".to_string() } else { "FALSE".to_string() }, + }, + serde_json::Value::Number(n) => n.to_string(), + serde_json::Value::String(s) => { + format!("'{}'", s.replace('\'', "''")) + } + _ => { + let s = val.to_string(); + format!("'{}'", s.replace('\'', "''")) + } + } +} + +fn map_column_type(source_type: &str, _source_db: &DatabaseType, target_db: &DatabaseType) -> String { + let t = source_type.to_lowercase(); + let base = t.split('(').next().unwrap_or(&t).trim(); + + match base { + "int" | "integer" | "int4" | "mediumint" => match target_db { + DatabaseType::Postgres => "INTEGER".into(), + DatabaseType::Mysql => "INT".into(), + DatabaseType::SqlServer => "INT".into(), + _ => "INTEGER".into(), + }, + "bigint" | "int8" => "BIGINT".into(), + "smallint" | "int2" => "SMALLINT".into(), + "tinyint" => match target_db { + DatabaseType::Postgres => "SMALLINT".into(), + _ => "TINYINT".into(), + }, + "serial" | "bigserial" | "smallserial" => match target_db { + DatabaseType::Postgres => source_type.to_uppercase(), + DatabaseType::Mysql => "BIGINT AUTO_INCREMENT".into(), + _ => "INTEGER".into(), + }, + "float" | "float4" | "real" => match target_db { + DatabaseType::Postgres => "REAL".into(), + _ => "FLOAT".into(), + }, + "double" | "double precision" | "float8" => match target_db { + DatabaseType::Postgres => "DOUBLE PRECISION".into(), + _ => "DOUBLE".into(), + }, + "decimal" | "numeric" | "number" => { + if t.contains('(') { + match target_db { + DatabaseType::Mysql | DatabaseType::Postgres | DatabaseType::SqlServer | DatabaseType::Oracle => { + format!("DECIMAL{}", &t[t.find('(').unwrap()..]) + } + _ => "NUMERIC".into(), + } + } else { + "NUMERIC".into() + } + } + "varchar" | "nvarchar" | "character varying" | "varchar2" => { + if t.contains('(') { + let len_part = &t[t.find('(').unwrap()..]; + match target_db { + DatabaseType::Postgres => format!("VARCHAR{len_part}"), + DatabaseType::Mysql => format!("VARCHAR{len_part}"), + DatabaseType::SqlServer => format!("NVARCHAR{len_part}"), + _ => format!("VARCHAR{len_part}"), + } + } else { + "VARCHAR(255)".into() + } + } + "char" | "nchar" | "character" => { + if t.contains('(') { + let len_part = &t[t.find('(').unwrap()..]; + format!("CHAR{len_part}") + } else { + "CHAR(1)".into() + } + } + "text" | "longtext" | "mediumtext" | "tinytext" | "clob" | "ntext" => "TEXT".into(), + "bool" | "boolean" => match target_db { + DatabaseType::Mysql => "TINYINT(1)".into(), + DatabaseType::SqlServer => "BIT".into(), + _ => "BOOLEAN".into(), + }, + "date" => "DATE".into(), + "time" => "TIME".into(), + "datetime" => match target_db { + DatabaseType::Postgres => "TIMESTAMP".into(), + _ => "DATETIME".into(), + }, + "timestamp" | "timestamptz" | "timestamp with time zone" + | "timestamp without time zone" => match target_db { + DatabaseType::Mysql => "DATETIME".into(), + DatabaseType::SqlServer => "DATETIME2".into(), + _ => "TIMESTAMP".into(), + }, + "blob" | "longblob" | "mediumblob" | "tinyblob" | "binary" | "varbinary" | "image" => { + match target_db { + DatabaseType::Postgres => "BYTEA".into(), + DatabaseType::Mysql => "BLOB".into(), + DatabaseType::SqlServer => "VARBINARY(MAX)".into(), + _ => "BLOB".into(), + } + } + "bytea" => match target_db { + DatabaseType::Postgres => "BYTEA".into(), + DatabaseType::Mysql => "BLOB".into(), + _ => "BLOB".into(), + }, + "json" | "jsonb" => match target_db { + DatabaseType::Postgres => "JSONB".into(), + DatabaseType::Mysql => "JSON".into(), + _ => "TEXT".into(), + }, + "uuid" => match target_db { + DatabaseType::Postgres => "UUID".into(), + _ => "VARCHAR(36)".into(), + }, + "bit" => match target_db { + DatabaseType::Postgres => "BOOLEAN".into(), + _ => "BIT".into(), + }, + _ => "TEXT".into(), + } +} + +fn generate_create_table_ddl( + columns: &[db::ColumnInfo], + table: &str, + schema: &str, + target_db: &DatabaseType, + source_db: &DatabaseType, +) -> String { + let full_table = qualified_table(table, schema, target_db); + + let col_lines: Vec = columns + .iter() + .map(|c| { + let mapped_type = map_column_type(&c.data_type, source_db, target_db); + let mut line = format!(" {} {}", quote_identifier(&c.name, target_db), mapped_type); + if !c.is_nullable { + line.push_str(" NOT NULL"); + } + line + }) + .collect(); + + let pks: Vec = columns + .iter() + .filter(|c| c.is_primary_key) + .map(|c| quote_identifier(&c.name, target_db)) + .collect(); + + let mut ddl = match target_db { + DatabaseType::SqlServer => format!("IF NOT EXISTS (SELECT * FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_NAME = '{table}')\n"), + _ => String::new(), + }; + + let create_prefix = match target_db { + DatabaseType::SqlServer => "CREATE TABLE", + _ => "CREATE TABLE IF NOT EXISTS", + }; + + ddl.push_str(&format!("{create_prefix} {full_table} (\n")); + ddl.push_str(&col_lines.join(",\n")); + + if !pks.is_empty() { + ddl.push_str(&format!(",\n PRIMARY KEY ({})", pks.join(", "))); + } + + ddl.push_str("\n)"); + + if matches!(target_db, DatabaseType::ClickHouse) { + ddl.push_str(" ENGINE = MergeTree() ORDER BY tuple()"); + } + + ddl +} + +fn generate_insert( + columns: &[String], + rows: &[Vec], + table: &str, + schema: &str, + db_type: &DatabaseType, +) -> String { + if rows.is_empty() { + return String::new(); + } + + let full_table = qualified_table(table, schema, db_type); + let col_list = columns + .iter() + .map(|c| quote_identifier(c, db_type)) + .collect::>() + .join(", "); + + let value_rows: Vec = rows + .iter() + .map(|row| { + let vals: Vec = row.iter().map(|v| escape_value(v, db_type)).collect(); + format!("({})", vals.join(", ")) + }) + .collect(); + + format!("INSERT INTO {full_table} ({col_list}) VALUES\n{}", value_rows.join(",\n")) +} + +fn pagination_sql( + columns: &[String], + table: &str, + schema: &str, + db_type: &DatabaseType, + offset: u64, + limit: usize, +) -> String { + let full_table = qualified_table(table, schema, db_type); + let col_list = columns + .iter() + .map(|c| quote_identifier(c, db_type)) + .collect::>() + .join(", "); + + match db_type { + DatabaseType::SqlServer | DatabaseType::Oracle => { + format!( + "SELECT {col_list} FROM {full_table} ORDER BY (SELECT NULL) OFFSET {offset} ROWS FETCH NEXT {limit} ROWS ONLY" + ) + } + _ => { + format!("SELECT {col_list} FROM {full_table} LIMIT {limit} OFFSET {offset}") + } + } +} + +fn count_sql(table: &str, schema: &str, db_type: &DatabaseType) -> String { + let full_table = qualified_table(table, schema, db_type); + format!("SELECT COUNT(*) FROM {full_table}") +} + +async fn execute_on_pool( + state: &AppState, + pool_key: &str, + sql: &str, +) -> Result { + let connections = state.connections.lock().await; + let pool = connections.get(pool_key).ok_or("Connection not found")?; + + match pool { + PoolKind::Mysql(p) => { + let p = p.clone(); + drop(connections); + db::mysql::execute_query(&p, sql).await + } + PoolKind::Postgres(p) => { + let p = p.clone(); + drop(connections); + db::postgres::execute_query(&p, sql).await + } + PoolKind::Sqlite(p) => { + let p = p.clone(); + drop(connections); + db::sqlite::execute_query(&p, sql).await + } + PoolKind::ClickHouse(client) => { + let client = client.clone(); + let database = pool_key.split(':').nth(1).unwrap_or("default").to_string(); + drop(connections); + db::clickhouse_driver::execute_query(&client, &database, sql).await + } + PoolKind::SqlServer(client) => { + let client = client.clone(); + drop(connections); + let mut client = client.lock().await; + db::sqlserver::execute_query(&mut client, sql).await + } + PoolKind::Oracle(client) => { + let client = client.clone(); + 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(); + drop(connections); + tokio::task::spawn_blocking(move || { + let con = con.lock().map_err(|e| e.to_string())?; + let start = std::time::Instant::now(); + let trimmed = sql.trim().to_uppercase(); + if trimmed.starts_with("SELECT") || trimmed.starts_with("SHOW") + || trimmed.starts_with("DESCRIBE") || trimmed.starts_with("WITH") + || trimmed.starts_with("PRAGMA") + { + let mut stmt = con.prepare(&sql).map_err(|e| e.to_string())?; + let mut rows = stmt.query([]).map_err(|e| e.to_string())?; + let stmt_ref = rows.as_ref().ok_or("DuckDB statement unavailable")?; + let col_count = stmt_ref.column_count(); + let columns: Vec = (0..col_count) + .map(|i| stmt_ref.column_name(i).map(|s| s.to_string()).unwrap_or_else(|_| "?".to_string())) + .collect(); + let mut result_rows = Vec::new(); + while let Some(row) = rows.next().map_err(|e| e.to_string())? { + let vals: Vec = (0..col_count).map(|i| { + row.get::<_, String>(i).map(serde_json::Value::String) + .or_else(|_| row.get::<_, i64>(i).map(|v| serde_json::Value::Number(v.into()))) + .or_else(|_| row.get::<_, f64>(i).map(|v| { + serde_json::Number::from_f64(v).map(serde_json::Value::Number).unwrap_or(serde_json::Value::Null) + })) + .or_else(|_| row.get::<_, bool>(i).map(serde_json::Value::Bool)) + .unwrap_or(serde_json::Value::Null) + }).collect(); + result_rows.push(vals); + } + Ok(db::QueryResult { columns, rows: result_rows, affected_rows: 0, execution_time_ms: start.elapsed().as_millis(), truncated: false }) + } else { + let affected = con.execute(&sql, []).map_err(|e| e.to_string())?; + Ok(db::QueryResult { columns: vec![], rows: vec![], affected_rows: affected as u64, execution_time_ms: start.elapsed().as_millis(), truncated: false }) + } + }) + .await + .map_err(|e| e.to_string())? + } + _ => Err("Unsupported database type for transfer".to_string()), + } +} + +async fn get_db_type(state: &AppState, connection_id: &str) -> Result { + let configs = state.configs.lock().await; + configs + .get(connection_id) + .map(|c| c.db_type.clone()) + .ok_or_else(|| format!("Connection config not found: {connection_id}")) +} + +async fn get_columns_for_transfer( + state: &AppState, + pool_key: &str, + _connection_id: &str, + database: &str, + schema: &str, + table: &str, +) -> Result, String> { + let connections = state.connections.lock().await; + + if let Some(PoolKind::DuckDb(con)) = connections.get(pool_key) { + let con = con.clone(); + drop(connections); + let table = table.to_string(); + return tokio::task::spawn_blocking(move || { + let con = con.lock().map_err(|e| e.to_string())?; + // Inline DuckDB column query + let mut stmt = con.prepare( + "SELECT column_name, data_type, is_nullable, column_default + FROM information_schema.columns + WHERE table_schema = 'main' AND table_name = ? + ORDER BY ordinal_position" + ).map_err(|e| e.to_string())?; + let rows = stmt.query_map([&table], |row| { + Ok(db::ColumnInfo { + name: row.get::<_, String>(0)?, + data_type: row.get::<_, String>(1)?, + is_nullable: row.get::<_, String>(2).unwrap_or_default() == "YES", + column_default: row.get::<_, Option>(3)?, + is_primary_key: false, + extra: None, + comment: None, + }) + }).map_err(|e| e.to_string())?; + Ok(rows.filter_map(|r| r.ok()).collect()) + }).await.map_err(|e| e.to_string())?; + } + + if let Some(PoolKind::ClickHouse(client)) = connections.get(pool_key) { + let client = client.clone(); + let database = database.to_string(); + let table = table.to_string(); + drop(connections); + return db::clickhouse_driver::get_columns(&client, &database, &table).await; + } + if let Some(PoolKind::SqlServer(client)) = connections.get(pool_key) { + let client = client.clone(); + let schema = schema.to_string(); + let table = table.to_string(); + drop(connections); + let mut client = client.lock().await; + return db::sqlserver::get_columns(&mut client, &schema, &table).await; + } + if let Some(PoolKind::Oracle(client)) = connections.get(pool_key) { + let client = client.clone(); + 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(); + match pool { + PoolKind::Mysql(p) => { + let p = p.clone(); + drop(connections); + db::mysql::get_columns(&p, &schema, &table).await + } + PoolKind::Postgres(p) => { + let p = p.clone(); + drop(connections); + db::postgres::get_columns(&p, &schema, &table).await + } + PoolKind::Sqlite(p) => { + let p = p.clone(); + drop(connections); + db::sqlite::get_columns(&p, &schema, &table).await + } + _ => Err("Unsupported database type".to_string()), + } +} + +fn emit_progress(app: &AppHandle, progress: TransferProgress) { + let _ = app.emit("transfer-progress", progress); +} + +async fn is_cancelled(transfer_id: &str) -> bool { + CANCELLED.read().await.contains(transfer_id) +} + +async fn transfer_table( + app: &AppHandle, + state: &AppState, + request: &TransferRequest, + table: &str, + table_index: usize, + source_db_type: &DatabaseType, + target_db_type: &DatabaseType, + source_pool_key: &str, + target_pool_key: &str, +) -> Result { + let total_tables = request.tables.len(); + + // Get source columns (deduplicate by name) + let columns = { + let raw = get_columns_for_transfer( + state, source_pool_key, &request.source_connection_id, + &request.source_database, &request.source_schema, table, + ).await?; + let mut seen = std::collections::HashSet::new(); + raw.into_iter().filter(|c| seen.insert(c.name.clone())).collect::>() + }; + + if columns.is_empty() { + return Err(format!("No columns found for table {table}")); + } + + let col_names: Vec = columns.iter().map(|c| c.name.clone()).collect(); + log::info!("[transfer] {} has {} columns, counting rows...", table, columns.len()); + + // Count source rows + let total_rows = { + let sql = count_sql(table, &request.source_schema, source_db_type); + match execute_on_pool(state, source_pool_key, &sql).await { + Ok(result) => result.rows.first() + .and_then(|r| r.first()) + .and_then(|v| match v { + serde_json::Value::Number(n) => n.as_u64(), + serde_json::Value::String(s) => s.parse::().ok(), + _ => None, + }), + Err(e) => { + log::warn!("[transfer] count failed for {}: {}", table, e); + None + } + } + }; + log::info!("[transfer] {} total_rows={:?}", table, total_rows); + + // Create table on target if requested + if request.create_table { + let ddl = generate_create_table_ddl(&columns, table, &request.target_schema, target_db_type, source_db_type); + log::info!("[transfer] creating target table: {}", &ddl[..ddl.len().min(200)]); + if let Err(e) = execute_on_pool(state, target_pool_key, &ddl).await { + let err_lower = e.to_lowercase(); + if !err_lower.contains("already exists") && !err_lower.contains("there is already") { + return Err(format!("Failed to create table: {e}")); + } + } + } + + // Truncate target if requested + if request.truncate_before { + let full_table = qualified_table(table, &request.target_schema, target_db_type); + let truncate_sql = match target_db_type { + DatabaseType::Sqlite | DatabaseType::DuckDb => format!("DELETE FROM {full_table}"), + _ => format!("TRUNCATE TABLE {full_table}"), + }; + execute_on_pool(state, target_pool_key, &truncate_sql).await + .map_err(|e| format!("Failed to truncate: {e}"))?; + } + + // Transfer data in batches + let batch_size = if request.batch_size == 0 { 1000 } else { request.batch_size }; + let mut offset: u64 = 0; + let mut total_transferred: u64 = 0; + + loop { + if is_cancelled(&request.transfer_id).await { + return Err("Cancelled".to_string()); + } + + let sql = pagination_sql(&col_names, table, &request.source_schema, source_db_type, offset, batch_size); + let result = execute_on_pool(state, source_pool_key, &sql).await?; + let row_count = result.rows.len(); + + if row_count == 0 { + break; + } + + let insert_sql = generate_insert(&col_names, &result.rows, table, &request.target_schema, target_db_type); + if !insert_sql.is_empty() { + execute_on_pool(state, target_pool_key, &insert_sql).await + .map_err(|e| format!("Insert failed at offset {offset}: {e}"))?; + } + + total_transferred += row_count as u64; + log::info!("[transfer] {} batch +{} rows (total {})", table, row_count, total_transferred); + offset += row_count as u64; + + emit_progress(app, TransferProgress { + transfer_id: request.transfer_id.clone(), + table: table.to_string(), + table_index, + total_tables, + rows_transferred: total_transferred, + total_rows, + status: TransferStatus::Running, + error: None, + }); + + if row_count < batch_size { + break; + } + } + + Ok(total_transferred) +} + +#[tauri::command] +pub async fn start_transfer( + app: AppHandle, + state: State<'_, Arc>, + request: TransferRequest, +) -> Result<(), String> { + let state = state.inner().clone(); + let transfer_id = request.transfer_id.clone(); + + // Validate connections exist + let source_db_type = get_db_type(&state, &request.source_connection_id).await?; + let target_db_type = get_db_type(&state, &request.target_connection_id).await?; + + // Ensure pools + let source_pool_key = state + .get_or_create_pool(&request.source_connection_id, Some(&request.source_database)) + .await?; + let target_pool_key = state + .get_or_create_pool(&request.target_connection_id, Some(&request.target_database)) + .await?; + + tokio::spawn(async move { + let total_tables = request.tables.len(); + log::info!("[transfer] starting transfer_id={} tables={}", transfer_id, total_tables); + + for (i, table) in request.tables.iter().enumerate() { + if is_cancelled(&transfer_id).await { + emit_progress(&app, TransferProgress { + transfer_id: transfer_id.clone(), + table: table.clone(), + table_index: i, + total_tables, + rows_transferred: 0, + total_rows: None, + status: TransferStatus::Cancelled, + error: None, + }); + CANCELLED.write().await.remove(&transfer_id); + return; + } + + log::info!("[transfer] table {}/{}: {}", i + 1, total_tables, table); + + match transfer_table( + &app, &state, &request, table, i, + &source_db_type, &target_db_type, + &source_pool_key, &target_pool_key, + ).await { + Ok(rows) => { + emit_progress(&app, TransferProgress { + transfer_id: transfer_id.clone(), + table: table.clone(), + table_index: i, + total_tables, + rows_transferred: rows, + total_rows: Some(rows), + status: if i == total_tables - 1 { TransferStatus::Done } else { TransferStatus::TableDone }, + error: None, + }); + } + Err(e) => { + if e == "Cancelled" { + emit_progress(&app, TransferProgress { + transfer_id: transfer_id.clone(), + table: table.clone(), + table_index: i, + total_tables, + rows_transferred: 0, + total_rows: None, + status: TransferStatus::Cancelled, + error: None, + }); + CANCELLED.write().await.remove(&transfer_id); + return; + } + emit_progress(&app, TransferProgress { + transfer_id: transfer_id.clone(), + table: table.clone(), + table_index: i, + total_tables, + rows_transferred: 0, + total_rows: None, + status: TransferStatus::Error, + error: Some(e), + }); + } + } + } + + CANCELLED.write().await.remove(&transfer_id); + }); + + Ok(()) +} + +#[tauri::command] +pub async fn cancel_transfer(transfer_id: String) -> Result<(), String> { + CANCELLED.write().await.insert(transfer_id); + Ok(()) +} diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 4bd9bbaf84..2094e812f3 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -77,6 +77,8 @@ pub fn run() { commands::history::clear_history, commands::history::delete_history_entry, commands::update::check_for_updates, + commands::transfer::start_transfer, + commands::transfer::cancel_transfer, ]) .run(tauri::generate_context!()) .expect("error while running tauri application"); diff --git a/src/App.vue b/src/App.vue index aa12baa9ea..d6a9709a19 100644 --- a/src/App.vue +++ b/src/App.vue @@ -33,6 +33,7 @@ import MongoDocBrowser from "@/components/mongo/MongoDocBrowser.vue"; import DatabaseIcon from "@/components/icons/DatabaseIcon.vue"; import QueryHistory from "@/components/editor/QueryHistory.vue"; import DangerConfirmDialog from "@/components/editor/DangerConfirmDialog.vue"; +import DataTransferDialog from "@/components/transfer/DataTransferDialog.vue"; import type { ConnectionConfig } from "@/types/database"; import { useConnectionStore } from "@/stores/connectionStore"; import { useQueryStore } from "@/stores/queryStore"; @@ -78,6 +79,9 @@ const pendingDangerSql = ref(""); const selectedSql = ref(""); const formatSqlRequestId = ref(0); const showDangerDialog = ref(false); +const showTransferDialog = ref(false); +const transferPrefillConnectionId = ref(""); +const transferPrefillDatabase = ref(""); const databaseOptions = ref>({}); const loadingDatabaseOptions = ref>({}); const checkingUpdates = ref(false); @@ -99,6 +103,15 @@ watch(showConnectionDialog, (v) => { if (!v) connectionStore.stopEditing(); }); +watch(() => connectionStore.transferSource, (v) => { + if (v) { + transferPrefillConnectionId.value = v.connectionId; + transferPrefillDatabase.value = v.database; + showTransferDialog.value = true; + connectionStore.transferSource = null; + } +}); + function onConnectionConnectStarted(name: string) { toast(t("connection.connecting", { name }), 30000); } @@ -1059,6 +1072,11 @@ async function setupFileDrop() { @connect-failed="onConnectionConnectFailed" /> + diff --git a/src/components/sidebar/TreeItem.vue b/src/components/sidebar/TreeItem.vue index ab26c4f467..26e83c1ea2 100644 --- a/src/components/sidebar/TreeItem.vue +++ b/src/components/sidebar/TreeItem.vue @@ -5,7 +5,7 @@ import { Database, Table, Columns3, Eye, ChevronRight, ChevronDown, Loader2, FolderOpen, Trash2, TerminalSquare, RefreshCw, Copy, TableProperties, Key, Link, Zap, ListTree, Pencil, Plug, Unplug, - Pin, + Pin, ArrowRightLeft, } from "lucide-vue-next"; import { ContextMenu, ContextMenuContent, ContextMenuItem, @@ -239,9 +239,22 @@ function disconnectConnection() { } } +function openTransfer() { + if (props.node.connectionId) { + connectionStore.transferSource = { + connectionId: props.node.connectionId, + database: props.node.database ?? "", + }; + } +} + const canExpand = !leafTypes.has(props.node.type); const canPin = computed(() => pinnableTypes.has(props.node.type)); const isPinned = computed(() => props.node.pinned || connectionStore.isTreeNodePinned(props.node.id)); +const hasTypeMenu = computed(() => { + const t = props.node.type; + return t === "connection" || t === "database" || t === "schema" || t === "table" || t === "view" || isGroupLabel(props.node); +}); const columnComment = computed(() => props.node.type === "column" && props.node.meta && "comment" in props.node.meta ? (props.node.meta as any).comment : null); const paddingLeft = `${props.depth * 16 + 8}px`; const isConnected = computed(() => @@ -329,67 +342,71 @@ async function showMore() { - + - + {{ isPinned ? t('contextMenu.unpin') : t('contextMenu.pin') }} - + - + - {{ t('contextMenu.copyName') }} + {{ t('contextMenu.copyName') }} diff --git a/src/components/transfer/DataTransferDialog.vue b/src/components/transfer/DataTransferDialog.vue new file mode 100644 index 0000000000..480b3a3d89 --- /dev/null +++ b/src/components/transfer/DataTransferDialog.vue @@ -0,0 +1,512 @@ + + + diff --git a/src/i18n/locales/en.ts b/src/i18n/locales/en.ts index 394bf05b67..ea89b0acec 100644 --- a/src/i18n/locales/en.ts +++ b/src/i18n/locales/en.ts @@ -123,6 +123,7 @@ export default { common: { language: "Language", loading: "Loading...", + close: "Close", }, ai: { placeholder: "Describe your query in natural language...", @@ -235,4 +236,36 @@ export default { cancel: "Cancel", confirm: "Execute", }, + transfer: { + title: "Data Transfer", + source: "Source", + target: "Target", + sourceConnection: "Source Connection", + sourceDatabase: "Source Database", + targetConnection: "Target Connection", + targetDatabase: "Target Database", + tables: "Tables", + selectAll: "Select All", + deselectAll: "Deselect All", + createTable: "Create table if not exists", + truncateBefore: "Truncate target table before insert", + batchSize: "Batch size", + start: "Start Transfer", + cancel: "Cancel", + progress: "Transferring...", + transferring: "Transferring {table}...", + completed: "Transfer completed", + failed: "Transfer failed", + cancelled: "Transfer cancelled", + rowsTransferred: "{count} rows transferred", + tableProgress: "{current} / {total} tables", + noSqlConnections: "No SQL connections available", + selectTables: "Select tables to transfer", + searchTables: "Search tables...", + noTables: "No tables", + selectSourceFirst: "Select source connection and database first", + tableError: "Table {table} failed: {error}", + overallProgress: "Overall progress", + dataTransfer: "Data Transfer", + }, }; diff --git a/src/i18n/locales/zh-CN.ts b/src/i18n/locales/zh-CN.ts index b2909a7fef..c46f2b4f6b 100644 --- a/src/i18n/locales/zh-CN.ts +++ b/src/i18n/locales/zh-CN.ts @@ -123,6 +123,7 @@ export default { common: { language: "语言", loading: "加载中...", + close: "关闭", }, ai: { placeholder: "描述你想查什么...", @@ -240,4 +241,36 @@ export default { cancel: "取消", confirm: "执行", }, + transfer: { + title: "数据传输", + source: "源", + target: "目标", + sourceConnection: "源连接", + sourceDatabase: "源数据库", + targetConnection: "目标连接", + targetDatabase: "目标数据库", + tables: "表", + selectAll: "全选", + deselectAll: "取消全选", + createTable: "目标表不存在时自动建表", + truncateBefore: "插入前清空目标表", + batchSize: "每批行数", + start: "开始传输", + cancel: "取消", + progress: "传输中...", + transferring: "正在传输 {table}...", + completed: "传输完成", + failed: "传输失败", + cancelled: "传输已取消", + rowsTransferred: "已传输 {count} 行", + tableProgress: "{current} / {total} 张表", + noSqlConnections: "暂无可用的 SQL 连接", + selectTables: "选择要传输的表", + searchTables: "搜索表名...", + noTables: "暂无表", + selectSourceFirst: "请先选择源连接和数据库", + tableError: "表 {table} 传输失败：{error}", + overallProgress: "整体进度", + dataTransfer: "数据传输", + }, }; diff --git a/src/lib/tauri.ts b/src/lib/tauri.ts index 7a9e8abc40..6745913bec 100644 --- a/src/lib/tauri.ts +++ b/src/lib/tauri.ts @@ -261,3 +261,55 @@ export async function clearHistory(): Promise { export async function deleteHistoryEntry(id: string): Promise { return invoke("delete_history_entry", { id }); } + +// --- Data Transfer --- +export interface TransferRequest { + transferId: string; + sourceConnectionId: string; + sourceDatabase: string; + sourceSchema: string; + targetConnectionId: string; + targetDatabase: string; + targetSchema: string; + tables: string[]; + createTable: boolean; + truncateBefore: boolean; + batchSize: number; +} + +export interface TransferProgress { + transferId: string; + table: string; + tableIndex: number; + totalTables: number; + rowsTransferred: number; + totalRows: number | null; + status: "running" | "tableDone" | "done" | "error" | "cancelled"; + error: string | null; +} + +export async function startTransfer( + request: TransferRequest, + onProgress: (progress: TransferProgress) => void, +): Promise { + const unlisten: UnlistenFn = await listen("transfer-progress", (event) => { + if (event.payload.transferId === request.transferId) { + onProgress(event.payload); + if (event.payload.status === "done" || event.payload.status === "error" || event.payload.status === "cancelled") { + if (event.payload.tableIndex === event.payload.totalTables - 1 || event.payload.status !== "error") { + unlisten(); + } + } + } + }); + try { + await invoke("start_transfer", { request }); + } catch (e) { + unlisten(); + throw e; + } +} + +export async function cancelTransfer(transferId: string): Promise { + return invoke("cancel_transfer", { transferId }); +} diff --git a/src/stores/connectionStore.ts b/src/stores/connectionStore.ts index aade51b918..b54758b967 100644 --- a/src/stores/connectionStore.ts +++ b/src/stores/connectionStore.ts @@ -16,6 +16,7 @@ export const useConnectionStore = defineStore("connection", () => { const editingConnectionId = ref(null); const completionTablesCache = ref>({}); const completionColumnsCache = ref>({}); + const transferSource = ref<{ connectionId: string; database: string } | null>(null); function startEditing(id: string) { editingConnectionId.value = id; @@ -640,5 +641,6 @@ export const useConnectionStore = defineStore("connection", () => { listCompletionColumns, exportConnectionsToFile, importConnectionsFromFile, + transferSource, }; }); diff --git a/src/styles/globals.css b/src/styles/globals.css index e7a57e9ea3..ef4f7b09a1 100644 --- a/src/styles/globals.css +++ b/src/styles/globals.css @@ -136,8 +136,10 @@ [role="switch"], select, summary, + label, a[href] { cursor: pointer; + user-select: none; } }