feat: implement data transfer functionality with progress tracking

src-tauri/Cargo.lock

@@ -1111,7 +1111,7 @@ checksum = "a4ae5f15dda3c708c0ade84bfee31ccab44a3da4f88015ed22f63732abe300c8"
 
 [[package]]
 name = "dbx"
-version = "0.2.1"
+version = "0.3.0"
 dependencies = [
  "anyhow",
  "chrono",

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;

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<RwLock<HashSet<String>>> =
+    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<String>,
+    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<u64>,
+    pub status: TransferStatus,
+    pub error: Option<String>,
+}
+
+#[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<String> = 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<String> = 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<serde_json::Value>],
+    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::<Vec<_>>()
+        .join(", ");
+
+    let value_rows: Vec<String> = rows
+        .iter()
+        .map(|row| {
+            let vals: Vec<String> = 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::<Vec<_>>()
+        .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<db::QueryResult, String> {
+    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<String> = (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<serde_json::Value> = (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<DatabaseType, String> {
+    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<Vec<db::ColumnInfo>, 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<String>>(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<u64, String> {
+    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::<Vec<_>>()
+    };
+
+    if columns.is_empty() {
+        return Err(format!("No columns found for table {table}"));
+    }
+
+    let col_names: Vec<String> = 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::<u64>().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<AppState>>,
+    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(())
+}

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");

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<Record<string, string[]>>({});
 const loadingDatabaseOptions = ref<Record<string, boolean>>({});
 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"
       />
       <DangerConfirmDialog v-model:open="showDangerDialog" :sql="dangerSql" @confirm="onDangerConfirm" />
+      <DataTransferDialog
+        v-model:open="showTransferDialog"
+        :prefill-connection-id="transferPrefillConnectionId"
+        :prefill-database="transferPrefillDatabase"
+      />
       <Dialog v-model:open="showUpdateDialog">
         <DialogContent class="sm:max-w-[520px]">
           <DialogHeader>

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() {
       </div>
     </ContextMenuTrigger>
 
-    <ContextMenuContent class="w-48">
+    <ContextMenuContent class="w-auto min-w-36">
       <ContextMenuItem v-if="canPin" @click="togglePin">
-        <Pin class="w-3.5 h-3.5 mr-2" :class="{ 'fill-current': isPinned }" />
+        <Pin class="w-4 h-4" :class="{ 'fill-current': isPinned }" />
         {{ isPinned ? t('contextMenu.unpin') : t('contextMenu.pin') }}
       </ContextMenuItem>
-      <ContextMenuSeparator v-if="canPin" />
+      <ContextMenuSeparator v-if="canPin && hasTypeMenu" />
 
       <template v-if="node.type === 'connection'">
         <ContextMenuItem v-if="!isConnected" @click="toggle">
-          <Plug class="w-3.5 h-3.5 mr-2" /> {{ t('contextMenu.openConnection') }}
+          <Plug class="w-4 h-4" /> {{ t('contextMenu.openConnection') }}
         </ContextMenuItem>
         <ContextMenuItem v-else @click="disconnectConnection">
-          <Unplug class="w-3.5 h-3.5 mr-2" /> {{ t('contextMenu.closeConnection') }}
+          <Unplug class="w-4 h-4" /> {{ t('contextMenu.closeConnection') }}
         </ContextMenuItem>
         <ContextMenuItem @click="newQuery">
-          <TerminalSquare class="w-3.5 h-3.5 mr-2" /> {{ t('contextMenu.newQuery') }}
+          <TerminalSquare class="w-4 h-4" /> {{ t('contextMenu.newQuery') }}
         </ContextMenuItem>
         <ContextMenuSeparator />
         <ContextMenuItem @click="refresh">
-          <RefreshCw class="w-3.5 h-3.5 mr-2" /> {{ t('contextMenu.refreshChildren') }}
+          <RefreshCw class="w-4 h-4" /> {{ t('contextMenu.refreshChildren') }}
         </ContextMenuItem>
         <ContextMenuItem @click="editConnection">
-          <Pencil class="w-3.5 h-3.5 mr-2" /> {{ t('contextMenu.editConnection') }}
+          <Pencil class="w-4 h-4" /> {{ t('contextMenu.editConnection') }}
         </ContextMenuItem>
         <ContextMenuSeparator />
         <ContextMenuItem class="text-destructive" @click="deleteConnection">
-          <Trash2 class="w-3.5 h-3.5 mr-2" /> {{ t('contextMenu.deleteConnection') }}
+          <Trash2 class="w-4 h-4" /> {{ t('contextMenu.deleteConnection') }}
         </ContextMenuItem>
       </template>
 
       <template v-if="node.type === 'database' || node.type === 'schema'">
         <ContextMenuItem @click="newQuery">
-          <TerminalSquare class="w-3.5 h-3.5 mr-2" /> {{ t('contextMenu.newQuery') }}
+          <TerminalSquare class="w-4 h-4" /> {{ t('contextMenu.newQuery') }}
         </ContextMenuItem>
         <ContextMenuItem @click="refresh">
-          <RefreshCw class="w-3.5 h-3.5 mr-2" /> {{ t('contextMenu.refreshChildren') }}
+          <RefreshCw class="w-4 h-4" /> {{ t('contextMenu.refreshChildren') }}
+        </ContextMenuItem>
+        <ContextMenuSeparator />
+        <ContextMenuItem @click="openTransfer">
+          <ArrowRightLeft class="w-4 h-4" /> {{ t('transfer.dataTransfer') }}
         </ContextMenuItem>
       </template>
 
       <template v-if="node.type === 'table' || node.type === 'view'">
         <ContextMenuItem @click="openData">
-          <TableProperties class="w-3.5 h-3.5 mr-2" /> {{ t('contextMenu.viewData') }}
+          <TableProperties class="w-4 h-4" /> {{ t('contextMenu.viewData') }}
         </ContextMenuItem>
         <ContextMenuItem @click="newQuery">
-          <TerminalSquare class="w-3.5 h-3.5 mr-2" /> {{ t('contextMenu.newQuery') }}
+          <TerminalSquare class="w-4 h-4" /> {{ t('contextMenu.newQuery') }}
         </ContextMenuItem>
         <ContextMenuSeparator />
         <ContextMenuItem @click="refresh">
-          <RefreshCw class="w-3.5 h-3.5 mr-2" /> {{ t('contextMenu.refreshChildren') }}
+          <RefreshCw class="w-4 h-4" /> {{ t('contextMenu.refreshChildren') }}
         </ContextMenuItem>
       </template>
 
       <template v-if="isGroupLabel(node)">
         <ContextMenuItem @click="refresh">
-          <RefreshCw class="w-3.5 h-3.5 mr-2" /> {{ t('contextMenu.refreshChildren') }}
+          <RefreshCw class="w-4 h-4" /> {{ t('contextMenu.refreshChildren') }}
         </ContextMenuItem>
       </template>
 
-      <ContextMenuSeparator />
+      <ContextMenuSeparator v-if="hasTypeMenu" />
       <ContextMenuItem @click="copyName">
-        <Copy class="w-3.5 h-3.5 mr-2" /> {{ t('contextMenu.copyName') }}
+        <Copy class="w-4 h-4" /> {{ t('contextMenu.copyName') }}
       </ContextMenuItem>
     </ContextMenuContent>
   </ContextMenu>

src/components/transfer/DataTransferDialog.vue

@@ -0,0 +1,512 @@
+<script setup lang="ts">
+import { computed, ref, watch } from "vue";
+import { useI18n } from "vue-i18n";
+import {
+  Dialog, DialogHeader, DialogTitle,
+  DialogFooter, DialogScrollContent,
+} from "@/components/ui/dialog";
+import { Button } from "@/components/ui/button";
+import { Input } from "@/components/ui/input";
+import { Label } from "@/components/ui/label";
+import {
+  Select, SelectContent, SelectItem, SelectTrigger, SelectValue,
+} from "@/components/ui/select";
+import { useConnectionStore } from "@/stores/connectionStore";
+import DatabaseIcon from "@/components/icons/DatabaseIcon.vue";
+import * as api from "@/lib/tauri";
+import type { TransferProgress } from "@/lib/tauri";
+import type { DatabaseType } from "@/types/database";
+import {
+  ArrowRightLeft, Check, X, Loader2, Square, CheckSquare,
+} from "lucide-vue-next";
+
+const { t } = useI18n();
+const open = defineModel<boolean>("open", { default: false });
+
+const props = defineProps<{
+  prefillConnectionId?: string;
+  prefillDatabase?: string;
+}>();
+
+const store = useConnectionStore();
+
+const SQL_TYPES: DatabaseType[] = ["mysql", "postgres", "sqlite", "sqlserver", "oracle", "clickhouse", "duckdb"];
+
+const sqlConnections = computed(() =>
+  store.connections.filter((c) => SQL_TYPES.includes(c.db_type))
+);
+
+// Source state
+const sourceConnectionId = ref("");
+const sourceDatabase = ref("");
+const sourceDatabases = ref<string[]>([]);
+const sourceSchema = ref("");
+const sourceTables = ref<string[]>([]);
+const selectedTables = ref<Set<string>>(new Set());
+const tableSearch = ref("");
+const loadingTables = ref(false);
+
+// Target state
+const targetConnectionId = ref("");
+const targetDatabase = ref("");
+const targetDatabases = ref<string[]>([]);
+const targetSchema = ref("");
+
+// Options
+const createTable = ref(true);
+const truncateBefore = ref(false);
+const batchSize = ref(1000);
+
+// Transfer state
+const isTransferring = ref(false);
+const transferProgress = ref<Map<string, TransferProgress>>(new Map());
+const currentTable = ref("");
+const overallDone = ref(false);
+const overallError = ref(false);
+const overallCancelled = ref(false);
+const transferId = ref("");
+
+const filteredTables = computed(() => {
+  const q = tableSearch.value.toLowerCase();
+  return q ? sourceTables.value.filter((t) => t.toLowerCase().includes(q)) : sourceTables.value;
+});
+
+const allSelected = computed(() =>
+  filteredTables.value.length > 0 && filteredTables.value.every((t) => selectedTables.value.has(t))
+);
+
+const canStart = computed(() =>
+  sourceConnectionId.value &&
+  sourceDatabase.value &&
+  targetConnectionId.value &&
+  targetDatabase.value &&
+  selectedTables.value.size > 0 &&
+  sourceConnectionId.value + sourceDatabase.value !== targetConnectionId.value + targetDatabase.value
+);
+
+function toggleSelectAll() {
+  if (allSelected.value) {
+    filteredTables.value.forEach((t) => selectedTables.value.delete(t));
+  } else {
+    filteredTables.value.forEach((t) => selectedTables.value.add(t));
+  }
+}
+
+function toggleTable(table: string) {
+  if (selectedTables.value.has(table)) {
+    selectedTables.value.delete(table);
+  } else {
+    selectedTables.value.add(table);
+  }
+}
+
+async function loadDatabases(connectionId: string, target: "source" | "target") {
+  if (!connectionId) return;
+  try {
+    await store.ensureConnected(connectionId);
+    const dbs = await api.listDatabases(connectionId);
+    const names = dbs.map((d) => d.name);
+    if (target === "source") {
+      sourceDatabases.value = names;
+      sourceDatabase.value = names.length === 1 ? names[0] : "";
+    } else {
+      targetDatabases.value = names;
+      targetDatabase.value = names.length === 1 ? names[0] : "";
+    }
+  } catch {
+    if (target === "source") sourceDatabases.value = [];
+    else targetDatabases.value = [];
+  }
+}
+
+async function loadTables() {
+  if (!sourceConnectionId.value || !sourceDatabase.value) {
+    sourceTables.value = [];
+    return;
+  }
+  loadingTables.value = true;
+  try {
+    const config = store.getConfig(sourceConnectionId.value);
+    const needsSchema = config?.db_type === "postgres" || config?.db_type === "sqlserver" || config?.db_type === "oracle";
+    if (needsSchema) {
+      const schemas = await api.listSchemas(sourceConnectionId.value, sourceDatabase.value);
+      sourceSchema.value = schemas.includes("public") ? "public" : (schemas[0] ?? "");
+    } else {
+      sourceSchema.value = "";
+    }
+    const tables = await api.listTables(sourceConnectionId.value, sourceDatabase.value, sourceSchema.value);
+    sourceTables.value = tables.filter((t) => t.table_type === "TABLE" || t.table_type === "BASE TABLE").map((t) => t.name);
+    selectedTables.value = new Set(sourceTables.value);
+  } catch {
+    sourceTables.value = [];
+  } finally {
+    loadingTables.value = false;
+  }
+}
+
+const skipSourceWatch = ref(false);
+
+watch(sourceConnectionId, (id) => {
+  if (skipSourceWatch.value) { skipSourceWatch.value = false; return; }
+  sourceDatabase.value = "";
+  sourceTables.value = [];
+  selectedTables.value.clear();
+  loadDatabases(id, "source");
+});
+
+watch(sourceDatabase, () => loadTables());
+
+watch(targetConnectionId, (id) => {
+  targetDatabase.value = "";
+  loadDatabases(id, "target");
+});
+
+watch(open, async (val) => {
+  if (val) {
+    resetState();
+    if (props.prefillConnectionId) {
+      skipSourceWatch.value = true;
+      sourceConnectionId.value = props.prefillConnectionId;
+      await loadDatabases(props.prefillConnectionId, "source");
+      if (props.prefillDatabase) {
+        sourceDatabase.value = props.prefillDatabase;
+      }
+    }
+  }
+});
+
+function resetState() {
+  sourceConnectionId.value = "";
+  sourceDatabase.value = "";
+  sourceDatabases.value = [];
+  sourceSchema.value = "";
+  sourceTables.value = [];
+  selectedTables.value.clear();
+  tableSearch.value = "";
+  targetConnectionId.value = "";
+  targetDatabase.value = "";
+  targetDatabases.value = [];
+  targetSchema.value = "";
+  createTable.value = true;
+  truncateBefore.value = false;
+  batchSize.value = 1000;
+  isTransferring.value = false;
+  transferProgress.value.clear();
+  currentTable.value = "";
+  overallDone.value = false;
+  overallError.value = false;
+  overallCancelled.value = false;
+}
+
+async function startTransfer() {
+  isTransferring.value = true;
+  overallDone.value = false;
+  overallError.value = false;
+  overallCancelled.value = false;
+  transferProgress.value.clear();
+
+  transferId.value = crypto.randomUUID();
+
+  // Auto-detect target schema
+  const targetConfig = store.getConfig(targetConnectionId.value);
+  const targetNeedsSchema = targetConfig?.db_type === "postgres" || targetConfig?.db_type === "sqlserver" || targetConfig?.db_type === "oracle";
+  if (targetNeedsSchema && !targetSchema.value) {
+    try {
+      const schemas = await api.listSchemas(targetConnectionId.value, targetDatabase.value);
+      targetSchema.value = schemas.includes("public") ? "public" : (schemas[0] ?? "");
+    } catch { /* use empty */ }
+  }
+
+  const request: api.TransferRequest = {
+    transferId: transferId.value,
+    sourceConnectionId: sourceConnectionId.value,
+    sourceDatabase: sourceDatabase.value,
+    sourceSchema: sourceSchema.value,
+    targetConnectionId: targetConnectionId.value,
+    targetDatabase: targetDatabase.value,
+    targetSchema: targetSchema.value,
+    tables: [...selectedTables.value],
+    createTable: createTable.value,
+    truncateBefore: truncateBefore.value,
+    batchSize: batchSize.value,
+  };
+
+  try {
+    await api.startTransfer(request, (progress) => {
+      transferProgress.value.set(progress.table, progress);
+      transferProgress.value = new Map(transferProgress.value);
+      currentTable.value = progress.table;
+
+      if (progress.status === "done") {
+        overallDone.value = true;
+      } else if (progress.status === "cancelled") {
+        overallCancelled.value = true;
+      }
+    });
+  } catch (e: any) {
+    overallError.value = true;
+  }
+}
+
+async function cancelTransfer() {
+  if (transferId.value) {
+    await api.cancelTransfer(transferId.value);
+  }
+}
+
+function getConnectionName(id: string) {
+  return store.connections.find((c) => c.id === id)?.name ?? id;
+}
+
+function getConnectionType(id: string): DatabaseType {
+  return store.connections.find((c) => c.id === id)?.db_type ?? "mysql";
+}
+
+const completedTables = computed(() =>
+  [...transferProgress.value.values()].filter((p) => p.status === "tableDone" || p.status === "done").length
+);
+
+const totalTransferred = computed(() =>
+  [...transferProgress.value.values()].reduce((sum, p) => sum + p.rowsTransferred, 0)
+);
+</script>
+
+<template>
+  <Dialog v-model:open="open">
+    <DialogScrollContent class="sm:max-w-[560px]" :trap-focus="false" @interact-outside.prevent>
+      <DialogHeader>
+        <DialogTitle class="flex items-center gap-2">
+          <ArrowRightLeft class="w-4 h-4" />
+          {{ t('transfer.title') }}
+        </DialogTitle>
+      </DialogHeader>
+
+      <!-- Config View -->
+      <div v-if="!isTransferring" class="grid gap-4 py-3">
+        <!-- Source Section -->
+        <div class="space-y-3">
+          <div class="text-xs font-medium text-muted-foreground uppercase tracking-wider">{{ t('transfer.source') }}</div>
+
+          <div class="grid grid-cols-2 gap-3">
+            <div class="space-y-1.5">
+              <Label class="text-xs">{{ t('transfer.sourceConnection') }}</Label>
+              <Select v-model="sourceConnectionId">
+                <SelectTrigger class="h-8 text-xs">
+                  <div v-if="sourceConnectionId" class="flex items-center gap-1.5">
+                    <DatabaseIcon :db-type="getConnectionType(sourceConnectionId)" class="w-3.5 h-3.5" />
+                    <span class="truncate">{{ getConnectionName(sourceConnectionId) }}</span>
+                  </div>
+                  <SelectValue v-else />
+                </SelectTrigger>
+                <SelectContent>
+                  <SelectItem v-for="c in sqlConnections" :key="c.id" :value="c.id">
+                    <div class="flex items-center gap-1.5">
+                      <DatabaseIcon :db-type="c.db_type" class="w-3.5 h-3.5" />
+                      {{ c.name }}
+                    </div>
+                  </SelectItem>
+                </SelectContent>
+              </Select>
+            </div>
+
+            <div class="space-y-1.5">
+              <Label class="text-xs">{{ t('transfer.sourceDatabase') }}</Label>
+              <Select v-model="sourceDatabase" :disabled="!sourceDatabases.length">
+                <SelectTrigger class="h-8 text-xs">
+                  <SelectValue />
+                </SelectTrigger>
+                <SelectContent>
+                  <SelectItem v-for="db in sourceDatabases" :key="db" :value="db">{{ db }}</SelectItem>
+                </SelectContent>
+              </Select>
+            </div>
+          </div>
+        </div>
+
+        <!-- Target Section -->
+        <div class="space-y-3">
+          <div class="text-xs font-medium text-muted-foreground uppercase tracking-wider">{{ t('transfer.target') }}</div>
+
+          <div class="grid grid-cols-2 gap-3">
+            <div class="space-y-1.5">
+              <Label class="text-xs">{{ t('transfer.targetConnection') }}</Label>
+              <Select v-model="targetConnectionId">
+                <SelectTrigger class="h-8 text-xs">
+                  <div v-if="targetConnectionId" class="flex items-center gap-1.5">
+                    <DatabaseIcon :db-type="getConnectionType(targetConnectionId)" class="w-3.5 h-3.5" />
+                    <span class="truncate">{{ getConnectionName(targetConnectionId) }}</span>
+                  </div>
+                  <SelectValue v-else />
+                </SelectTrigger>
+                <SelectContent>
+                  <SelectItem v-for="c in sqlConnections" :key="c.id" :value="c.id">
+                    <div class="flex items-center gap-1.5">
+                      <DatabaseIcon :db-type="c.db_type" class="w-3.5 h-3.5" />
+                      {{ c.name }}
+                    </div>
+                  </SelectItem>
+                </SelectContent>
+              </Select>
+            </div>
+
+            <div class="space-y-1.5">
+              <Label class="text-xs">{{ t('transfer.targetDatabase') }}</Label>
+              <Select v-model="targetDatabase" :disabled="!targetDatabases.length">
+                <SelectTrigger class="h-8 text-xs">
+                  <SelectValue />
+                </SelectTrigger>
+                <SelectContent>
+                  <SelectItem v-for="db in targetDatabases" :key="db" :value="db">{{ db }}</SelectItem>
+                </SelectContent>
+              </Select>
+            </div>
+          </div>
+        </div>
+
+        <!-- Tables Section -->
+        <div class="space-y-2">
+          <div class="flex items-center justify-between">
+            <div class="text-xs font-medium text-muted-foreground uppercase tracking-wider">
+              {{ t('transfer.tables') }}
+              <span v-if="sourceTables.length" class="text-muted-foreground/60">({{ selectedTables.size }}/{{ sourceTables.length }})</span>
+            </div>
+            <Button v-if="sourceTables.length" variant="ghost" size="sm" class="h-6 text-xs px-2" @click="toggleSelectAll">
+              {{ allSelected ? t('transfer.deselectAll') : t('transfer.selectAll') }}
+            </Button>
+          </div>
+
+          <Input
+            v-if="sourceTables.length > 5"
+            v-model="tableSearch"
+            :placeholder="t('transfer.searchTables')"
+            class="h-7 text-xs"
+          />
+
+          <div v-if="loadingTables" class="flex items-center gap-2 text-xs text-muted-foreground py-4 justify-center">
+            <Loader2 class="w-3.5 h-3.5 animate-spin" />
+            {{ t('common.loading') }}
+          </div>
+          <div v-else-if="!sourceConnectionId || !sourceDatabase" class="text-xs text-muted-foreground py-4 text-center">
+            {{ t('transfer.selectSourceFirst') }}
+          </div>
+          <div v-else-if="sourceTables.length === 0" class="text-xs text-muted-foreground py-4 text-center">
+            {{ t('transfer.noTables') }}
+          </div>
+          <div v-else class="border rounded-md max-h-[200px] overflow-y-auto">
+            <div
+              v-for="table in filteredTables"
+              :key="table"
+              class="flex items-center gap-2 px-2.5 py-1.5 hover:bg-muted/50 cursor-pointer text-xs"
+              @click="toggleTable(table)"
+            >
+              <CheckSquare v-if="selectedTables.has(table)" class="w-3.5 h-3.5 text-primary shrink-0" />
+              <Square v-else class="w-3.5 h-3.5 text-muted-foreground/40 shrink-0" />
+              <span class="truncate">{{ table }}</span>
+            </div>
+          </div>
+        </div>
+
+        <!-- Options -->
+        <div class="space-y-2.5">
+          <div
+            class="flex items-center gap-2 cursor-pointer text-xs"
+            @click="createTable = !createTable"
+          >
+            <CheckSquare v-if="createTable" class="w-3.5 h-3.5 text-primary shrink-0" />
+            <Square v-else class="w-3.5 h-3.5 text-muted-foreground/40 shrink-0" />
+            {{ t('transfer.createTable') }}
+          </div>
+          <div
+            class="flex items-center gap-2 cursor-pointer text-xs"
+            @click="truncateBefore = !truncateBefore"
+          >
+            <CheckSquare v-if="truncateBefore" class="w-3.5 h-3.5 text-primary shrink-0" />
+            <Square v-else class="w-3.5 h-3.5 text-muted-foreground/40 shrink-0" />
+            {{ t('transfer.truncateBefore') }}
+          </div>
+          <div class="flex items-center gap-3">
+            <Label class="text-xs shrink-0">{{ t('transfer.batchSize') }}</Label>
+            <Input v-model.number="batchSize" type="number" min="100" max="10000" step="100" class="h-7 text-xs w-24" />
+          </div>
+        </div>
+      </div>
+
+      <!-- Progress View -->
+      <div v-else class="py-3 space-y-3">
+        <div class="flex items-center justify-between text-xs text-muted-foreground">
+          <span>
+            {{ t('transfer.overallProgress') }}: {{ completedTables }} / {{ selectedTables.size }} {{ t('transfer.tables').toLowerCase() }}
+            · {{ totalTransferred.toLocaleString() }} {{ t('grid.rows', { count: '' }).trim() }}
+          </span>
+          <span v-if="overallDone" class="text-green-600 font-medium">{{ t('transfer.completed') }}</span>
+          <span v-else-if="overallCancelled" class="text-yellow-600 font-medium">{{ t('transfer.cancelled') }}</span>
+        </div>
+
+        <div class="w-full bg-muted rounded-full h-2 overflow-hidden">
+          <div
+            class="h-full rounded-full transition-all duration-300"
+            :class="overallError ? 'bg-destructive' : overallCancelled ? 'bg-yellow-500' : 'bg-primary'"
+            :style="{ width: `${selectedTables.size ? (completedTables / selectedTables.size) * 100 : 0}%` }"
+          />
+        </div>
+
+        <div class="border rounded-md max-h-[280px] overflow-y-auto">
+          <div
+            v-for="table in [...selectedTables]"
+            :key="table"
+            class="flex items-center justify-between px-2.5 py-1.5 text-xs border-b last:border-b-0"
+          >
+            <span class="truncate">{{ table }}</span>
+            <div class="flex items-center gap-1.5 shrink-0 text-muted-foreground">
+              <template v-if="transferProgress.get(table)">
+                <template v-if="transferProgress.get(table)!.status === 'running'">
+                  <Loader2 class="w-3 h-3 animate-spin text-primary" />
+                  <span>{{ transferProgress.get(table)!.rowsTransferred.toLocaleString() }}</span>
+                </template>
+                <template v-else-if="transferProgress.get(table)!.status === 'tableDone' || transferProgress.get(table)!.status === 'done'">
+                  <Check class="w-3 h-3 text-green-500" />
+                  <span>{{ transferProgress.get(table)!.rowsTransferred.toLocaleString() }}</span>
+                </template>
+                <template v-else-if="transferProgress.get(table)!.status === 'error'">
+                  <X class="w-3 h-3 text-destructive" />
+                  <span class="text-destructive truncate max-w-[160px]" :title="transferProgress.get(table)!.error ?? ''">
+                    {{ transferProgress.get(table)!.error }}
+                  </span>
+                </template>
+                <template v-else-if="transferProgress.get(table)!.status === 'cancelled'">
+                  <X class="w-3 h-3 text-yellow-500" />
+                  <span>{{ t('transfer.cancelled') }}</span>
+                </template>
+              </template>
+              <span v-else class="text-muted-foreground/40">—</span>
+            </div>
+          </div>
+        </div>
+
+        <!-- Status message -->
+      </div>
+
+      <DialogFooter>
+        <template v-if="!isTransferring">
+          <Button variant="outline" size="sm" @click="open = false">
+            {{ t('transfer.cancel') }}
+          </Button>
+          <Button size="sm" :disabled="!canStart" @click="startTransfer">
+            <ArrowRightLeft class="w-3.5 h-3.5 mr-1.5" />
+            {{ t('transfer.start') }}
+          </Button>
+        </template>
+        <template v-else-if="overallDone || overallCancelled">
+          <Button size="sm" @click="open = false">
+            {{ t('common.close') }}
+          </Button>
+        </template>
+        <template v-else>
+          <Button variant="destructive" size="sm" @click="cancelTransfer">
+            {{ t('transfer.cancel') }}
+          </Button>
+        </template>
+      </DialogFooter>
+    </DialogScrollContent>
+  </Dialog>
+</template>

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",
+  },
 };

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: "数据传输",
+  },
 };

src/lib/tauri.ts

@@ -261,3 +261,55 @@ export async function clearHistory(): Promise<void> {
 export async function deleteHistoryEntry(id: string): Promise<void> {
   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<void> {
+  const unlisten: UnlistenFn = await listen<TransferProgress>("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<void> {
+  return invoke("cancel_transfer", { transferId });
+}

src/stores/connectionStore.ts

@@ -16,6 +16,7 @@ export const useConnectionStore = defineStore("connection", () => {
   const editingConnectionId = ref<string | null>(null);
   const completionTablesCache = ref<Record<string, SqlCompletionTable[]>>({});
   const completionColumnsCache = ref<Record<string, ColumnInfo[]>>({});
+  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,
   };
 });

src/styles/globals.css

@@ -136,8 +136,10 @@
   [role="switch"],
   select,
   summary,
+  label,
   a[href] {
     cursor: pointer;
+    user-select: none;
     }
 }