diff --git a/crates/dbx-core/src/query.rs b/crates/dbx-core/src/query.rs index 568301f55d..a7022072f2 100644 --- a/crates/dbx-core/src/query.rs +++ b/crates/dbx-core/src/query.rs @@ -361,10 +361,14 @@ pub async fn execute_statements( }) } -/// Execute multiple SQL statements within a single transaction (BEGIN ... COMMIT). -/// If any statement fails, ROLLBACK is issued and an error is returned. -/// For databases that don't support explicit transactions (Redis, MongoDB), this -/// falls back to executing statements sequentially. +/// 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, @@ -378,82 +382,158 @@ pub async fn execute_statements_in_transaction( state.get_or_create_pool(connection_id, Some(database)).await? }; - // Check if this connection supports transactions - let supports_tx = { - let connections = state.connections.lock().await; - matches!( - connections.get(&pool_key), - Some(PoolKind::Mysql(_, _)) - | Some(PoolKind::Postgres(_)) - | Some(PoolKind::Sqlite(_)) - | Some(PoolKind::ClickHouse(_)) - | Some(PoolKind::SqlServer(_)) - | Some(PoolKind::Dameng(_)) - | Some(PoolKind::Gaussdb(_)) - ) - }; - let start = std::time::Instant::now(); - if !supports_tx { - // Fallback: execute statements sequentially without transaction - 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 - )); - } + // 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 } - } - return Ok(db::QueryResult { - columns: vec![], - rows: vec![], - affected_rows: total_affected, - execution_time_ms: start.elapsed().as_millis(), - truncated: false, - }); - } + PoolKind::DuckDb(_) + | PoolKind::Redis(_) + | PoolKind::MongoDb(_) + | PoolKind::Oracle(_) + | PoolKind::Elasticsearch(_) => TxPath::None, + }) + }; - // Execute within transaction - match do_execute(state, &pool_key, "BEGIN", schema, None).await { - Ok(_) => {} - Err(e) => { - log::warn!("BEGIN failed: {}, proceeding without transaction", e); - // Fallback without transaction - let mut total_affected: u64 = 0; - for sql in statements { - match do_execute(state, &pool_key, sql, schema, None).await { - Ok(result) => total_affected += result.affected_rows, - Err(e2) => return Err(e2), - } + 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)); } - return Ok(db::QueryResult { - columns: vec![], - rows: vec![], - affected_rows: total_affected, - execution_time_ms: start.elapsed().as_millis(), - truncated: false, - }); } } + 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 { + match do_execute(state, pool_key, sql, schema, None).await { Ok(result) => { total_affected += result.affected_rows; } Err(e) => { - // Rollback on failure - if let Err(rb_err) = do_execute(state, &pool_key, "ROLLBACK", schema, None).await { + 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)); @@ -461,11 +541,38 @@ pub async fn execute_statements_in_transaction( } } - // Commit - match do_execute(state, &pool_key, "COMMIT", schema, None).await { - Ok(_) => {} - Err(e) => { - return Err(format!("COMMIT failed: {}", 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 + )); + } } } diff --git a/src/components/grid/DataGrid.vue b/src/components/grid/DataGrid.vue index cf88a82d3a..ff03c25362 100644 --- a/src/components/grid/DataGrid.vue +++ b/src/components/grid/DataGrid.vue @@ -476,9 +476,7 @@ function exitTransaction() { transactionActive.value = false; } -function shouldUseTransaction(): boolean { - return props.editable && !!props.connectionId && !!props.database && !!props.tableMeta; -} +const useTransaction = computed(() => props.editable && !!props.connectionId && !!props.database && !!props.tableMeta); async function onToolbarRefresh() { if (transactionActive.value) { @@ -795,7 +793,7 @@ function commitEdit() { 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 (shouldUseTransaction() && !transactionActive.value) { + if (useTransaction.value && !transactionActive.value) { enterTransaction(); } } else { @@ -828,7 +826,7 @@ function onEditKeydown(e: KeyboardEvent) { function addRow() { rowStatusFilter.value = rowStatusFilterAfterAddingRow(rowStatusFilter.value); newRows.value.push(props.result.columns.map(() => null)); - if (shouldUseTransaction() && !transactionActive.value) { + if (useTransaction.value && !transactionActive.value) { enterTransaction(); } const rowId = -newRows.value.length; @@ -849,7 +847,7 @@ function applyDeleteRow(rowId: number) { deletedRows.value.add(item.sourceIndex); } if (editingCell.value?.rowId === rowId) editingCell.value = null; - if (shouldUseTransaction() && !transactionActive.value) { + if (useTransaction.value && !transactionActive.value) { enterTransaction(); } } @@ -904,7 +902,7 @@ async function saveChanges() { saveError.value = ""; isSaving.value = true; - if (shouldUseTransaction() && 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) { @@ -1272,7 +1270,7 @@ function escapeAndHighlightKeywords(s: string): string {
diff --git a/src/stores/queryStore.ts b/src/stores/queryStore.ts index 1146b7607d..f97f78c4fe 100644 --- a/src/stores/queryStore.ts +++ b/src/stores/queryStore.ts @@ -234,41 +234,27 @@ export const useQueryStore = defineStore("query", () => { * If editable, fetches table metadata and sets queryAnalysis + tableMeta on the tab. */ async function analyzeQueryEditability(tab: QueryTab, sql: string) { - console.log("[DBX] === analyzeQueryEditability CALLED ===", { tabId: tab.id }); - // Only analyze for query mode tabs with a valid result set - if (tab.mode !== "query") { - console.log("[DBX] SKIP: mode is not query, mode =", tab.mode); - return; - } + if (tab.mode !== "query") return; if (!tab.result || !tab.result.columns.length) { - console.log("[DBX] SKIP: no result or no columns", { - hasResult: !!tab.result, - colCount: tab.result?.columns.length, - }); tab.queryAnalysis = undefined; tab.tableMeta = undefined; return; } - console.log("[DBX] analyzeQueryEditability:", { - sql: sql.substring(0, 100), - mode: tab.mode, - columns: tab.result.columns, - connectionId: tab.connectionId, - database: tab.database, - schema: tab.schema, - }); - const analysis = analyzeEditableQuery(sql); - console.log("[DBX] analyzeEditableQuery result:", analysis); if (!analysis) { - console.log("[DBX] SKIP: analyzeEditableQuery returned null"); tab.queryAnalysis = undefined; tab.tableMeta = undefined; return; } - // Resolve schema: use explicit schema from analysis/tab, else fall back to db-type defaults + if (!tab.connectionId || !tab.database) { + tab.queryAnalysis = undefined; + tab.tableMeta = undefined; + return; + } + + // Resolve schema per database type const connStore = useConnectionStore(); const conn = connStore.getConfig(tab.connectionId); const dbType = conn?.db_type || ""; @@ -277,115 +263,38 @@ export const useQueryStore = defineStore("query", () => { if (dbType === "postgres") schema = "public"; else schema = ""; } - console.log("[DBX] Resolved schema:", schema, "dbType:", dbType); - if (analysis.selectStar && tab.connectionId && tab.database) { - try { - console.log("[DBX] Fetching columns:", { - connectionId: tab.connectionId, - database: tab.database, - schema, - tableName: analysis.tableName, - }); - const columns = await api.getColumns(tab.connectionId, tab.database, schema, analysis.tableName); - console.log( - "[DBX] Columns fetched:", - columns.length, - "columns, PKs:", - columns.filter((c) => c.is_primary_key).map((c) => c.name), - ); - const primaryKeys = columns.filter((c) => c.is_primary_key).map((c) => c.name); - if (primaryKeys.length === 0) { - console.log("[DBX] SKIP: no primary keys found for table", analysis.tableName); - tab.queryAnalysis = undefined; - tab.tableMeta = undefined; - return; - } + try { + const columns = await api.getColumns(tab.connectionId, tab.database, schema, analysis.tableName); + const primaryKeys = columns.filter((c) => c.is_primary_key).map((c) => c.name); - tab.tableMeta = { - schema: schema || undefined, - tableName: analysis.tableName, - columns, - primaryKeys, - }; - - // Verify all PK columns are present in result - if (!allPrimaryKeysPresent(primaryKeys, tab.result.columns)) { - console.log( - "[DBX] SKIP: not all PKs in result columns. PKs:", - primaryKeys, - "Result cols:", - tab.result.columns, - ); - tab.queryAnalysis = undefined; - tab.tableMeta = undefined; - return; - } - - console.log("[DBX] SET queryAnalysis + tableMeta successfully"); - tab.queryAnalysis = analysis; - } catch (err) { - console.error("[DBX] ERROR fetching columns:", err); - tab.queryAnalysis = undefined; - tab.tableMeta = undefined; - } - } else { - // Non-SELECT *: need to fetch table metadata to get PK info - console.log("[DBX] Non-SELECT* path, fetching table metadata"); - if (!tab.connectionId || !tab.database) { - console.log("[DBX] SKIP: no connectionId or database"); + if (primaryKeys.length === 0) { tab.queryAnalysis = undefined; tab.tableMeta = undefined; return; } - try { - const columns = await api.getColumns(tab.connectionId, tab.database, schema, analysis.tableName); - console.log( - "[DBX] Columns fetched:", - columns.length, - "columns, PKs:", - columns.filter((c) => c.is_primary_key).map((c) => c.name), - ); - const primaryKeys = columns.filter((c) => c.is_primary_key).map((c) => c.name); - if (primaryKeys.length === 0) { - console.log("[DBX] SKIP: no primary keys found for table", analysis.tableName); - tab.queryAnalysis = undefined; - tab.tableMeta = undefined; - return; - } - - // Check all PKs are in the result columns - if (!allPrimaryKeysPresent(primaryKeys, tab.result.columns)) { - console.log( - "[DBX] SKIP: not all PKs in result columns. PKs:", - primaryKeys, - "Result cols:", - tab.result.columns, - ); - tab.queryAnalysis = undefined; - tab.tableMeta = undefined; - return; - } - - console.log("[DBX] SET queryAnalysis + tableMeta successfully (non-SELECT* path)"); - tab.tableMeta = { - schema: schema || undefined, - tableName: analysis.tableName, - columns, - primaryKeys, - }; - tab.queryAnalysis = analysis; - } catch (err) { - console.error("[DBX] ERROR fetching columns (non-SELECT* path):", err); + if (!allPrimaryKeysPresent(primaryKeys, tab.result.columns)) { tab.queryAnalysis = undefined; tab.tableMeta = undefined; + return; } + + tab.tableMeta = { + schema: schema || undefined, + tableName: analysis.tableName, + columns, + primaryKeys, + }; + tab.queryAnalysis = analysis; + } catch (err) { + console.error("[DBX] ERROR fetching columns for editable query:", err); + tab.queryAnalysis = undefined; + tab.tableMeta = undefined; } } async function executeTabSql(id: string, sql: string) { - console.log("[DBX] === executeTabSql CALLED ===", { tabId: id, sql: sql.substring(0, 80) }); const tab = tabs.value.find((t) => t.id === id); if (!tab || !sql.trim()) return;