diff --git a/crates/dbx-core/src/query.rs b/crates/dbx-core/src/query.rs index 5524a150cd..564fa1fbbc 100644 --- a/crates/dbx-core/src/query.rs +++ b/crates/dbx-core/src/query.rs @@ -368,6 +368,230 @@ pub async fn execute_statements( }) } +/// Execute multiple SQL statements within a single transaction. +/// For sqlx-based pools (Postgres/MySQL/SQLite), uses the Transaction API to +/// guarantee all statements run on the same physical connection. +/// For custom drivers (ClickHouse/SqlServer/Dameng/Gaussdb), uses explicit +/// BEGIN/COMMIT/ROLLBACK on the already-single-connection client. +/// For databases that don't support explicit transactions (Redis, MongoDB, Oracle), +/// executes statements sequentially without transaction. +/// If BEGIN fails, returns an error — no silent fallback to auto-commit. +pub async fn execute_statements_in_transaction( + state: &AppState, + connection_id: &str, + database: &str, + statements: &[String], + schema: Option<&str>, +) -> Result { + let pool_key = if database.is_empty() { + connection_id.to_string() + } else { + state.get_or_create_pool(connection_id, Some(database)).await? + }; + + let start = std::time::Instant::now(); + + // Clone the pool handle within the lock, then drop it before any async work. + let path = { + let conns = state.connections.lock().await; + conns.get(&pool_key).map(|p| match p { + PoolKind::Postgres(pg) => TxPath::Pg(pg.clone()), + PoolKind::Mysql(mp, bare) => TxPath::Mysql(mp.clone(), *bare), + PoolKind::Sqlite(sq) => TxPath::Sqlite(sq.clone()), + PoolKind::ClickHouse(_) | PoolKind::SqlServer(_) | PoolKind::Dameng(_) | PoolKind::Gaussdb(_) => { + TxPath::Explicit + } + PoolKind::DuckDb(_) + | PoolKind::Redis(_) + | PoolKind::MongoDb(_) + | PoolKind::Oracle(_) + | PoolKind::Elasticsearch(_) => TxPath::None, + }) + }; + + match path { + Some(TxPath::Pg(pool)) => exec_tx_pg_inner(pool, statements, schema, start).await, + Some(TxPath::Mysql(pool, _bare)) => exec_tx_mysql_inner(pool, statements, start).await, + Some(TxPath::Sqlite(pool)) => exec_tx_sqlite_inner(pool, statements, start).await, + Some(TxPath::Explicit) => exec_tx_explicit_inner(state, &pool_key, statements, schema, start).await, + Some(TxPath::None) => exec_tx_none_inner(state, &pool_key, statements, schema, start).await, + None => Err("Connection not found for transaction".to_string()), + } +} + +/// Owned pool variants for safe dispatch across async boundaries. +enum TxPath { + Pg(sqlx::postgres::PgPool), + Mysql(sqlx::mysql::MySqlPool, bool), + Sqlite(sqlx::sqlite::SqlitePool), + Explicit, + None, +} + +// Each of these acquires a dedicated connection and runs all statements within +// BEGIN ... COMMIT/ROLLBACK, guaranteeing a single physical connection. +// This avoids sqlx::Transaction which has Send/lifetime incompatibility with Tauri macro. + +async fn exec_tx_pg_inner( + pool: sqlx::postgres::PgPool, + statements: &[String], + schema: Option<&str>, + start: std::time::Instant, +) -> Result { + let mut conn = pool.acquire().await.map_err(|e| format!("Failed to acquire connection: {}", e))?; + // Set schema first + if let Some(s) = schema { + let sp = format!("SET search_path TO \"{}\", public", s); + sqlx::query(&sp).execute(&mut *conn).await.map_err(|e| format!("SET search_path failed: {}", e))?; + } + sqlx::query("BEGIN").execute(&mut *conn).await.map_err(|e| format!("Failed to begin transaction: {}", e))?; + let mut total_affected: u64 = 0; + for (i, sql) in statements.iter().enumerate() { + match sqlx::query(sql).execute(&mut *conn).await { + Ok(r) => total_affected += r.rows_affected(), + Err(e) => { + let _ = sqlx::query("ROLLBACK").execute(&mut *conn).await; + return Err(format!("Statement {} failed: {}", i + 1, e)); + } + } + } + sqlx::query("COMMIT").execute(&mut *conn).await.map_err(|e| format!("COMMIT failed: {}", e))?; + Ok(db::QueryResult { + columns: vec![], + rows: vec![], + affected_rows: total_affected, + execution_time_ms: start.elapsed().as_millis(), + truncated: false, + }) +} + +async fn exec_tx_mysql_inner( + pool: sqlx::mysql::MySqlPool, + statements: &[String], + start: std::time::Instant, +) -> Result { + let mut conn = pool.acquire().await.map_err(|e| format!("Failed to acquire connection: {}", e))?; + sqlx::query("START TRANSACTION") + .execute(&mut *conn) + .await + .map_err(|e| format!("Failed to begin transaction: {}", e))?; + let mut total_affected: u64 = 0; + for (i, sql) in statements.iter().enumerate() { + match sqlx::query(sql).execute(&mut *conn).await { + Ok(r) => total_affected += r.rows_affected(), + Err(e) => { + let _ = sqlx::query("ROLLBACK").execute(&mut *conn).await; + return Err(format!("Statement {} failed: {}", i + 1, e)); + } + } + } + sqlx::query("COMMIT").execute(&mut *conn).await.map_err(|e| format!("COMMIT failed: {}", e))?; + Ok(db::QueryResult { + columns: vec![], + rows: vec![], + affected_rows: total_affected, + execution_time_ms: start.elapsed().as_millis(), + truncated: false, + }) +} + +async fn exec_tx_sqlite_inner( + pool: sqlx::sqlite::SqlitePool, + statements: &[String], + start: std::time::Instant, +) -> Result { + let mut conn = pool.acquire().await.map_err(|e| format!("Failed to acquire connection: {}", e))?; + sqlx::query("BEGIN").execute(&mut *conn).await.map_err(|e| format!("Failed to begin transaction: {}", e))?; + let mut total_affected: u64 = 0; + for (i, sql) in statements.iter().enumerate() { + match sqlx::query(sql).execute(&mut *conn).await { + Ok(r) => total_affected += r.rows_affected(), + Err(e) => { + let _ = sqlx::query("ROLLBACK").execute(&mut *conn).await; + return Err(format!("Statement {} failed: {}", i + 1, e)); + } + } + } + sqlx::query("COMMIT").execute(&mut *conn).await.map_err(|e| format!("COMMIT failed: {}", e))?; + Ok(db::QueryResult { + columns: vec![], + rows: vec![], + affected_rows: total_affected, + execution_time_ms: start.elapsed().as_millis(), + truncated: false, + }) +} + +async fn exec_tx_explicit_inner( + state: &AppState, + pool_key: &str, + statements: &[String], + schema: Option<&str>, + start: std::time::Instant, +) -> Result { + do_execute(state, pool_key, "BEGIN", schema, None) + .await + .map_err(|e| format!("Failed to begin transaction: {}", e))?; + + let mut total_affected: u64 = 0; + for (i, sql) in statements.iter().enumerate() { + match do_execute(state, pool_key, sql, schema, None).await { + Ok(result) => { + total_affected += result.affected_rows; + } + Err(e) => { + if let Err(rb_err) = do_execute(state, pool_key, "ROLLBACK", schema, None).await { + log::error!("ROLLBACK failed after statement {} error: {}", i + 1, rb_err); + } + return Err(format!("Statement {} failed: {}", i + 1, e)); + } + } + } + + do_execute(state, pool_key, "COMMIT", schema, None).await.map_err(|e| format!("COMMIT failed: {}", e))?; + + Ok(db::QueryResult { + columns: vec![], + rows: vec![], + affected_rows: total_affected, + execution_time_ms: start.elapsed().as_millis(), + truncated: false, + }) +} + +async fn exec_tx_none_inner( + state: &AppState, + pool_key: &str, + statements: &[String], + schema: Option<&str>, + start: std::time::Instant, +) -> Result { + let mut total_affected: u64 = 0; + for (i, sql) in statements.iter().enumerate() { + match do_execute(state, pool_key, sql, schema, None).await { + Ok(result) => { + total_affected += result.affected_rows; + } + Err(e) => { + log::warn!("Statement {} failed (no transaction support): {}", i + 1, e); + return Err(format!( + "Statement {} failed: {}. No transaction support for this database type.", + i + 1, + e + )); + } + } + } + + Ok(db::QueryResult { + columns: vec![], + rows: vec![], + affected_rows: total_affected, + execution_time_ms: start.elapsed().as_millis(), + truncated: false, + }) +} + #[cfg(test)] mod tests { use super::*; diff --git a/src-tauri/src/commands/query.rs b/src-tauri/src/commands/query.rs index eacdae4987..387d6a5873 100644 --- a/src-tauri/src/commands/query.rs +++ b/src-tauri/src/commands/query.rs @@ -74,3 +74,21 @@ pub async fn execute_script( ) .await } + +#[tauri::command] +pub async fn execute_in_transaction( + state: State<'_, Arc>, + connection_id: String, + database: String, + statements: Vec, + schema: Option, +) -> Result { + dbx_core::query::execute_statements_in_transaction( + &state, + &connection_id, + &database, + &statements, + schema.as_deref(), + ) + .await +} diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 5a8fad1d3b..87d5b5a645 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -77,6 +77,7 @@ pub fn run() { commands::query::cancel_query, commands::query::execute_batch, commands::query::execute_script, + commands::query::execute_in_transaction, commands::sql_file::preview_sql_file, commands::sql_file::execute_sql_file, commands::sql_file::cancel_sql_file_execution, diff --git a/src-web/src/main.rs b/src-web/src/main.rs index 4c22e466dd..94ccbcb786 100644 --- a/src-web/src/main.rs +++ b/src-web/src/main.rs @@ -94,6 +94,7 @@ async fn main() { .route("/query/execute-multi", post(routes::query::execute_multi)) .route("/query/execute-batch", post(routes::query::execute_batch)) .route("/query/execute-script", post(routes::query::execute_script)) + .route("/query/execute-in-transaction", post(routes::query::execute_in_transaction)) .route("/query/cancel", post(routes::query::cancel_query)) // Redis .route("/redis/list-databases", post(routes::redis::list_databases)) diff --git a/src-web/src/routes/query.rs b/src-web/src/routes/query.rs index efd2402625..125868800d 100644 --- a/src-web/src/routes/query.rs +++ b/src-web/src/routes/query.rs @@ -122,3 +122,20 @@ pub async fn execute_script( Ok(Json(serde_json::to_value(result).map_err(|e| AppError(e.to_string()))?)) } + +pub async fn execute_in_transaction( + State(state): State>, + Json(req): Json, +) -> Result, AppError> { + let result = dbx_core::query::execute_statements_in_transaction( + &state.app, + &req.connection_id, + &req.database, + &req.statements, + req.schema.as_deref(), + ) + .await + .map_err(AppError)?; + + Ok(Json(serde_json::to_value(result).map_err(|e| AppError(e.to_string()))?)) +} diff --git a/src/components/grid/DataGrid.vue b/src/components/grid/DataGrid.vue index 903af8738a..ff03c25362 100644 --- a/src/components/grid/DataGrid.vue +++ b/src/components/grid/DataGrid.vue @@ -28,6 +28,9 @@ import { Info, Rows3, TriangleAlert, + RefreshCcw, + RotateCcw, + Table2, } from "lucide-vue-next"; import { Button } from "@/components/ui/button"; import { @@ -461,6 +464,35 @@ const deletedRowCount = computed(() => deletedRows.value.size); const pendingChangeCount = computed(() => dirtyRowCount.value + newRowCount.value + deletedRowCount.value); const hasPendingChanges = computed(() => pendingChangeCount.value > 0); +// --- Transaction state --- +const transactionActive = ref(false); +const isSaving = ref(false); + +function enterTransaction() { + transactionActive.value = true; +} + +function exitTransaction() { + transactionActive.value = false; +} + +const useTransaction = computed(() => props.editable && !!props.connectionId && !!props.database && !!props.tableMeta); + +async function onToolbarRefresh() { + if (transactionActive.value) { + discardChanges(); + } + emit("reload"); +} + +async function onToolbarCommit() { + await saveChanges(); +} + +function onToolbarRollback() { + discardChanges(); +} + const sortedRows = computed(() => { let rows = props.result.rows.map((row, sourceIndex) => ({ row, sourceIndex })); if (clientSearchText.value) { @@ -760,6 +792,10 @@ function commitEdit() { if (newVal !== oldVal) { if (!dirtyRows.value.has(item.sourceIndex)) dirtyRows.value.set(item.sourceIndex, new Map()); dirtyRows.value.get(item.sourceIndex)!.set(col, newVal); + // Enter transaction mode on first edit + if (useTransaction.value && !transactionActive.value) { + enterTransaction(); + } } else { const rowChanges = dirtyRows.value.get(item.sourceIndex); rowChanges?.delete(col); @@ -790,6 +826,9 @@ function onEditKeydown(e: KeyboardEvent) { function addRow() { rowStatusFilter.value = rowStatusFilterAfterAddingRow(rowStatusFilter.value); newRows.value.push(props.result.columns.map(() => null)); + if (useTransaction.value && !transactionActive.value) { + enterTransaction(); + } const rowId = -newRows.value.length; nextTick(() => { const el = getScrollerElement(); @@ -808,6 +847,9 @@ function applyDeleteRow(rowId: number) { deletedRows.value.add(item.sourceIndex); } if (editingCell.value?.rowId === rowId) editingCell.value = null; + if (useTransaction.value && !transactionActive.value) { + enterTransaction(); + } } const showDeleteRowConfirm = ref(false); @@ -858,12 +900,22 @@ async function saveChanges() { const stmts = generateSaveStatements(); if (stmts.length === 0) return; saveError.value = ""; + isSaving.value = true; - if (props.connectionId && props.database) { + if (useTransaction.value && props.connectionId && props.database) { + try { + await api.executeInTransaction(props.connectionId, props.database, stmts, props.tableMeta?.schema); + } catch (e: any) { + saveError.value = String(e.message || e); + isSaving.value = false; + return; + } + } else if (props.connectionId && props.database) { try { await api.executeBatch(props.connectionId, props.database, stmts); } catch (e: any) { saveError.value = String(e.message || e); + isSaving.value = false; return; } } else if (props.onExecuteSql) { @@ -873,12 +925,15 @@ async function saveChanges() { } } catch (e: any) { saveError.value = String(e.message || e); + isSaving.value = false; return; } } dirtyRows.value.clear(); newRows.value = []; deletedRows.value.clear(); + exitTransaction(); + isSaving.value = false; emit("reload"); } @@ -887,6 +942,7 @@ function discardChanges() { newRows.value = []; deletedRows.value.clear(); editingCell.value = null; + exitTransaction(); } // --- Cell selection and detail --- @@ -1009,6 +1065,7 @@ watch( detailCell.value = null; showTranspose.value = false; transposeRowIndex.value = null; + exitTransaction(); }, ); @@ -1211,6 +1268,55 @@ function escapeAndHighlightKeywords(s: string): string {
+ +
+ + + {{ tableMeta.tableName }} + + + + {{ t("grid.transactionActive") }} + + + + + +
@@ -1673,7 +1779,7 @@ function escapeAndHighlightKeywords(s: string): string { {{ result.execution_time_ms }}ms {{ selectionSummary }} -