From d55a708d6a035e406f55d9a499f0fdb55b48fa7c Mon Sep 17 00:00:00 2001
From: Abeautifulsnow <lcprunstone@163.com>
Date: Thu, 2 Jul 2026 01:30:31 +0800
Subject: [PATCH] feat(query): add manual transaction controls

---
 apps/desktop/src/App.vue                      |  11 +
 .../src/components/layout/EditorToolbar.vue   |  63 ++-
 apps/desktop/src/i18n/locales/en.ts           |   7 +
 apps/desktop/src/i18n/locales/es.ts           |   7 +
 apps/desktop/src/i18n/locales/it.ts           |   7 +
 apps/desktop/src/i18n/locales/ja.ts           |   7 +
 apps/desktop/src/i18n/locales/pt-BR.ts        |   7 +
 apps/desktop/src/i18n/locales/zh-CN.ts        |   7 +
 apps/desktop/src/i18n/locales/zh-TW.ts        |   7 +
 .../__tests__/databaseFeatureSupport.spec.ts  |  28 ++
 apps/desktop/src/lib/api.ts                   |   4 +
 .../desktop/src/lib/databaseFeatureSupport.ts |  10 +
 apps/desktop/src/lib/http.ts                  |  16 +
 apps/desktop/src/lib/tauri.ts                 |  17 +
 apps/desktop/src/stores/connectionStore.ts    |   2 +
 apps/desktop/src/stores/queryStore.ts         | 133 +++++-
 apps/desktop/src/types/database.ts            |   7 +
 crates/dbx-core/src/connection.rs             |  20 +-
 crates/dbx-core/src/db/mysql.rs               |   4 +-
 crates/dbx-core/src/db/postgres.rs            |   2 +-
 crates/dbx-core/src/query.rs                  | 430 +++++++++++++++++-
 crates/dbx-core/src/query_result_export.rs    |   1 +
 crates/dbx-core/src/schema.rs                 |  10 +-
 .../dbx-core/tests/live_manual_transaction.rs | 105 +++++
 crates/dbx-web/src/routes/query.rs            |   2 +
 src-tauri/src/commands/query.rs               |  49 ++
 src-tauri/src/lib.rs                          |   4 +
 27 files changed, 927 insertions(+), 40 deletions(-)
 create mode 100644 apps/desktop/src/lib/__tests__/databaseFeatureSupport.spec.ts
 create mode 100644 crates/dbx-core/tests/live_manual_transaction.rs

diff --git a/apps/desktop/src/App.vue b/apps/desktop/src/App.vue
index b4099a1529..4b691bffa3 100644
--- a/apps/desktop/src/App.vue
+++ b/apps/desktop/src/App.vue
@@ -1622,8 +1622,19 @@ onUnmounted(() => {
                   :block-dangerous-redis-commands="blockDangerousRedisCommands"
                   :sql-keyword-case="settingsStore.editorSettings.sqlFormatter.keywordCase"
                   :database-required-signal="databaseRequiredTabId === activeTab.id ? databaseRequiredSignal : 0"
+                  :auto-commit="activeTab.autoCommit ?? true"
+                  :txn-session-id="activeTab?.txnSessionId"
+                  :txn-auto-rolled-back="activeTab?.txnAutoRolledBack"
                   @update:explain-mode="(m: 'explain' | 'autotrace') => (explainMode = m)"
                   @update:block-dangerous-redis-commands="(v: boolean) => (blockDangerousRedisCommands = v)"
+                  @update:auto-commit="
+                    (v: boolean) => {
+                      if (activeTab) queryStore.setAutoCommit(activeTab.id, v);
+                    }
+                  "
+                  @commit="activeTab && queryStore.commitTransaction(activeTab.id)"
+                  @rollback="activeTab && queryStore.rollbackTransaction(activeTab.id)"
+                  @dismiss-txn-rolled-back="activeTab && (activeTab.txnAutoRolledBack = false)"
                   @execute="requestActiveEditorExecute()"
                   @cancel="cancelActiveExecution()"
                   @explain="tryExplain()"
diff --git a/apps/desktop/src/components/layout/EditorToolbar.vue b/apps/desktop/src/components/layout/EditorToolbar.vue
index 3ee3d38c2b..6711f571d5 100644
--- a/apps/desktop/src/components/layout/EditorToolbar.vue
+++ b/apps/desktop/src/components/layout/EditorToolbar.vue
@@ -1,7 +1,7 @@
 <script setup lang="ts">
 import { computed, ref, watch, watchEffect } from "vue";
 import { useI18n } from "vue-i18n";
-import { Play, Loader2, Square, Database, Check, Table2, AlignLeft, GitBranch, Save, FolderOpen, Layers, X, Shield, Upload } from "@lucide/vue";
+import { Play, Loader2, Square, Database, Check, Table2, AlignLeft, GitBranch, Save, FolderOpen, Layers, X, Shield, Upload, RotateCcw, AlertTriangle } from "@lucide/vue";
 import { Button } from "@/components/ui/button";
 import { SearchableSelect } from "@/components/ui/searchable-select";
 import { Tooltip, TooltipTrigger, TooltipContent } from "@/components/ui/tooltip";
@@ -13,7 +13,7 @@ import { useSchemaOptions } from "@/composables/useSchemaOptions";
 import { connectionIconType } from "@/lib/connectionPresentation";
 import { formatDatabaseLabel, isDefaultDatabase } from "@/lib/defaultDatabase";
 import { connectionDisplayName } from "@/lib/tabPresentation";
-import { isSingleDatabase } from "@/lib/databaseCapabilities";
+import { isSingleDatabase, supportsTransaction as supportsTransactionFeature } from "@/lib/databaseCapabilities";
 import { hexToRgba } from "@/lib/color";
 import type { QueryTab, ConnectionConfig } from "@/types/database";
 
@@ -25,6 +25,9 @@ const props = defineProps<{
   blockDangerousRedisCommands?: boolean;
   sqlKeywordCase: "preserve" | "upper" | "lower";
   databaseRequiredSignal?: number;
+  autoCommit?: boolean;
+  txnSessionId?: string;
+  txnAutoRolledBack?: boolean;
 }>();
 
 const emit = defineEmits<{
@@ -43,6 +46,10 @@ const emit = defineEmits<{
   setDefaultDatabase: [];
   clearDefaultDatabase: [];
   "update:blockDangerousRedisCommands": [value: boolean];
+  "update:autoCommit": [value: boolean];
+  commit: [];
+  rollback: [];
+  dismissTxnRolledBack: [];
 }>();
 
 const { t } = useI18n();
@@ -64,12 +71,22 @@ const supportsExplain = computed(() => {
   return dbType !== "redis" && dbType !== "mongodb" && dbType !== "elasticsearch" && dbType !== "qdrant" && dbType !== "milvus" && dbType !== "weaviate" && dbType !== "chromadb" && dbType !== "etcd" && dbType !== "zookeeper" && dbType !== "mq" && dbType !== "nacos";
 });
 const isSingleDb = computed(() => isSingleDatabase(props.activeConnection?.db_type));
+const supportsTransaction = computed(() => supportsTransactionFeature(props.activeConnection?.db_type));
 const hasDefaultDatabaseOption = computed(() => activeDatabaseOptions.value.includes(""));
 const schemaDatabaseKey = computed(() => props.activeTab.database || (isSingleDb.value ? "_" : ""));
 const saveTooltip = computed(() => (props.activeTab.objectSource ? t("objects.saveSource") : t("toolbar.saveSql")));
 const canSaveSql = computed(() => !!props.activeTab.externalSqlPath || !!props.activeTab.sql.trim());
 const keywordCaseIsLower = computed(() => props.sqlKeywordCase === "lower");
 const keywordCaseToggleTooltip = computed(() => (keywordCaseIsLower.value ? t("toolbar.keywordCaseUpper") : t("toolbar.keywordCaseLower")));
+const transactionTooltip = computed(() => {
+  const isAgent = (props.activeConnection?.db_type as string) === "agent";
+  const isManual = props.autoCommit === false;
+  if (isAgent && isManual) return t("toolbar.manualTransactionAgent");
+  if (isAgent) return t("toolbar.autoCommitAgent");
+  return isManual ? t("toolbar.manualTransaction") : t("toolbar.autoCommit");
+});
+
+const isTransactionActive = computed(() => !!props.txnSessionId);
 
 const showSchemaSelector = computed(() => {
   const connection = props.activeConnection;
@@ -175,6 +192,41 @@ function connectionById(connectionId: string): ConnectionConfig | undefined {
       >
         <span class="font-bold" style="font-size: 9px">A</span>
       </Button>
+      <!-- Transaction toggle -->
+      <Tooltip v-if="supportsTransaction">
+        <TooltipTrigger as-child>
+          <Button
+            variant="ghost"
+            size="icon"
+            class="h-6 w-6"
+            :class="isTransactionActive || autoCommit === false ? 'text-orange-600 bg-orange-100 dark:text-orange-300 dark:bg-orange-900/30' : 'text-muted-foreground/50'"
+            :disabled="activeTab.isExecuting || activeTab.isExplaining"
+            @click="emit('update:autoCommit', autoCommit === false)"
+          >
+            <span class="font-bold" style="font-size: 9px">Tx</span>
+          </Button>
+        </TooltipTrigger>
+        <TooltipContent>{{ transactionTooltip }}</TooltipContent>
+      </Tooltip>
+      <!-- Commit button (only when transaction is active) -->
+      <Tooltip v-if="isTransactionActive">
+        <TooltipTrigger as-child>
+          <Button variant="ghost" size="icon" class="h-6 w-6 text-green-600 hover:bg-green-500/10 hover:text-green-700 dark:text-green-300 dark:hover:text-green-200" :disabled="activeTab.isExecuting" @click="emit('commit')">
+            <Check class="h-3.5 w-3.5" />
+          </Button>
+        </TooltipTrigger>
+        <TooltipContent>{{ t("toolbar.commit") }}</TooltipContent>
+      </Tooltip>
+
+      <!-- Rollback button (only when transaction is active) -->
+      <Tooltip v-if="isTransactionActive">
+        <TooltipTrigger as-child>
+          <Button variant="ghost" size="icon" class="h-6 w-6 text-red-600 hover:bg-red-500/10 hover:text-red-700 dark:text-red-300 dark:hover:text-red-200" :disabled="activeTab.isExecuting" @click="emit('rollback')">
+            <RotateCcw class="h-3.5 w-3.5" />
+          </Button>
+        </TooltipTrigger>
+        <TooltipContent>{{ t("toolbar.rollback") }}</TooltipContent>
+      </Tooltip>
       <Tooltip>
         <TooltipTrigger as-child>
           <Button variant="ghost" size="icon" class="h-6 w-6 text-amber-600 hover:bg-amber-500/10 hover:text-amber-700 dark:text-amber-300 dark:hover:text-amber-200" :disabled="activeTab.isExecuting || activeTab.isExplaining || !activeTab.sql.trim()" @click="emit('formatSql')">
@@ -342,6 +394,13 @@ function connectionById(connectionId: string): ConnectionConfig | undefined {
       <span class="truncate">{{ activeTab.tableMeta.columns.length }} {{ t("tree.columns") }}</span>
     </div>
   </div>
+  <div v-if="txnAutoRolledBack" class="flex items-center gap-2 px-3 py-1 text-xs bg-amber-500/10 text-amber-700 dark:text-amber-300 border-b border-amber-500/20">
+    <AlertTriangle class="h-3.5 w-3.5 shrink-0" />
+    <span>{{ t("toolbar.txnAutoRolledBack") }}</span>
+    <Button variant="ghost" size="icon" class="h-5 w-5 ml-auto" @click="emit('dismissTxnRolledBack')">
+      <X class="h-3 w-3" />
+    </Button>
+  </div>
 </template>
 
 <style scoped>
diff --git a/apps/desktop/src/i18n/locales/en.ts b/apps/desktop/src/i18n/locales/en.ts
index 987f4ee229..35ea986801 100644
--- a/apps/desktop/src/i18n/locales/en.ts
+++ b/apps/desktop/src/i18n/locales/en.ts
@@ -34,6 +34,13 @@ export default {
     formatSqlFailed: "Failed to format SQL",
     keywordCaseLower: "Use lower-case SQL keywords",
     keywordCaseUpper: "Use upper-case SQL keywords",
+    autoCommit: "Auto Commit",
+    manualTransaction: "Manual Transaction",
+    autoCommitAgent: "Auto Commit (depends on Agent backend)",
+    manualTransactionAgent: "Manual Transaction (depends on Agent backend)",
+    commit: "Commit",
+    rollback: "Rollback",
+    txnAutoRolledBack: "Transaction auto-rolled back after 5 minutes of inactivity",
     previewSql: "Preview SQL",
     hidePreviewSql: "Hide SQL Preview",
     saveSql: "Save to SQL Library",
diff --git a/apps/desktop/src/i18n/locales/es.ts b/apps/desktop/src/i18n/locales/es.ts
index 3a80625546..a0f9cbe77a 100644
--- a/apps/desktop/src/i18n/locales/es.ts
+++ b/apps/desktop/src/i18n/locales/es.ts
@@ -36,6 +36,13 @@ export default withEnglishFallback({
     formatSqlFailed: "Error al formatear el SQL",
     keywordCaseLower: "Usar palabras clave SQL en minúsculas",
     keywordCaseUpper: "Usar palabras clave SQL en mayúsculas",
+    autoCommit: "Auto Commit",
+    manualTransaction: "Transacción Manual",
+    autoCommitAgent: "Auto Commit (depende del backend del Agente)",
+    manualTransactionAgent: "Transacción Manual (depende del backend del Agente)",
+    commit: "Confirmar",
+    rollback: "Revertir",
+    txnAutoRolledBack: "Transacción revertida automáticamente tras 5 minutos de inactividad",
     previewSql: "Vista previa de SQL",
     hidePreviewSql: "Ocultar vista previa de SQL",
     saveSql: "Guardar en biblioteca SQL",
diff --git a/apps/desktop/src/i18n/locales/it.ts b/apps/desktop/src/i18n/locales/it.ts
index 59ecf0cece..5863b65f67 100644
--- a/apps/desktop/src/i18n/locales/it.ts
+++ b/apps/desktop/src/i18n/locales/it.ts
@@ -35,6 +35,13 @@ export default withEnglishFallback({
     formatSqlFailed: "Impossibile formattare SQL",
     keywordCaseLower: "Usa parole chiave SQL minuscole",
     keywordCaseUpper: "Usa parole chiave SQL maiuscole",
+    autoCommit: "Auto Commit",
+    manualTransaction: "Transazione Manuale",
+    autoCommitAgent: "Auto Commit (dipende dal backend dell'Agente)",
+    manualTransactionAgent: "Transazione Manuale (dipende dal backend dell'Agente)",
+    commit: "Commit",
+    rollback: "Rollback",
+    txnAutoRolledBack: "Transazione annullata automaticamente dopo 5 minuti di inattività",
     previewSql: "Anteprima SQL",
     hidePreviewSql: "Nascondi anteprima SQL",
     saveSql: "Salva nella Libreria SQL",
diff --git a/apps/desktop/src/i18n/locales/ja.ts b/apps/desktop/src/i18n/locales/ja.ts
index d2b4fb6565..a250af82de 100644
--- a/apps/desktop/src/i18n/locales/ja.ts
+++ b/apps/desktop/src/i18n/locales/ja.ts
@@ -36,6 +36,13 @@ export default withEnglishFallback({
     formatSqlFailed: "SQLのフォーマットに失敗しました",
     keywordCaseLower: "SQLキーワードを小文字にする",
     keywordCaseUpper: "SQLキーワードを大文字にする",
+    autoCommit: "自動コミット",
+    manualTransaction: "手動トランザクション",
+    autoCommitAgent: "自動コミット（Agentバックエンドに依存）",
+    manualTransactionAgent: "手動トランザクション（Agentバックエンドに依存）",
+    commit: "コミット",
+    rollback: "ロールバック",
+    txnAutoRolledBack: "5分間操作がなかったため、トランザクションは自動的にロールバックされました",
     previewSql: "SQLをプレビュー",
     hidePreviewSql: "SQLプレビューを非表示",
     saveSql: "SQLライブラリに保存",
diff --git a/apps/desktop/src/i18n/locales/pt-BR.ts b/apps/desktop/src/i18n/locales/pt-BR.ts
index 922c1a81e6..4d5e1f1d79 100644
--- a/apps/desktop/src/i18n/locales/pt-BR.ts
+++ b/apps/desktop/src/i18n/locales/pt-BR.ts
@@ -36,6 +36,13 @@ export default withEnglishFallback({
     formatSqlFailed: "Falha ao formatar SQL",
     keywordCaseLower: "Usar palavras-chave SQL em minúsculas",
     keywordCaseUpper: "Usar palavras-chave SQL em maiúsculas",
+    autoCommit: "Auto Commit",
+    manualTransaction: "Transação Manual",
+    autoCommitAgent: "Auto Commit (depende do backend do Agente)",
+    manualTransactionAgent: "Transação Manual (depende do backend do Agente)",
+    commit: "Confirmar",
+    rollback: "Reverter",
+    txnAutoRolledBack: "Transação revertida automaticamente após 5 minutos de inatividade",
     previewSql: "Visualizar SQL",
     hidePreviewSql: "Ocultar Visualização SQL",
     saveSql: "Salvar na Biblioteca SQL",
diff --git a/apps/desktop/src/i18n/locales/zh-CN.ts b/apps/desktop/src/i18n/locales/zh-CN.ts
index b508458692..fc9b7ce476 100644
--- a/apps/desktop/src/i18n/locales/zh-CN.ts
+++ b/apps/desktop/src/i18n/locales/zh-CN.ts
@@ -36,6 +36,13 @@ export default withEnglishFallback({
     formatSqlFailed: "SQL 格式化失败",
     keywordCaseLower: "使用小写 SQL 关键字",
     keywordCaseUpper: "使用大写 SQL 关键字",
+    autoCommit: "自动提交",
+    manualTransaction: "手动事务",
+    autoCommitAgent: "自动提交（取决于 Agent 后端数据库）",
+    manualTransactionAgent: "手动事务（取决于 Agent 后端数据库）",
+    commit: "提交",
+    rollback: "回滚",
+    txnAutoRolledBack: "事务已因 5 分钟无操作而自动回滚",
     previewSql: "预览 SQL",
     hidePreviewSql: "隐藏 SQL 预览",
     saveSql: "保存到 SQL 库",
diff --git a/apps/desktop/src/i18n/locales/zh-TW.ts b/apps/desktop/src/i18n/locales/zh-TW.ts
index cc9545847a..98dfd39552 100644
--- a/apps/desktop/src/i18n/locales/zh-TW.ts
+++ b/apps/desktop/src/i18n/locales/zh-TW.ts
@@ -36,6 +36,13 @@ export default withEnglishFallback({
     formatSqlFailed: "SQL 格式化失敗",
     keywordCaseLower: "使用小寫 SQL 關鍵字",
     keywordCaseUpper: "使用大寫 SQL 關鍵字",
+    autoCommit: "自動提交",
+    manualTransaction: "手動事務",
+    autoCommitAgent: "自動提交（取決於 Agent 後端資料庫）",
+    manualTransactionAgent: "手動事務（取決於 Agent 後端資料庫）",
+    commit: "提交",
+    rollback: "回溯",
+    txnAutoRolledBack: "事務已因 5 分鐘無操作而自動回溯",
     previewSql: "預覽 SQL",
     hidePreviewSql: "隱藏 SQL 預覽",
     saveSql: "儲存到 SQL 庫",
diff --git a/apps/desktop/src/lib/__tests__/databaseFeatureSupport.spec.ts b/apps/desktop/src/lib/__tests__/databaseFeatureSupport.spec.ts
new file mode 100644
index 0000000000..d60c20d370
--- /dev/null
+++ b/apps/desktop/src/lib/__tests__/databaseFeatureSupport.spec.ts
@@ -0,0 +1,28 @@
+import { describe, expect, it } from "vitest";
+import { supportsTransaction } from "@/lib/databaseFeatureSupport";
+
+describe("supportsTransaction", () => {
+  it("returns true for supported database types", () => {
+    expect(supportsTransaction("postgres")).toBe(true);
+    expect(supportsTransaction("mysql")).toBe(true);
+  });
+
+  it("returns false for unsupported database types", () => {
+    expect(supportsTransaction("redis")).toBe(false);
+    expect(supportsTransaction("mongodb")).toBe(false);
+    expect(supportsTransaction("duckdb")).toBe(false);
+    expect(supportsTransaction("qdrant")).toBe(false);
+    expect(supportsTransaction("turso")).toBe(false);
+    expect(supportsTransaction("sqlite")).toBe(false);
+    expect(supportsTransaction("clickhouse")).toBe(false);
+    expect(supportsTransaction("sqlserver")).toBe(false);
+    expect(supportsTransaction("oracle")).toBe(false);
+    expect(supportsTransaction("dameng")).toBe(false);
+    expect(supportsTransaction("rqlite")).toBe(false);
+    expect(supportsTransaction("agent")).toBe(false);
+  });
+
+  it("returns false for undefined or empty input", () => {
+    expect(supportsTransaction(undefined)).toBe(false);
+  });
+});
diff --git a/apps/desktop/src/lib/api.ts b/apps/desktop/src/lib/api.ts
index 85175560ae..2f45a04b5e 100644
--- a/apps/desktop/src/lib/api.ts
+++ b/apps/desktop/src/lib/api.ts
@@ -142,6 +142,10 @@ export const executeMulti = forward("executeMulti");
 export const executeBatch = forward("executeBatch");
 export const executeScript = forward("executeScript");
 export const executeInTransaction = forward("executeInTransaction");
+export const beginManualTransaction = forward("beginManualTransaction");
+export const executeInManualTransaction = forward("executeInManualTransaction");
+export const commitManualTransaction = forward("commitManualTransaction");
+export const rollbackManualTransaction = forward("rollbackManualTransaction");
 export const cancelQuery = forward("cancelQuery");
 export const closeQuerySession = forward("closeQuerySession");
 export const closeClientConnectionSession = forward("closeClientConnectionSession");
diff --git a/apps/desktop/src/lib/databaseFeatureSupport.ts b/apps/desktop/src/lib/databaseFeatureSupport.ts
index dc072da93d..dc46fcbeb9 100644
--- a/apps/desktop/src/lib/databaseFeatureSupport.ts
+++ b/apps/desktop/src/lib/databaseFeatureSupport.ts
@@ -88,3 +88,13 @@ export function supportsTableTruncate(dbType?: DatabaseType): boolean {
 export function usesPostgresLikeStructureCopy(dbType?: DatabaseType): boolean {
   return !!dbType && PG_LIKE_STRUCTURE_TYPES.has(dbType);
 }
+
+const TRANSACTION_SUPPORTED_TYPES: readonly string[] = ["postgres", "mysql"];
+
+/**
+ * Returns true if the given database type supports explicit transaction control
+ * (i.e. toggling between auto-commit and manual transaction mode via BEGIN/COMMIT).
+ */
+export function supportsTransaction(dbType?: string): boolean {
+  return !!dbType && TRANSACTION_SUPPORTED_TYPES.includes(dbType);
+}
diff --git a/apps/desktop/src/lib/http.ts b/apps/desktop/src/lib/http.ts
index d752aed47d..ee89281f74 100644
--- a/apps/desktop/src/lib/http.ts
+++ b/apps/desktop/src/lib/http.ts
@@ -657,6 +657,22 @@ export async function executeInTransaction(connectionId: string, database: strin
   return post("/api/query/execute-in-transaction", { connectionId, database, statements, schema });
 }
 
+export async function beginManualTransaction(_connectionId: string, _database: string, _schema?: string): Promise<string> {
+  throw new Error("Manual transaction management is only available in the desktop app.");
+}
+
+export async function executeInManualTransaction(_txnSessionId: string, _sql: string, _database: string, _schema?: string, _maxRows?: number): Promise<QueryResult[]> {
+  throw new Error("Manual transaction management is only available in the desktop app.");
+}
+
+export async function commitManualTransaction(_txnSessionId: string): Promise<QueryResult> {
+  throw new Error("Manual transaction management is only available in the desktop app.");
+}
+
+export async function rollbackManualTransaction(_txnSessionId: string): Promise<QueryResult> {
+  throw new Error("Manual transaction management is only available in the desktop app.");
+}
+
 export async function cancelQuery(executionId: string): Promise<boolean> {
   const result = await post<boolean | { cancelled?: boolean }>("/api/query/cancel", { executionId });
   return typeof result === "boolean" ? result : result.cancelled === true;
diff --git a/apps/desktop/src/lib/tauri.ts b/apps/desktop/src/lib/tauri.ts
index 95700300fb..9f40b42e11 100644
--- a/apps/desktop/src/lib/tauri.ts
+++ b/apps/desktop/src/lib/tauri.ts
@@ -615,6 +615,7 @@ export async function executeMulti(
     resultSessionId?: string;
     clientSessionId?: string;
     timeoutSecs?: number;
+    useTransaction?: boolean;
   },
 ): Promise<QueryResult[]> {
   return invoke("execute_multi", { connectionId, database, sql, schema, executionId, ...options });
@@ -648,6 +649,22 @@ export async function executeInTransaction(connectionId: string, database: strin
   return invoke("execute_in_transaction", { connectionId, database, statements, schema });
 }
 
+export async function beginManualTransaction(connectionId: string, database: string, schema?: string): Promise<string> {
+  return invoke("begin_manual_transaction", { connectionId, database, schema });
+}
+
+export async function executeInManualTransaction(txnSessionId: string, sql: string, database: string, schema?: string, maxRows?: number): Promise<QueryResult[]> {
+  return invoke("execute_in_manual_transaction", { txnSessionId, sql, database, schema, maxRows });
+}
+
+export async function commitManualTransaction(txnSessionId: string): Promise<QueryResult> {
+  return invoke("commit_manual_transaction", { txnSessionId });
+}
+
+export async function rollbackManualTransaction(txnSessionId: string): Promise<QueryResult> {
+  return invoke("rollback_manual_transaction", { txnSessionId });
+}
+
 export async function analyzeSqlReferences(sql: string, dialect?: string): Promise<SqlReferenceAnalysis> {
   return invoke("analyze_sql_references", { sql, dialect });
 }
diff --git a/apps/desktop/src/stores/connectionStore.ts b/apps/desktop/src/stores/connectionStore.ts
index 96c0cbb7f1..7833931b40 100644
--- a/apps/desktop/src/stores/connectionStore.ts
+++ b/apps/desktop/src/stores/connectionStore.ts
@@ -1317,6 +1317,7 @@ export const useConnectionStore = defineStore("connection", () => {
         queryStore.releaseConnectionTabs(connectionId);
         break;
       case "keep-tabs-keep-results":
+        queryStore.rollbackConnectionTransactions(connectionId);
         break;
     }
     connectedIds.value.delete(connectionId);
@@ -1349,6 +1350,7 @@ export const useConnectionStore = defineStore("connection", () => {
         queryStore.releaseDatabaseTabs(connectionId, database);
         break;
       case "keep-tabs-keep-results":
+        queryStore.rollbackDatabaseTransactions(connectionId, database);
         break;
     }
     const node = findDatabaseTreeNode(treeNodes.value, connectionId, database);
diff --git a/apps/desktop/src/stores/queryStore.ts b/apps/desktop/src/stores/queryStore.ts
index 52f540fe78..c808f66120 100644
--- a/apps/desktop/src/stores/queryStore.ts
+++ b/apps/desktop/src/stores/queryStore.ts
@@ -859,6 +859,7 @@ export const useQueryStore = defineStore("query", () => {
     const idx = tabs.value.findIndex((t) => t.id === id);
     if (idx < 0) return;
     clearDataGridPendingSnapshotsForTab(id);
+    if (tabs.value[idx].txnSessionId) void rollbackTransaction(id);
     if (tabs.value[idx].isExecuting) void cancelTabExecution(id);
     if (tabs.value[idx].isExplaining) void cancelTabExplain(id);
     void closeResultSession(tabs.value[idx]);
@@ -987,6 +988,7 @@ export const useQueryStore = defineStore("query", () => {
       .filter((tab) => closingIds.has(tab.id))
       .forEach((tab) => {
         clearDataGridPendingSnapshotsForTab(tab.id);
+        if (tab.txnSessionId) void rollbackTransaction(tab.id);
         if (tab.isExecuting) void cancelTabExecution(tab.id);
         if (tab.isExplaining) void cancelTabExplain(tab.id);
         void closeResultSession(tab);
@@ -1015,6 +1017,7 @@ export const useQueryStore = defineStore("query", () => {
     tabs.value
       .filter((tab) => predicate(tab))
       .forEach((tab) => {
+        rollbackTabTransaction(tab, { resetAutoCommit: true });
         if (tab.isExecuting) void cancelTabExecution(tab.id);
         if (tab.isExplaining) void cancelTabExplain(tab.id);
         void closeResultSession(tab);
@@ -1035,6 +1038,18 @@ export const useQueryStore = defineStore("query", () => {
     return tabs.value.some((tab) => tab.connectionId === connectionId && tab.database === database);
   }
 
+  function rollbackTabsWhere(predicate: (tab: QueryTab) => boolean, options?: { resetAutoCommit?: boolean }) {
+    tabs.value.filter((tab) => predicate(tab)).forEach((tab) => rollbackTabTransaction(tab, options));
+  }
+
+  function rollbackConnectionTransactions(connectionId: string) {
+    rollbackTabsWhere((tab) => tab.connectionId === connectionId, { resetAutoCommit: true });
+  }
+
+  function rollbackDatabaseTransactions(connectionId: string, database: string) {
+    rollbackTabsWhere((tab) => tab.connectionId === connectionId && tab.database === database, { resetAutoCommit: true });
+  }
+
   function updateSql(id: string, sql: string) {
     const tab = tabs.value.find((t) => t.id === id);
     if (tab) {
@@ -1042,6 +1057,49 @@ export const useQueryStore = defineStore("query", () => {
     }
   }
 
+  function setAutoCommit(id: string, autoCommit: boolean) {
+    const tab = tabs.value.find((t) => t.id === id);
+    if (tab) {
+      const wasManual = tab.autoCommit === false;
+      tab.autoCommit = autoCommit;
+      if (autoCommit && wasManual) {
+        if (tab.txnSessionId) {
+          void rollbackTransaction(id);
+        } else {
+          tab.txnAutoRolledBack = false;
+        }
+      }
+    }
+  }
+
+  function rollbackTabTransaction(tab: QueryTab, options?: { resetAutoCommit?: boolean }) {
+    if (tab.txnSessionId) void rollbackTransaction(tab.id);
+    if (options?.resetAutoCommit) tab.autoCommit = true;
+    tab.txnAutoRolledBack = false;
+  }
+
+  async function commitTransaction(id: string) {
+    const tab = tabs.value.find((t) => t.id === id);
+    if (!tab?.txnSessionId) return;
+    try {
+      await api.commitManualTransaction(tab.txnSessionId);
+    } finally {
+      tab.txnSessionId = undefined;
+      tab.txnAutoRolledBack = false;
+    }
+  }
+
+  async function rollbackTransaction(id: string) {
+    const tab = tabs.value.find((t) => t.id === id);
+    if (!tab?.txnSessionId) return;
+    try {
+      await api.rollbackManualTransaction(tab.txnSessionId);
+    } finally {
+      tab.txnSessionId = undefined;
+      tab.txnAutoRolledBack = false;
+    }
+  }
+
   function updateEditorViewport(id: string, viewport: { scrollTop: number; scrollLeft: number }) {
     const tab = tabs.value.find((t) => t.id === id);
     if (!tab) return;
@@ -1154,6 +1212,7 @@ export const useQueryStore = defineStore("query", () => {
   function updateDatabase(id: string, database: string) {
     const tab = tabs.value.find((t) => t.id === id);
     if (!tab || tab.database === database) return;
+    rollbackTabTransaction(tab);
     void closeResultSession(tab);
     void closeClientConnectionSession(tab);
     tab.database = database;
@@ -1170,6 +1229,7 @@ export const useQueryStore = defineStore("query", () => {
   function updateSchema(id: string, schema: string | undefined) {
     const tab = tabs.value.find((t) => t.id === id);
     if (!tab || tab.schema === schema) return;
+    rollbackTabTransaction(tab);
     tab.schema = schema;
     if (tab.mode === "objects") tab.objectBrowser = { ...tab.objectBrowser, schema };
   }
@@ -1177,6 +1237,7 @@ export const useQueryStore = defineStore("query", () => {
   function updateConnection(id: string, connectionId: string, database = "") {
     const tab = tabs.value.find((t) => t.id === id);
     if (!tab || tab.connectionId === connectionId) return;
+    rollbackTabTransaction(tab, { resetAutoCommit: true });
     void closeResultSession(tab);
     void closeClientConnectionSession(tab);
     tab.connectionId = connectionId;
@@ -1912,32 +1973,44 @@ export const useQueryStore = defineStore("query", () => {
         return;
       }
 
-      console.info("[DBX][executeTabSql:execute-multi:start]", { traceId, elapsed: elapsed() });
-      const clientSessionId = tab.mode === "query" || tab.mode === "data" ? tabClientSessionId(tab) : undefined;
-      const executionOptions = {
-        ...(typeof pageLimit === "number"
-          ? useAgentResultSession
-            ? {
-                fetchSize: pageLimit,
-                pageSize: pageLimit,
-                resultSessionId: options?.pagination?.sessionId,
-              }
-            : { maxRows: pageLimit, fetchSize: pageLimit }
-          : {}),
-        ...(clientSessionId ? { clientSessionId } : {}),
-        timeoutSecs: queryTimeoutSecs,
-      };
       const executionSchema = connectionUsesSchemaExecutionContext(conn) ? tab.schema || tab.database : tab.mode === "data" || connectionUsesDatabaseObjectTreeMode(conn) ? undefined : tab.schema;
-      console.info("[DBX][executeTabSql:execute-multi:invoke]", {
-        traceId,
-        elapsed: elapsed(),
-        executionSchema,
-        optionKeys: Object.keys(executionOptions),
-        clientSession: Boolean(clientSessionId),
-      });
-      const executionPromise = api.executeMulti(tab.connectionId, tab.database, sqlToExecute, executionSchema, executionId, executionOptions);
       const frontendTimeoutSecs = Math.max(queryTimeoutSecs * 2, 60);
       const sourceLabelDatabase = tab.database || conn?.database;
+
+      let executionPromise: Promise<QueryResult[]>;
+      if (tab.autoCommit === false) {
+        if (!tab.txnSessionId) {
+          console.info("[DBX][executeTabSql:begin-manual-txn:start]", { traceId, elapsed: elapsed() });
+          tab.txnSessionId = await api.beginManualTransaction(tab.connectionId, tab.database, executionSchema);
+          console.info("[DBX][executeTabSql:begin-manual-txn:done]", { traceId, txnSessionId: tab.txnSessionId, elapsed: elapsed() });
+        }
+        console.info("[DBX][executeTabSql:execute-in-txn:invoke]", { traceId, txnSessionId: tab.txnSessionId, elapsed: elapsed() });
+        executionPromise = api.executeInManualTransaction(tab.txnSessionId, sqlToExecute, tab.database, executionSchema, pageLimit);
+      } else {
+        console.info("[DBX][executeTabSql:execute-multi:start]", { traceId, elapsed: elapsed() });
+        const clientSessionId = tab.mode === "query" || tab.mode === "data" ? tabClientSessionId(tab) : undefined;
+        const executionOptions = {
+          ...(typeof pageLimit === "number"
+            ? useAgentResultSession
+              ? {
+                  fetchSize: pageLimit,
+                  pageSize: pageLimit,
+                  resultSessionId: options?.pagination?.sessionId,
+                }
+              : { maxRows: pageLimit, fetchSize: pageLimit }
+            : {}),
+          ...(clientSessionId ? { clientSessionId } : {}),
+          timeoutSecs: queryTimeoutSecs,
+        };
+        console.info("[DBX][executeTabSql:execute-multi:invoke]", {
+          traceId,
+          elapsed: elapsed(),
+          executionSchema,
+          optionKeys: Object.keys(executionOptions),
+          clientSession: Boolean(clientSessionId),
+        });
+        executionPromise = api.executeMulti(tab.connectionId, tab.database, sqlToExecute, executionSchema, executionId, executionOptions);
+      }
       const results = annotateQueryResultSources(markQueryResultsRowsRaw(await withFrontendQueryTimeout(executionPromise, queryTimeoutSecs === 0 ? 0 : frontendTimeoutSecs, t("editor.queryTimeoutError", { seconds: frontendTimeoutSecs }))), queryBaseSql, sourceLabelDatabase, effectiveDbType);
       console.info("[DBX][executeTabSql:execute-multi:done]", {
         traceId,
@@ -2016,6 +2089,15 @@ export const useQueryStore = defineStore("query", () => {
       console.error("[DBX][executeTabSql:error]", { traceId, elapsed: elapsed(), error: e });
       // Sync connection state if the error indicates a lost connection
       useConnectionStore().recordConnectionLostError(tab.connectionId, e);
+      // Handle manual transaction auto-rollback (e.g. deadlock detected by server,
+      // statement error inside a manual transaction, or idle timeout).
+      if (tab.autoCommit === false) {
+        const errMsg: string = e?.message ?? String(e);
+        if (/rolled.?back/i.test(errMsg) || errMsg.includes("已自动回滚")) {
+          tab.txnSessionId = undefined;
+          tab.txnAutoRolledBack = true;
+        }
+      }
       const current = tabs.value.find((t) => t.id === id);
       if (current?.executionId === executionId) {
         current.result = toErrorResult(e);
@@ -2581,9 +2663,14 @@ export const useQueryStore = defineStore("query", () => {
     releaseConnectionTabs,
     releaseDatabaseTabs,
     isDatabaseOpen,
+    rollbackConnectionTransactions,
+    rollbackDatabaseTransactions,
     updateSql,
     updateEditorViewport,
     updateEditorSelection,
+    setAutoCommit,
+    commitTransaction,
+    rollbackTransaction,
     renameTab,
     openObjectBrowser,
     openUserAdmin,
diff --git a/apps/desktop/src/types/database.ts b/apps/desktop/src/types/database.ts
index 1fbd284efb..a2b78e65cf 100644
--- a/apps/desktop/src/types/database.ts
+++ b/apps/desktop/src/types/database.ts
@@ -687,6 +687,13 @@ export interface QueryTab {
   resultEvicted?: boolean;
   whereInput?: string;
   previewSql?: string;
+  /** Whether to use auto-commit mode (default true). When false, multiple statements are
+   *  wrapped in a single transaction. */
+  autoCommit?: boolean;
+  /** Session ID for an active manual transaction, set after beginManualTransaction */
+  txnSessionId?: string;
+  /** Set to true when a manual transaction was auto-rolled back due to inactivity */
+  txnAutoRolledBack?: boolean;
 }
 
 export interface SavedSqlFolder {
diff --git a/crates/dbx-core/src/connection.rs b/crates/dbx-core/src/connection.rs
index d9baf52f51..0e95d5da67 100644
--- a/crates/dbx-core/src/connection.rs
+++ b/crates/dbx-core/src/connection.rs
@@ -2,7 +2,7 @@ use std::collections::{HashMap, HashSet};
 use std::path::PathBuf;
 use std::sync::Arc;
 use std::time::{Duration, Instant};
-use tokio::sync::RwLock;
+use tokio::sync::{Mutex, RwLock};
 use tokio::task::JoinHandle;
 
 use mysql_async::prelude::Queryable;
@@ -96,6 +96,22 @@ pub enum PoolKind {
     Nacos,
 }
 
+/// Held connection for a manual transaction session
+pub enum TxnConnection {
+    Postgres(Box<deadpool_postgres::Object>),
+    Mysql(mysql_async::Conn),
+}
+
+pub struct TransactionSession {
+    pub connection: Arc<Mutex<TxnConnection>>,
+    pub pool_key: String,
+    pub last_activity: std::time::Instant,
+    pub busy: bool,
+    pub connection_id: String,
+    pub database: String,
+    pub schema: Option<String>,
+}
+
 macro_rules! agent_connection_pool_database_type {
     () => {
         DatabaseType::Dameng
@@ -152,6 +168,7 @@ pub struct AppState {
     /// PostgreSQL TLS cancel context, keyed by pool_key.
     /// Used to reconstruct a TLS connector compatible with the original connection when cancelling.
     postgres_cancel_contexts: Arc<RwLock<HashMap<String, db::postgres::PostgresCancelContext>>>,
+    pub transaction_sessions: Arc<RwLock<HashMap<String, TransactionSession>>>,
     #[cfg(feature = "mq-admin")]
     pub mq_registry: crate::mq::MqAdminRegistry,
 }
@@ -419,6 +436,7 @@ impl AppState {
             ),
             nacos_registry: crate::nacos::NacosAdminRegistry::new(),
             postgres_cancel_contexts: Arc::new(RwLock::new(HashMap::new())),
+            transaction_sessions: Arc::new(RwLock::new(HashMap::new())),
             #[cfg(feature = "mq-admin")]
             mq_registry: crate::mq::MqAdminRegistry::new(),
         }
diff --git a/crates/dbx-core/src/db/mysql.rs b/crates/dbx-core/src/db/mysql.rs
index 08cea22f6a..2343ca9494 100644
--- a/crates/dbx-core/src/db/mysql.rs
+++ b/crates/dbx-core/src/db/mysql.rs
@@ -218,7 +218,7 @@ fn mysql_bytes_to_json(bytes: Vec<u8>, column: &mysql_async::Column) -> serde_js
 /// Map a MySQL column to a user-facing type name for the result-grid header.
 /// Returns the bare lowercase type name (no length/precision/signedness), which
 /// is enough for display; unknown variants fall back to a lowercased debug name.
-fn mysql_column_type_name(ty: ColumnType) -> String {
+pub(crate) fn mysql_column_type_name(ty: ColumnType) -> String {
     use mysql_async::consts::ColumnType::*;
     match ty {
         MYSQL_TYPE_TINY => "tinyint",
@@ -251,7 +251,7 @@ fn mysql_column_type_name(ty: ColumnType) -> String {
     .to_string()
 }
 
-fn mysql_value_to_json(row: &mysql_async::Row, idx: usize) -> serde_json::Value {
+pub(crate) fn mysql_value_to_json(row: &mysql_async::Row, idx: usize) -> serde_json::Value {
     let Some(column) = row.columns_ref().get(idx) else {
         return serde_json::Value::Null;
     };
diff --git a/crates/dbx-core/src/db/postgres.rs b/crates/dbx-core/src/db/postgres.rs
index e965470c60..279c3bf503 100644
--- a/crates/dbx-core/src/db/postgres.rs
+++ b/crates/dbx-core/src/db/postgres.rs
@@ -304,7 +304,7 @@ fn format_pg_timestamptz(value: DateTime<Local>) -> String {
     value.to_rfc3339()
 }
 
-fn pg_value_to_json(row: &Row, idx: usize, type_name: &str) -> serde_json::Value {
+pub(crate) fn pg_value_to_json(row: &Row, idx: usize, type_name: &str) -> serde_json::Value {
     let upper = type_name.to_uppercase();
 
     if upper == "BYTEA" {
diff --git a/crates/dbx-core/src/query.rs b/crates/dbx-core/src/query.rs
index bdae7d9f4d..ac0dfdfb0e 100644
--- a/crates/dbx-core/src/query.rs
+++ b/crates/dbx-core/src/query.rs
@@ -2,6 +2,7 @@
 use chrono::{DateTime, Duration as ChronoDuration, NaiveDate, NaiveDateTime, NaiveTime, Utc};
 #[cfg(feature = "duckdb-bundled")]
 use duckdb::types::{TimeUnit, Value, ValueRef};
+use futures::StreamExt;
 use mysql_async::prelude::Queryable;
 use sqlparser::ast::{visit_relations_mut, Ident, ObjectName, ObjectNamePart, ObjectType, Statement};
 use sqlparser::dialect::{GenericDialect, PostgreSqlDialect};
@@ -9,6 +10,7 @@ use sqlparser::parser::Parser;
 use std::collections::HashSet;
 use std::future::Future;
 use std::ops::ControlFlow;
+use std::sync::Arc;
 use std::time::Duration;
 #[cfg(feature = "duckdb-bundled")]
 use tokio::task::JoinHandle;
@@ -17,13 +19,13 @@ use tokio::time::sleep;
 use tokio::time::timeout;
 use tokio_util::sync::CancellationToken;
 
-use crate::connection::{AppState, PoolKind};
+use crate::connection::{AppState, PoolKind, TransactionSession, TxnConnection};
 use crate::database_capabilities;
 use crate::db;
 use crate::models::connection::{ConnectionConfig, DatabaseType};
 #[cfg(feature = "duckdb-bundled")]
 use crate::sql::starts_with_duckdb_result_sql_keyword;
-use crate::sql::{split_sql_batches, split_sql_statements};
+use crate::sql::{split_sql_batches, split_sql_statements, starts_with_executable_sql_keyword};
 
 pub const QUERY_TIMEOUT: Duration = Duration::from_secs(30);
 pub const MAX_ROWS: usize = 10000;
@@ -259,6 +261,10 @@ pub struct QueryExecutionOptions {
     /// `Some(0)` disables the timeout entirely.
     pub timeout_secs: Option<u64>,
     pub execution_id: Option<String>,
+    /// When `Some(true)`, multiple statements are executed within a single transaction
+    /// (BEGIN … COMMIT) instead of auto-commit mode. `None` and `Some(false)` behave
+    /// identically — auto-commit for each statement.
+    pub use_transaction: Option<bool>,
 }
 
 fn query_result_row_limit(max_rows: Option<usize>) -> usize {
@@ -1567,6 +1573,13 @@ pub async fn execute_multi_core_with_options(
         return Ok(vec![empty_query_result(0)]);
     }
 
+    // When use_transaction is explicitly true and we have multiple statements,
+    // route through the transaction wrapper instead of the sequential auto-commit loop.
+    if options.use_transaction == Some(true) && statements.len() > 1 {
+        let result = execute_statements_in_transaction(state, connection_id, database, &statements, schema).await?;
+        return Ok(vec![result]);
+    }
+
     let mysql_pool = {
         let connections = state.connections.read().await;
         match connections.get(&pool_key) {
@@ -2355,6 +2368,401 @@ async fn exec_tx_none_inner(
     })
 }
 
+/// Start a manual transaction session, holding a connection from the pool.
+/// Returns a transaction session ID that must be passed to subsequent calls.
+pub async fn begin_manual_transaction(
+    state: &AppState,
+    connection_id: &str,
+    database: &str,
+    schema: Option<&str>,
+) -> Result<String, String> {
+    let pool_key = if database.is_empty() {
+        connection_id.to_string()
+    } else {
+        state.get_or_create_pool(connection_id, Some(database)).await?
+    };
+
+    // Clone the pool handle under a brief read lock, then drop the lock before
+    // any async I/O — same pattern as do_execute throughout this file.
+    enum TxnPoolHandle {
+        Postgres(deadpool_postgres::Pool),
+        Mysql(db::mysql::MySqlPool),
+    }
+    let pool_handle = {
+        let connections = state.connections.read().await;
+        match connections.get(&pool_key).ok_or("Connection not found")? {
+            PoolKind::Postgres(pg) => TxnPoolHandle::Postgres(pg.clone()),
+            PoolKind::Mysql(mp, _) => TxnPoolHandle::Mysql(mp.clone()),
+            _ => return Err("Manual transaction is not supported for this database type".to_string()),
+        }
+    }; // connections lock released here
+
+    let txn_conn = match pool_handle {
+        TxnPoolHandle::Postgres(pg_pool) => {
+            let conn = pg_pool.get().await.map_err(|e| format!("Failed to get Postgres connection: {e}"))?;
+            conn.execute("BEGIN", &[]).await.map_err(|e| format!("BEGIN failed: {e}"))?;
+            if let Some(schema) = schema {
+                conn.execute(&format!("SET LOCAL search_path TO {}", db::postgres::pg_quote_ident(schema)), &[])
+                    .await
+                    .map_err(|e| format!("SET search_path failed: {e}"))?;
+            }
+            TxnConnection::Postgres(Box::new(conn))
+        }
+        TxnPoolHandle::Mysql(mysql_pool) => {
+            let mut conn = mysql_pool.get_conn().await.map_err(|e| format!("Failed to get MySQL connection: {e}"))?;
+            conn.query_drop("START TRANSACTION").await.map_err(|e| format!("START TRANSACTION failed: {e}"))?;
+            TxnConnection::Mysql(conn)
+        }
+    };
+
+    let txn_session_id = uuid::Uuid::new_v4().to_string();
+    let session = TransactionSession {
+        connection: Arc::new(tokio::sync::Mutex::new(txn_conn)),
+        pool_key: pool_key.clone(),
+        last_activity: std::time::Instant::now(),
+        busy: false,
+        connection_id: connection_id.to_string(),
+        database: database.to_string(),
+        schema: schema.map(|s| s.to_string()),
+    };
+
+    {
+        let mut sessions = state.transaction_sessions.write().await;
+        sessions.insert(txn_session_id.clone(), session);
+    }
+
+    // Schedule idle timeout watcher
+    spawn_txn_idle_watcher(state, txn_session_id.clone());
+
+    log::info!("[query][manual_txn:begin] session_id={}", txn_session_id);
+    Ok(txn_session_id)
+}
+
+/// Execute SQL within an existing manual transaction session.
+pub async fn execute_in_manual_transaction(
+    state: &AppState,
+    txn_session_id: &str,
+    sql: &str,
+    _database: &str,
+    _schema: Option<&str>,
+    max_rows: Option<usize>,
+) -> Result<Vec<db::QueryResult>, String> {
+    const TXN_IDLE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(300);
+
+    // Resolve statements and validate before taking the per-session connection
+    // lock. The session stays visible in the map so close/disconnect cleanup can
+    // remove it and roll back once the current DB operation releases the lock.
+    let (pool_key, connection_id) = {
+        let sessions = state.transaction_sessions.read().await;
+        let session = sessions
+            .get(txn_session_id)
+            .ok_or("Transaction session not found or expired; it may have been auto-rolled back due to inactivity")?;
+        (session.pool_key.clone(), session.connection_id.clone())
+    };
+
+    let db_type = connection_database_type(state, &connection_id).await;
+    let statements = db_type.map_or_else(
+        || split_sql_statements(sql),
+        |db_type| crate::sql::split_sql_statements_for_database(sql, db_type),
+    );
+    if statements.is_empty() {
+        return Ok(vec![empty_query_result(0)]);
+    }
+
+    // Read-only check while the session is still in the map. If this fails the
+    // session remains intact.
+    check_read_only_for_connection_multi(state, &pool_key, &statements).await?;
+
+    let connection = {
+        let mut sessions = state.transaction_sessions.write().await;
+        let Some(session) = sessions.get_mut(txn_session_id) else {
+            return Err(
+                "Transaction session not found or expired; it may have been auto-rolled back due to inactivity"
+                    .to_string(),
+            );
+        };
+        if session.busy {
+            return Err("Transaction session is already executing".to_string());
+        }
+        if session.last_activity.elapsed() > TXN_IDLE_TIMEOUT {
+            let session = sessions.remove(txn_session_id).expect("session exists");
+            Some(session.connection)
+        } else {
+            session.busy = true;
+            session.last_activity = std::time::Instant::now();
+            None
+        }
+    };
+    if let Some(connection) = connection {
+        let mut conn = connection.lock().await;
+        let _ = rollback_manual_txn_connection(&mut conn).await;
+        return Err("Transaction was auto-rolled back due to 5 minutes of inactivity".to_string());
+    }
+
+    let connection = {
+        let sessions = state.transaction_sessions.read().await;
+        sessions
+            .get(txn_session_id)
+            .map(|session| Arc::clone(&session.connection))
+            .ok_or("Transaction session not found or expired; it may have been auto-rolled back due to inactivity")?
+    };
+    let row_limit = max_rows.unwrap_or(MAX_ROWS).max(1);
+    let mut results = Vec::with_capacity(statements.len());
+
+    let mut conn = connection.lock().await;
+    for (i, statement) in statements.iter().enumerate() {
+        let result = match &mut *conn {
+            TxnConnection::Postgres(conn) => {
+                execute_manual_txn_postgres_statement(conn.as_ref(), statement, row_limit).await
+            }
+            TxnConnection::Mysql(conn) => execute_manual_txn_mysql_statement(conn, statement, row_limit).await,
+        };
+        match result {
+            Ok(query_result) => results.push(query_result),
+            Err(e) => {
+                // Statement failure ends the transaction. If another cleanup path
+                // already removed the session, it owns the final rollback.
+                let should_rollback = {
+                    let mut sessions = state.transaction_sessions.write().await;
+                    sessions.remove(txn_session_id).is_some()
+                };
+                if should_rollback {
+                    let _ = rollback_manual_txn_connection(&mut conn).await;
+                }
+                return Err(format!("Statement {} failed: {}. Transaction was auto-rolled back.", i + 1, e));
+            }
+        }
+    }
+    drop(conn);
+
+    let should_watch = {
+        let mut sessions = state.transaction_sessions.write().await;
+        if let Some(session) = sessions.get_mut(txn_session_id) {
+            session.busy = false;
+            session.last_activity = std::time::Instant::now();
+            true
+        } else {
+            false
+        }
+    };
+    if should_watch {
+        spawn_txn_idle_watcher(state, txn_session_id.to_string());
+    }
+
+    Ok(results)
+}
+
+async fn rollback_manual_txn_connection(conn: &mut TxnConnection) -> Result<(), String> {
+    match conn {
+        TxnConnection::Postgres(conn) => {
+            conn.execute("ROLLBACK", &[]).await.map_err(|e| format!("ROLLBACK failed: {e}"))?;
+        }
+        TxnConnection::Mysql(conn) => {
+            conn.query_drop("ROLLBACK").await.map_err(|e| format!("ROLLBACK failed: {e}"))?;
+        }
+    }
+    Ok(())
+}
+
+/// Spawn a background task that removes and rolls back a transaction session
+/// after 5 minutes of inactivity. The task does not hold the global lock across
+/// I/O: it briefly checks the map, and if the session exists and is expired,
+/// removes it, drops the lock, then rolls back the held connection.
+///
+/// Safety: if multiple watchers exist for the same session ID (e.g. due to
+/// a race), only the one that actually finds the session in the map and
+/// observes an elapsed time >= timeout will remove and roll back. Others
+/// will see a missing session or a non-expired one and exit harmlessly.
+fn spawn_txn_idle_watcher(state: &AppState, txn_session_id: String) {
+    let sessions = Arc::clone(&state.transaction_sessions);
+    tokio::spawn(async move {
+        const TXN_IDLE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(300);
+        tokio::time::sleep(TXN_IDLE_TIMEOUT).await;
+
+        let removed: Option<TransactionSession> = {
+            let mut guard = sessions.write().await;
+            match guard.get(&txn_session_id) {
+                Some(session) if !session.busy && session.last_activity.elapsed() >= TXN_IDLE_TIMEOUT => {
+                    guard.remove(&txn_session_id)
+                }
+                _ => None,
+            }
+        };
+
+        if let Some(session) = removed {
+            let mut conn = session.connection.lock().await;
+            let _ = rollback_manual_txn_connection(&mut conn).await;
+            log::info!(
+                "[query][manual_txn:idle_timeout] session_id={} auto-rolled back after 5 minutes of inactivity",
+                txn_session_id
+            );
+        }
+    });
+}
+
+async fn execute_manual_txn_postgres_statement(
+    conn: &deadpool_postgres::Object,
+    sql: &str,
+    row_limit: usize,
+) -> Result<db::QueryResult, String> {
+    if starts_with_executable_sql_keyword(sql, &["SELECT", "SHOW", "EXPLAIN", "WITH", "TABLE"]) {
+        let start = std::time::Instant::now();
+        let stmt = conn.prepare_cached(sql).await.map_err(|e| format!("Prepare failed: {e}"))?;
+        let columns: Vec<String> = stmt.columns().iter().map(|c| c.name().to_string()).collect();
+        let column_types: Vec<String> = stmt.columns().iter().map(|c| c.type_().name().to_string()).collect();
+        let params: Vec<&(dyn tokio_postgres::types::ToSql + Sync)> = Vec::new();
+        let stream = conn.query_raw(&stmt, params).await.map_err(|e| format!("Query failed: {e}"))?;
+        tokio::pin!(stream);
+        let mut data: Vec<Vec<serde_json::Value>> = Vec::with_capacity(row_limit.min(1024));
+        let mut truncated = false;
+        while let Some(row_result) = stream.next().await {
+            if data.len() >= row_limit {
+                truncated = true;
+                break;
+            }
+            let row = row_result.map_err(|e| format!("Query failed: {e}"))?;
+            let values: Vec<serde_json::Value> = (0..row.columns().len())
+                .map(|i| db::postgres::pg_value_to_json(&row, i, column_types.get(i).map(String::as_str).unwrap_or("")))
+                .collect();
+            data.push(values);
+        }
+        Ok(db::QueryResult {
+            columns,
+            column_types,
+            column_sortables: vec![],
+            rows: data,
+            affected_rows: 0,
+            execution_time_ms: start.elapsed().as_millis(),
+            truncated,
+            session_id: None,
+            has_more: false,
+        })
+    } else {
+        let affected = conn.execute(sql, &[]).await.map_err(|e| format!("Query failed: {e}"))?;
+        Ok(db::QueryResult {
+            columns: vec![],
+            column_types: Vec::new(),
+            column_sortables: vec![],
+            rows: vec![],
+            affected_rows: affected,
+            execution_time_ms: 0,
+            truncated: false,
+            session_id: None,
+            has_more: false,
+        })
+    }
+}
+
+async fn execute_manual_txn_mysql_statement(
+    conn: &mut mysql_async::Conn,
+    sql: &str,
+    row_limit: usize,
+) -> Result<db::QueryResult, String> {
+    if starts_with_executable_sql_keyword(sql, &["SELECT", "SHOW", "DESCRIBE", "EXPLAIN", "WITH"]) {
+        let start = std::time::Instant::now();
+        let mut result = conn.query_iter(sql).await.map_err(|e| format!("Query failed: {e}"))?;
+        let columns: Vec<String> = result.columns_ref().iter().map(|c| c.name_str().to_string()).collect();
+        let column_types: Vec<String> =
+            result.columns_ref().iter().map(|c| db::mysql::mysql_column_type_name(c.column_type())).collect();
+        let mut data: Vec<Vec<serde_json::Value>> = Vec::with_capacity(row_limit.min(1024));
+        let mut stream = result
+            .stream::<mysql_async::Row>()
+            .await
+            .map_err(|e| format!("Query failed: {e}"))?
+            .ok_or_else(|| "Empty result set stream".to_string())?;
+        let mut truncated = false;
+        while let Some(row) = stream.next().await {
+            if data.len() >= row_limit {
+                truncated = true;
+                break;
+            }
+            let row = row.map_err(|e| format!("Query failed: {e}"))?;
+            data.push((0..row.len()).map(|i| db::mysql::mysql_value_to_json(&row, i)).collect());
+        }
+        Ok(db::QueryResult {
+            columns,
+            column_types,
+            column_sortables: vec![],
+            rows: data,
+            affected_rows: 0,
+            execution_time_ms: start.elapsed().as_millis(),
+            truncated,
+            session_id: None,
+            has_more: false,
+        })
+    } else {
+        let result = conn.query_iter(sql).await.map_err(|e| format!("Query failed: {e}"))?;
+        let affected_rows = result.affected_rows();
+        result.drop_result().await.map_err(|e| format!("Query failed: {e}"))?;
+        Ok(db::QueryResult {
+            columns: vec![],
+            column_types: Vec::new(),
+            column_sortables: vec![],
+            rows: vec![],
+            affected_rows,
+            execution_time_ms: 0,
+            truncated: false,
+            session_id: None,
+            has_more: false,
+        })
+    }
+}
+
+/// Commit an existing manual transaction session.
+pub async fn commit_manual_transaction(state: &AppState, txn_session_id: &str) -> Result<db::QueryResult, String> {
+    let session = {
+        let mut sessions = state.transaction_sessions.write().await;
+        sessions.remove(txn_session_id).ok_or("Transaction session not found")?
+    };
+
+    let mut conn = session.connection.lock().await;
+    match &mut *conn {
+        TxnConnection::Postgres(conn) => {
+            conn.execute("COMMIT", &[]).await.map_err(|e| format!("COMMIT failed: {e}"))?;
+        }
+        TxnConnection::Mysql(conn) => {
+            conn.query_drop("COMMIT").await.map_err(|e| format!("COMMIT failed: {e}"))?;
+        }
+    }
+
+    log::info!("[query][manual_txn:commit] session_id={}", txn_session_id);
+    Ok(db::QueryResult {
+        columns: vec![],
+        column_types: Vec::new(),
+        column_sortables: vec![],
+        rows: vec![],
+        affected_rows: 0,
+        execution_time_ms: 0,
+        truncated: false,
+        session_id: None,
+        has_more: false,
+    })
+}
+
+/// Rollback an existing manual transaction session.
+pub async fn rollback_manual_transaction(state: &AppState, txn_session_id: &str) -> Result<db::QueryResult, String> {
+    let session = {
+        let mut sessions = state.transaction_sessions.write().await;
+        sessions.remove(txn_session_id).ok_or("Transaction session not found")?
+    };
+
+    let mut conn = session.connection.lock().await;
+    rollback_manual_txn_connection(&mut conn).await?;
+
+    log::info!("[query][manual_txn:rollback] session_id={}", txn_session_id);
+    Ok(db::QueryResult {
+        columns: vec![],
+        column_types: Vec::new(),
+        column_sortables: vec![],
+        rows: vec![],
+        affected_rows: 0,
+        execution_time_ms: 0,
+        truncated: false,
+        session_id: None,
+        has_more: false,
+    })
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -3091,4 +3499,22 @@ mod tests {
         assert_eq!(normalized.rows[0][0], serde_json::json!("2041797190226354178"));
         assert_eq!(normalized.rows[0][1], serde_json::json!([1, "2041797190226354178"]));
     }
+
+    #[test]
+    fn query_execution_options_default_use_transaction_is_none() {
+        let opts = QueryExecutionOptions::default();
+        assert_eq!(opts.use_transaction, None);
+    }
+
+    #[test]
+    fn query_execution_options_use_transaction_some_true_is_preserved() {
+        let opts = QueryExecutionOptions { use_transaction: Some(true), ..Default::default() };
+        assert_eq!(opts.use_transaction, Some(true));
+    }
+
+    #[test]
+    fn query_execution_options_use_transaction_some_false_is_preserved() {
+        let opts = QueryExecutionOptions { use_transaction: Some(false), ..Default::default() };
+        assert_eq!(opts.use_transaction, Some(false));
+    }
 }
diff --git a/crates/dbx-core/src/query_result_export.rs b/crates/dbx-core/src/query_result_export.rs
index d75ab0af88..7fddc598cb 100644
--- a/crates/dbx-core/src/query_result_export.rs
+++ b/crates/dbx-core/src/query_result_export.rs
@@ -399,6 +399,7 @@ async fn export_query_result_core_inner(
                 timeout_secs: request.timeout_secs,
                 client_session_id: request.client_session_id.clone(),
                 execution_id: request.execution_id.clone(),
+                ..Default::default()
             }
         } else {
             QueryExecutionOptions {
diff --git a/crates/dbx-core/src/schema.rs b/crates/dbx-core/src/schema.rs
index e154668b0e..7b87d0577b 100644
--- a/crates/dbx-core/src/schema.rs
+++ b/crates/dbx-core/src/schema.rs
@@ -876,15 +876,7 @@ pub async fn get_table_comment_core(
                                 &sql,
                                 Some(database),
                                 Some(schema),
-                                QueryExecutionOptions {
-                                    max_rows: Some(1),
-                                    fetch_size: None,
-                                    page_size: None,
-                                    result_session_id: None,
-                                    client_session_id: None,
-                                    timeout_secs: None,
-                                    execution_id: None,
-                                },
+                                QueryExecutionOptions { max_rows: Some(1), ..Default::default() },
                             ),
                             timeout,
                         )
diff --git a/crates/dbx-core/tests/live_manual_transaction.rs b/crates/dbx-core/tests/live_manual_transaction.rs
new file mode 100644
index 0000000000..60f9995599
--- /dev/null
+++ b/crates/dbx-core/tests/live_manual_transaction.rs
@@ -0,0 +1,105 @@
+use dbx_core::connection::AppState;
+use dbx_core::models::connection::{ConnectionConfig, DatabaseType};
+use dbx_core::query::{
+    begin_manual_transaction, commit_manual_transaction, execute_in_manual_transaction, rollback_manual_transaction,
+};
+use dbx_core::storage::Storage;
+
+fn live_config(prefix: &str, db_type: DatabaseType, default_port: u16) -> ConnectionConfig {
+    let host = std::env::var(format!("{prefix}_HOST")).expect("live DB host env var");
+    let port = std::env::var(format!("{prefix}_PORT"))
+        .ok()
+        .and_then(|value| value.parse::<u16>().ok())
+        .unwrap_or(default_port);
+    let username = std::env::var(format!("{prefix}_USER")).expect("live DB user env var");
+    let password = std::env::var(format!("{prefix}_PASSWORD")).expect("live DB password env var");
+    let database = std::env::var(format!("{prefix}_DATABASE")).expect("live DB database env var");
+
+    serde_json::from_value(serde_json::json!({
+        "id": format!("manual-txn-{prefix}"),
+        "name": format!("Manual transaction {prefix}"),
+        "db_type": db_type,
+        "host": host,
+        "port": port,
+        "username": username,
+        "password": password,
+        "database": database,
+        "connect_timeout_secs": 5,
+        "query_timeout_secs": 30,
+        "idle_timeout_secs": 60,
+        "keepalive_interval_secs": 0
+    }))
+    .expect("live connection config should deserialize")
+}
+
+async fn app_state_with_config(config: ConnectionConfig) -> (AppState, std::path::PathBuf) {
+    let db_path = std::env::temp_dir().join(format!("dbx-live-manual-txn-{}.db", uuid::Uuid::new_v4().simple()));
+    let storage = Storage::open(&db_path).await.expect("open temp storage");
+    let state = AppState::new(storage);
+    state.configs.write().await.insert(config.id.clone(), config);
+    (state, db_path)
+}
+
+#[tokio::test]
+#[ignore = "requires DBX_LIVE_MANUAL_TXN_POSTGRES_* env vars pointing at writable PostgreSQL"]
+async fn live_manual_transaction_postgres_preserves_typed_selects_and_empty_metadata() {
+    let config = live_config("DBX_LIVE_MANUAL_TXN_POSTGRES", DatabaseType::Postgres, 5432);
+    let database = config.database.clone().expect("database");
+    let (state, db_path) = app_state_with_config(config.clone()).await;
+
+    let txn = begin_manual_transaction(&state, &config.id, &database, None).await.expect("begin");
+    let typed = execute_in_manual_transaction(
+        &state,
+        &txn,
+        "SELECT 1::int4 AS id, 'pg'::text AS label, true AS ok",
+        &database,
+        None,
+        Some(10),
+    )
+    .await
+    .expect("typed select");
+    assert_eq!(typed[0].columns, vec!["id", "label", "ok"]);
+    assert_eq!(typed[0].rows, vec![vec![serde_json::json!(1), serde_json::json!("pg"), serde_json::json!(true)]]);
+
+    let empty = execute_in_manual_transaction(
+        &state,
+        &txn,
+        "SELECT 1::int4 AS id, 'empty'::text AS label WHERE false",
+        &database,
+        None,
+        Some(10),
+    )
+    .await
+    .expect("empty select");
+    assert_eq!(empty[0].columns, vec!["id", "label"]);
+    assert!(empty[0].rows.is_empty());
+
+    commit_manual_transaction(&state, &txn).await.expect("commit");
+    let _ = std::fs::remove_file(db_path);
+}
+
+#[tokio::test]
+#[ignore = "requires DBX_LIVE_MANUAL_TXN_MYSQL_* env vars pointing at writable MySQL"]
+async fn live_manual_transaction_mysql_streams_with_row_limit() {
+    let config = live_config("DBX_LIVE_MANUAL_TXN_MYSQL", DatabaseType::Mysql, 3306);
+    let database = config.database.clone().expect("database");
+    let (state, db_path) = app_state_with_config(config.clone()).await;
+
+    let txn = begin_manual_transaction(&state, &config.id, &database, None).await.expect("begin");
+    let limited = execute_in_manual_transaction(
+        &state,
+        &txn,
+        "SELECT 1 AS id UNION ALL SELECT 2 UNION ALL SELECT 3",
+        &database,
+        None,
+        Some(2),
+    )
+    .await
+    .expect("limited select");
+    assert_eq!(limited[0].columns, vec!["id"]);
+    assert_eq!(limited[0].rows.len(), 2);
+    assert!(limited[0].truncated);
+
+    rollback_manual_transaction(&state, &txn).await.expect("rollback");
+    let _ = std::fs::remove_file(db_path);
+}
diff --git a/crates/dbx-web/src/routes/query.rs b/crates/dbx-web/src/routes/query.rs
index 32793f1ce4..b4f67eda75 100644
--- a/crates/dbx-web/src/routes/query.rs
+++ b/crates/dbx-web/src/routes/query.rs
@@ -309,6 +309,7 @@ pub async fn execute_query(
             client_session_id: req.client_session_id,
             timeout_secs: req.timeout_secs,
             execution_id: Some(execution_id),
+            ..Default::default()
         },
     )
     .await
@@ -345,6 +346,7 @@ pub async fn execute_multi(
             client_session_id: req.client_session_id,
             timeout_secs: req.timeout_secs,
             execution_id: Some(execution_id),
+            ..Default::default()
         },
     )
     .await
diff --git a/src-tauri/src/commands/query.rs b/src-tauri/src/commands/query.rs
index c1d9a6e3d5..5edc5eb6e3 100644
--- a/src-tauri/src/commands/query.rs
+++ b/src-tauri/src/commands/query.rs
@@ -48,6 +48,7 @@ pub async fn execute_query(
             client_session_id,
             timeout_secs,
             execution_id,
+            ..Default::default()
         },
     )
     .await
@@ -68,6 +69,7 @@ pub async fn execute_multi(
     result_session_id: Option<String>,
     client_session_id: Option<String>,
     timeout_secs: Option<u64>,
+    use_transaction: Option<bool>,
 ) -> Result<Vec<db::QueryResult>, String> {
     let execution_id = execution_id.filter(|id| !id.trim().is_empty());
     let registered_query = execution_id.as_ref().map(|id| {
@@ -103,6 +105,7 @@ pub async fn execute_multi(
             client_session_id,
             timeout_secs,
             execution_id,
+            use_transaction,
         },
     )
     .await;
@@ -211,6 +214,52 @@ pub async fn execute_in_transaction(
     .await
 }
 
+#[tauri::command]
+pub async fn begin_manual_transaction(
+    state: State<'_, Arc<AppState>>,
+    connection_id: String,
+    database: String,
+    schema: Option<String>,
+) -> Result<String, String> {
+    dbx_core::query::begin_manual_transaction(&state, &connection_id, &database, schema.as_deref()).await
+}
+
+#[tauri::command]
+pub async fn execute_in_manual_transaction(
+    state: State<'_, Arc<AppState>>,
+    txn_session_id: String,
+    sql: String,
+    database: String,
+    schema: Option<String>,
+    max_rows: Option<usize>,
+) -> Result<Vec<db::QueryResult>, String> {
+    dbx_core::query::execute_in_manual_transaction(
+        &state,
+        &txn_session_id,
+        &sql,
+        &database,
+        schema.as_deref(),
+        max_rows,
+    )
+    .await
+}
+
+#[tauri::command]
+pub async fn commit_manual_transaction(
+    state: State<'_, Arc<AppState>>,
+    txn_session_id: String,
+) -> Result<db::QueryResult, String> {
+    dbx_core::query::commit_manual_transaction(&state, &txn_session_id).await
+}
+
+#[tauri::command]
+pub async fn rollback_manual_transaction(
+    state: State<'_, Arc<AppState>>,
+    txn_session_id: String,
+) -> Result<db::QueryResult, String> {
+    dbx_core::query::rollback_manual_transaction(&state, &txn_session_id).await
+}
+
 #[tauri::command]
 pub async fn analyze_sql_references(
     sql: String,
diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs
index 31b8d09610..c51e6ee8a1 100644
--- a/src-tauri/src/lib.rs
+++ b/src-tauri/src/lib.rs
@@ -717,6 +717,10 @@ pub fn run() {
             commands::query::execute_batch,
             commands::query::execute_script,
             commands::query::execute_in_transaction,
+            commands::query::begin_manual_transaction,
+            commands::query::execute_in_manual_transaction,
+            commands::query::commit_manual_transaction,
+            commands::query::rollback_manual_transaction,
             commands::query::analyze_sql_references,
             commands::query::find_statement_at_cursor,
             commands::query::prepare_query_pagination_execution_plan,