From b5d3343f9a6c85fd4fe845af54b1b142c66e6722 Mon Sep 17 00:00:00 2001
From: t8y2 <1156263951@qq.com>
Date: Fri, 21 Aug 2026 01:05:38 +0800
Subject: [PATCH] feat(oracle): support manual query transactions
---
.../java/com/dbx/agent/AgentProtocol.java | 6 +
.../src/main/resources/agent-protocol-v1.json | 3 +
.../src/main/resources/agent-protocol-v2.json | 3 +
agents/docs/agent-protocol-v2.md | 10 +
agents/drivers/oracle-go/main.go | 136 +++++++-
agents/drivers/oracle-go/main_test.go | 309 ++++++++++++++++++
.../src/components/layout/EditorToolbar.vue | 83 ++---
.../EditorToolbar.transactionMode.spec.ts | 34 ++
.../database/databaseFeatureSupport.spec.ts | 16 +-
.../src/lib/app/openTabsPersistence.ts | 2 +
.../lib/database/databaseFeatureSupport.ts | 10 +-
.../queryStore.hiddenPrimaryKey.spec.ts | 19 ++
.../queryStore.multiStatementError.spec.ts | 3 +
apps/desktop/src/stores/connectionStore.ts | 24 +-
apps/desktop/src/stores/queryStore.ts | 32 +-
crates/dbx-core/assets/agent-protocol-v1.json | 3 +
crates/dbx-core/assets/agent-protocol-v2.json | 3 +
crates/dbx-core/src/connection.rs | 199 +++++++++--
crates/dbx-core/src/db/agent_driver.rs | 45 ++-
crates/dbx-core/src/query.rs | 127 ++++++-
.../app-tests/openTabsPersistence.test.ts | 8 +
packages/app-tests/queryStore.test.ts | 35 ++
22 files changed, 1011 insertions(+), 99 deletions(-)
create mode 100644 apps/desktop/src/components/layout/__tests__/EditorToolbar.transactionMode.spec.ts
diff --git a/agents/common/src/main/java/com/dbx/agent/AgentProtocol.java b/agents/common/src/main/java/com/dbx/agent/AgentProtocol.java
index 9c7ac95a7e..ff48c016ff 100644
--- a/agents/common/src/main/java/com/dbx/agent/AgentProtocol.java
+++ b/agents/common/src/main/java/com/dbx/agent/AgentProtocol.java
@@ -42,6 +42,9 @@ public final class AgentProtocol {
public static final String METHOD_GET_EXPLAIN_INFO = "get_explain_info";
public static final String METHOD_EXECUTE_BATCH = "execute_batch";
public static final String METHOD_EXECUTE_TRANSACTION = "execute_transaction";
+ public static final String METHOD_BEGIN_MANUAL_TRANSACTION = "begin_manual_transaction";
+ public static final String METHOD_COMMIT_MANUAL_TRANSACTION = "commit_manual_transaction";
+ public static final String METHOD_ROLLBACK_MANUAL_TRANSACTION = "rollback_manual_transaction";
public static final String METHOD_DISCONNECT = "disconnect";
public static final String METHOD_SHUTDOWN = "shutdown";
@@ -200,6 +203,9 @@ public final class AgentProtocol {
METHOD_GET_EXPLAIN_INFO,
METHOD_EXECUTE_BATCH,
METHOD_EXECUTE_TRANSACTION,
+ METHOD_BEGIN_MANUAL_TRANSACTION,
+ METHOD_COMMIT_MANUAL_TRANSACTION,
+ METHOD_ROLLBACK_MANUAL_TRANSACTION,
METHOD_DISCONNECT,
METHOD_SHUTDOWN
));
diff --git a/agents/common/src/main/resources/agent-protocol-v1.json b/agents/common/src/main/resources/agent-protocol-v1.json
index dc9b4d087b..250bed89cf 100644
--- a/agents/common/src/main/resources/agent-protocol-v1.json
+++ b/agents/common/src/main/resources/agent-protocol-v1.json
@@ -36,6 +36,9 @@
"get_explain_info",
"execute_batch",
"execute_transaction",
+ "begin_manual_transaction",
+ "commit_manual_transaction",
+ "rollback_manual_transaction",
"disconnect",
"shutdown"
],
diff --git a/agents/common/src/main/resources/agent-protocol-v2.json b/agents/common/src/main/resources/agent-protocol-v2.json
index 6492173590..55b86252ee 100644
--- a/agents/common/src/main/resources/agent-protocol-v2.json
+++ b/agents/common/src/main/resources/agent-protocol-v2.json
@@ -40,6 +40,9 @@
"get_explain_info",
"execute_batch",
"execute_transaction",
+ "begin_manual_transaction",
+ "commit_manual_transaction",
+ "rollback_manual_transaction",
"disconnect",
"shutdown"
],
diff --git a/agents/docs/agent-protocol-v2.md b/agents/docs/agent-protocol-v2.md
index 3bd3eb2c3d..65029bfc66 100644
--- a/agents/docs/agent-protocol-v2.md
+++ b/agents/docs/agent-protocol-v2.md
@@ -11,6 +11,16 @@ Protocol v2 allows one Agent process to serve multiple isolated database session
- `close_session` closes the session resources, query cursors, and table-read cursors without affecting other sessions.
- `shutdown` closes all sessions and terminates the runtime.
+## Manual (interactive) transactions
+
+When a runtime advertises the `transaction` capability and supports sticky sessions, DBX may open a dedicated workload session and call:
+
+- `begin_manual_transaction` `{ schema? }` — pin one physical connection / start an open transaction
+- `execute_query` (and related query methods) — run on the open transaction until commit/rollback
+- `commit_manual_transaction` / `rollback_manual_transaction` — end the interactive transaction
+
+This is separate from one-shot `execute_transaction`, which begins, runs a statement list, and commits/rolls back inside a single RPC. Runtimes that reconnect a session (`validate_session`) must clear any open manual transaction on that session.
+
`agentSessionId` identifies a logical database connection. Existing `sessionId` fields remain pagination cursor identifiers and must not be used as logical connection identifiers.
`sessionRole` is `workload` by default. DBX sends `metadata` for object-tree, completion, and other read-only metadata sessions. New runtimes use this role to preserve metadata checkout capacity; older runtimes may ignore the field.
diff --git a/agents/drivers/oracle-go/main.go b/agents/drivers/oracle-go/main.go
index 99edd2f36b..c15a24ea9e 100644
--- a/agents/drivers/oracle-go/main.go
+++ b/agents/drivers/oracle-go/main.go
@@ -528,6 +528,10 @@ type server struct {
db *sql.DB
params connectParams
legacyLOBFetchDeferred bool
+ // manualConn + manualTx pin one physical Oracle session for interactive
+ // commit/rollback control across multiple execute_query RPCs.
+ manualConn *sql.Conn
+ manualTx *sql.Tx
sessions map[string]*querySession
tableReadSessions map[string]*querySession
nextSessionID int64
@@ -631,7 +635,7 @@ func (r *runtimeServer) dispatch(method string, params map[string]json.RawMessag
return map[string]any{
"protocolVersion": multiSessionProtocolVersion,
"agentProtocolVersion": multiSessionProtocolVersion,
- "capabilities": []string{"connect", "test_connection", "metadata", "query", "ddl", "multi_session"},
+ "capabilities": []string{"connect", "test_connection", "metadata", "query", "transaction", "ddl", "multi_session"},
}, false, nil
case "open_session":
agentSessionID := stringParam(params, "agentSessionId")
@@ -817,7 +821,7 @@ func (s *server) dispatch(method string, params map[string]json.RawMessage) (any
return map[string]any{
"protocolVersion": protocolVersion,
"agentProtocolVersion": protocolVersion,
- "capabilities": []string{"connect", "test_connection", "metadata", "query", "ddl"},
+ "capabilities": []string{"connect", "test_connection", "metadata", "query", "transaction", "ddl"},
}, false, nil
case "connect":
var cp connectParams
@@ -947,6 +951,12 @@ func (s *server) dispatch(method string, params map[string]json.RawMessage) (any
case "execute_transaction":
result, err := s.executeTransaction(params)
return result, false, err
+ case "begin_manual_transaction":
+ return map[string]bool{"ok": true}, false, s.beginManualTransaction(stringParam(params, "schema"))
+ case "commit_manual_transaction":
+ return map[string]bool{"ok": true}, false, s.commitManualTransaction()
+ case "rollback_manual_transaction":
+ return map[string]bool{"ok": true}, false, s.rollbackManualTransaction()
case "disconnect":
return map[string]bool{"ok": true}, false, s.disconnect()
case "shutdown":
@@ -1127,6 +1137,7 @@ func withOracleLOBFetchPost(params connectParams) connectParams {
func (s *server) disconnect() error {
s.closeAllQuerySessions()
+ _ = s.rollbackManualTransactionQuiet()
s.legacyLOBFetchDeferred = false
if s.db == nil {
return nil
@@ -1136,6 +1147,75 @@ func (s *server) disconnect() error {
return err
}
+func (s *server) beginManualTransaction(schema string) error {
+ if s.manualTx != nil {
+ return errors.New("manual transaction already open")
+ }
+ db, err := s.requireDB()
+ if err != nil {
+ return err
+ }
+ // Hold one exclusive physical connection so DML/SELECT/schema stay on the
+ // same Oracle session for the life of the interactive transaction.
+ conn, err := db.Conn(context.Background())
+ if err != nil {
+ return fmt.Errorf("reserve connection for manual transaction: %w", err)
+ }
+ if strings.TrimSpace(schema) != "" {
+ if _, err := conn.ExecContext(context.Background(), "ALTER SESSION SET CURRENT_SCHEMA = "+quoteIdentifier(schema)); err != nil {
+ _ = conn.Close()
+ return err
+ }
+ }
+ tx, err := conn.BeginTx(context.Background(), nil)
+ if err != nil {
+ _ = conn.Close()
+ return fmt.Errorf("begin manual transaction: %w", err)
+ }
+ s.manualConn = conn
+ s.manualTx = tx
+ return nil
+}
+
+func (s *server) commitManualTransaction() error {
+ if s.manualTx == nil {
+ return errors.New("no manual transaction open")
+ }
+ err := s.manualTx.Commit()
+ s.clearManualTransaction()
+ return err
+}
+
+func (s *server) rollbackManualTransaction() error {
+ if s.manualTx == nil {
+ return errors.New("no manual transaction open")
+ }
+ err := s.manualTx.Rollback()
+ s.clearManualTransaction()
+ return err
+}
+
+func (s *server) rollbackManualTransactionQuiet() error {
+ if s.manualTx == nil {
+ return nil
+ }
+ err := s.manualTx.Rollback()
+ s.clearManualTransaction()
+ return err
+}
+
+func (s *server) clearManualTransaction() {
+ s.manualTx = nil
+ if s.manualConn != nil {
+ _ = s.manualConn.Close()
+ s.manualConn = nil
+ }
+}
+
+func (s *server) hasManualTransaction() bool {
+ return s.manualTx != nil
+}
+
func openDB(params connectParams) (*sql.DB, error) {
return openDBWithStringConverter(params, nil)
}
@@ -3552,6 +3632,9 @@ func restoreOracleCurrentSchema(conn *sql.Conn, schema string) {
}
func (s *server) executeTransaction(params map[string]json.RawMessage) (queryResult, error) {
+ if s.hasManualTransaction() {
+ return queryResult{}, errors.New("cannot start a one-shot transaction while a manual transaction is open")
+ }
var payload struct {
Statements []string `json:"statements"`
Schema string `json:"schema"`
@@ -3601,7 +3684,7 @@ func (s *server) executeTransaction(params map[string]json.RawMessage) (queryRes
func (s *server) executeQueryPage(opts queryOptions, pageSize int) (queryPageResult, error) {
start := time.Now()
- if strings.TrimSpace(opts.Schema) != "" {
+ if strings.TrimSpace(opts.Schema) != "" && !s.hasManualTransaction() {
if err := s.setSchema(opts.Schema); err != nil {
return queryPageResult{}, err
}
@@ -3677,7 +3760,7 @@ func (s *server) storeQuerySession(session *querySession) string {
func (s *server) startTableRead(opts queryOptions, pageSize int) (queryPageResult, error) {
start := time.Now()
- if strings.TrimSpace(opts.Schema) != "" {
+ if strings.TrimSpace(opts.Schema) != "" && !s.hasManualTransaction() {
if err := s.setSchema(opts.Schema); err != nil {
return queryPageResult{}, err
}
@@ -3810,7 +3893,7 @@ func readQuerySessionPage(session *querySession, pageSize int) (queryPageResult,
func (s *server) executeQuery(opts queryOptions) (queryResult, error) {
start := time.Now()
- if strings.TrimSpace(opts.Schema) != "" {
+ if strings.TrimSpace(opts.Schema) != "" && !s.hasManualTransaction() {
if err := s.setSchema(opts.Schema); err != nil {
return queryResult{}, err
}
@@ -3825,8 +3908,7 @@ func (s *server) executeQuery(opts queryOptions) (queryResult, error) {
result.ExecutionTimeMS = time.Since(start).Milliseconds()
return result, err
}
- db, err := s.requireDB()
- if err != nil {
+ if _, err := s.requireDB(); err != nil {
return queryResult{}, err
}
ctx, cancel := context.WithCancel(context.Background())
@@ -3856,7 +3938,7 @@ func (s *server) executeQuery(opts queryOptions) (queryResult, error) {
}
s.activeCancelMu.Unlock()
}()
- execResult, err := db.ExecContext(ctx, sqlText)
+ execResult, err := s.execContext(ctx, sqlText)
if err != nil {
return queryResult{}, err
}
@@ -3864,6 +3946,17 @@ func (s *server) executeQuery(opts queryOptions) (queryResult, error) {
return queryResult{Columns: []string{}, ColumnTypes: []string{}, Rows: [][]any{}, AffectedRows: affected, ExecutionTimeMS: time.Since(start).Milliseconds()}, nil
}
+func (s *server) execContext(ctx context.Context, sqlText string) (sql.Result, error) {
+ if s.manualTx != nil {
+ return s.manualTx.ExecContext(ctx, sqlText)
+ }
+ db, err := s.requireDB()
+ if err != nil {
+ return nil, err
+ }
+ return db.ExecContext(ctx, sqlText)
+}
+
func (s *server) executeSelect(sqlText string, maxRows int, timeoutSecs int, deferLOBs bool) (queryResult, error) {
return executeOracleSelectWithXMLTypeRetry(
sqlText,
@@ -4725,11 +4818,20 @@ func skipBlockCommentSQL(value string, pos int) int {
}
func (s *server) setSchema(schema string) error {
+ sqlText := "ALTER SESSION SET CURRENT_SCHEMA = " + quoteIdentifier(schema)
+ if s.manualTx != nil {
+ _, err := s.manualTx.Exec(sqlText)
+ return err
+ }
+ if s.manualConn != nil {
+ _, err := s.manualConn.ExecContext(context.Background(), sqlText)
+ return err
+ }
db, err := s.requireDB()
if err != nil {
return err
}
- _, err = db.Exec("ALTER SESSION SET CURRENT_SCHEMA = " + quoteIdentifier(schema))
+ _, err = db.Exec(sqlText)
return err
}
@@ -4738,8 +4840,7 @@ func (s *server) queryRows(sqlText string, args []any) (*sql.Rows, error) {
}
func (s *server) queryRowsWithTimeout(sqlText string, args []any, timeoutSecs int) (*sql.Rows, error) {
- db, err := s.requireDB()
- if err != nil {
+ if _, err := s.requireDB(); err != nil {
return nil, err
}
ctx, cancel := context.WithCancel(context.Background())
@@ -4761,7 +4862,18 @@ func (s *server) queryRowsWithTimeout(sqlText string, args []any, timeoutSecs in
s.activeTimer = timer
s.activeTimedOut = false
s.activeCancelMu.Unlock()
- rows, queryErr := db.QueryContext(ctx, sqlText, args...)
+ var rows *sql.Rows
+ var queryErr error
+ if s.manualTx != nil {
+ rows, queryErr = s.manualTx.QueryContext(ctx, sqlText, args...)
+ } else {
+ db, err := s.requireDB()
+ if err != nil {
+ cancel()
+ return nil, err
+ }
+ rows, queryErr = db.QueryContext(ctx, sqlText, args...)
+ }
s.activeCancelMu.Lock()
s.activeCancel = nil
if s.activeTimer != nil {
diff --git a/agents/drivers/oracle-go/main_test.go b/agents/drivers/oracle-go/main_test.go
index 954568d405..21a8f7d457 100644
--- a/agents/drivers/oracle-go/main_test.go
+++ b/agents/drivers/oracle-go/main_test.go
@@ -12,6 +12,7 @@ import (
"os"
"reflect"
"strings"
+ "sync"
"testing"
"time"
@@ -2846,3 +2847,311 @@ func TestOracleCursorSurvivesDeadlineWindow(t *testing.T) {
t.Fatalf("expected 3 rows, got %d", rowCount)
}
}
+
+func TestManualTransactionBeginCommitRollback(t *testing.T) {
+ db, driver := openOracleManualTxTestDB(t)
+ s := newServer()
+ s.db = db
+
+ if err := s.beginManualTransaction(""); err != nil {
+ t.Fatalf("begin: %v", err)
+ }
+ if !s.hasManualTransaction() {
+ t.Fatal("expected open manual transaction")
+ }
+ if err := s.beginManualTransaction(""); err == nil {
+ t.Fatal("second begin should fail")
+ }
+
+ result, err := s.executeQuery(queryOptions{SQL: "UPDATE t SET a = 1", MaxRows: 10})
+ if err != nil {
+ t.Fatalf("update in txn: %v", err)
+ }
+ if result.AffectedRows != 1 {
+ t.Fatalf("expected 1 affected row, got %d", result.AffectedRows)
+ }
+
+ selectResult, err := s.executeQuery(queryOptions{SQL: "SELECT a FROM t", MaxRows: 10})
+ if err != nil {
+ t.Fatalf("select in txn: %v", err)
+ }
+ if len(selectResult.Rows) != 1 {
+ t.Fatalf("expected 1 row, got %d", len(selectResult.Rows))
+ }
+
+ if err := s.commitManualTransaction(); err != nil {
+ t.Fatalf("commit: %v", err)
+ }
+ if s.hasManualTransaction() {
+ t.Fatal("transaction should be closed after commit")
+ }
+ if !driver.committed {
+ t.Fatal("expected driver commit")
+ }
+ if driver.rolledBack {
+ t.Fatal("commit path should not rollback")
+ }
+
+ if err := s.beginManualTransaction(""); err != nil {
+ t.Fatalf("re-begin: %v", err)
+ }
+ if _, err := s.executeQuery(queryOptions{SQL: "DELETE FROM t", MaxRows: 10}); err != nil {
+ t.Fatalf("delete in txn: %v", err)
+ }
+ if err := s.rollbackManualTransaction(); err != nil {
+ t.Fatalf("rollback: %v", err)
+ }
+ if !driver.rolledBack {
+ t.Fatal("expected driver rollback")
+ }
+ if err := s.commitManualTransaction(); err == nil {
+ t.Fatal("commit without open txn should fail")
+ }
+ if err := s.rollbackManualTransaction(); err == nil {
+ t.Fatal("rollback without open txn should fail")
+ }
+}
+
+// TestManualTransactionSkipsPerStatementSetSchema guards against Oracle's
+// implicit COMMIT before+after DDL (ALTER SESSION SET CURRENT_SCHEMA is DDL).
+// Once a manual transaction is open and the schema is pinned at BEGIN time,
+// executeQuery/executeQueryPage/startTableRead must NOT re-issue ALTER SESSION
+// per statement, otherwise prior DML in the same transaction is silently
+// committed and Rollback can no longer undo it.
+func TestManualTransactionSkipsPerStatementSetSchema(t *testing.T) {
+ db, driver := openOracleManualTxTestDB(t)
+ s := newServer()
+ s.db = db
+
+ if err := s.beginManualTransaction("APP_TEST"); err != nil {
+ t.Fatalf("begin: %v", err)
+ }
+
+ // Run a couple of statements with a non-empty schema; before the fix each
+ // call would issue an extra ALTER SESSION SET CURRENT_SCHEMA inside the tx.
+ if _, err := s.executeQuery(queryOptions{SQL: "UPDATE t SET a = 1", Schema: "APP_TEST", MaxRows: 10}); err != nil {
+ t.Fatalf("update in txn: %v", err)
+ }
+ if _, err := s.executeQuery(queryOptions{SQL: "DELETE FROM t", Schema: "APP_TEST", MaxRows: 10}); err != nil {
+ t.Fatalf("delete in txn: %v", err)
+ }
+
+ alterSessionCount := 0
+ for _, q := range driver.execs {
+ if strings.Contains(strings.ToUpper(q), "ALTER SESSION SET CURRENT_SCHEMA") {
+ alterSessionCount++
+ }
+ }
+ // Exactly one ALTER SESSION is expected: the one issued by
+ // beginManualTransaction when pinning the schema. Any additional one would
+ // mean per-statement setSchema ran inside the open transaction and would
+ // trigger an implicit COMMIT of the preceding DML.
+ if alterSessionCount != 1 {
+ t.Fatalf("expected exactly 1 ALTER SESSION (from begin), got %d; per-statement setSchema must be skipped during manual tx", alterSessionCount)
+ }
+
+ if err := s.rollbackManualTransaction(); err != nil {
+ t.Fatalf("rollback: %v", err)
+ }
+}
+
+func TestManualTransactionDisconnectRollsBack(t *testing.T) {
+ db, driver := openOracleManualTxTestDB(t)
+ s := newServer()
+ s.db = db
+ if err := s.beginManualTransaction(""); err != nil {
+ t.Fatalf("begin: %v", err)
+ }
+ if err := s.disconnect(); err != nil {
+ t.Fatalf("disconnect: %v", err)
+ }
+ if s.hasManualTransaction() {
+ t.Fatal("disconnect should clear manual transaction")
+ }
+ if !driver.rolledBack {
+ t.Fatal("disconnect should rollback open manual transaction")
+ }
+}
+
+func TestManualTransactionBlocksOneShotTransaction(t *testing.T) {
+ db, _ := openOracleManualTxTestDB(t)
+ s := newServer()
+ s.db = db
+ if err := s.beginManualTransaction(""); err != nil {
+ t.Fatalf("begin: %v", err)
+ }
+ _, err := s.executeTransaction(map[string]json.RawMessage{
+ "statements": json.RawMessage(`["UPDATE t SET a = 1"]`),
+ })
+ if err == nil || !strings.Contains(err.Error(), "manual transaction") {
+ t.Fatalf("expected manual transaction conflict, got %v", err)
+ }
+}
+
+func TestRuntimeHandshakeAdvertisesTransactionCapability(t *testing.T) {
+ runtime := newRuntimeServer()
+ resp, shutdown := runtime.handleLine(`{"jsonrpc":"2.0","id":7,"method":"handshake","params":{"appVersion":"dev"}}`)
+ if shutdown || resp.Error != nil {
+ t.Fatalf("unexpected handshake response: shutdown=%v error=%v", shutdown, resp.Error)
+ }
+ data, err := json.Marshal(resp.Result)
+ if err != nil {
+ t.Fatal(err)
+ }
+ var result struct {
+ Capabilities []string `json:"capabilities"`
+ }
+ if err := json.Unmarshal(data, &result); err != nil {
+ t.Fatal(err)
+ }
+ if !contains(result.Capabilities, "transaction") {
+ t.Fatalf("expected transaction capability, got %v", result.Capabilities)
+ }
+}
+
+func TestManualTransactionRPCDispatch(t *testing.T) {
+ db, driver := openOracleManualTxTestDB(t)
+ s := newServer()
+ s.db = db
+
+ resp, _ := s.handleLine(`{"jsonrpc":"2.0","id":1,"method":"begin_manual_transaction","params":{}}`)
+ if resp.Error != nil {
+ t.Fatalf("begin rpc: %v", resp.Error)
+ }
+ resp, _ = s.handleLine(`{"jsonrpc":"2.0","id":2,"method":"execute_query","params":{"sql":"UPDATE t SET a = 2","maxRows":10}}`)
+ if resp.Error != nil {
+ t.Fatalf("execute rpc: %v", resp.Error)
+ }
+ resp, _ = s.handleLine(`{"jsonrpc":"2.0","id":3,"method":"commit_manual_transaction","params":{}}`)
+ if resp.Error != nil {
+ t.Fatalf("commit rpc: %v", resp.Error)
+ }
+ if !driver.committed {
+ t.Fatal("expected commit via RPC")
+ }
+}
+
+type oracleManualTxDriver struct {
+ mu sync.Mutex
+ committed bool
+ rolledBack bool
+ execs []string
+ queries []string
+}
+
+func openOracleManualTxTestDB(t *testing.T) (*sql.DB, *oracleManualTxDriver) {
+ t.Helper()
+ driverName := "oracle-test-manual-tx-" + strings.ReplaceAll(t.Name(), "/", "-") + "-" + time.Now().Format("150405.000000000")
+ drv := &oracleManualTxDriver{}
+ sql.Register(driverName, drv)
+ db, err := sql.Open(driverName, "")
+ if err != nil {
+ t.Fatal(err)
+ }
+ db.SetMaxOpenConns(1)
+ t.Cleanup(func() {
+ _ = db.Close()
+ })
+ return db, drv
+}
+
+func (d *oracleManualTxDriver) Open(string) (driver.Conn, error) {
+ return &oracleManualTxConn{driver: d}, nil
+}
+
+type oracleManualTxConn struct {
+ driver *oracleManualTxDriver
+ closed bool
+}
+
+func (c *oracleManualTxConn) Prepare(string) (driver.Stmt, error) {
+ return nil, errors.New("use Query/Exec context APIs")
+}
+
+func (c *oracleManualTxConn) Close() error {
+ c.closed = true
+ return nil
+}
+
+func (c *oracleManualTxConn) Begin() (driver.Tx, error) {
+ return c.BeginTx(context.Background(), driver.TxOptions{})
+}
+
+func (c *oracleManualTxConn) BeginTx(context.Context, driver.TxOptions) (driver.Tx, error) {
+ c.driver.mu.Lock()
+ defer c.driver.mu.Unlock()
+ c.driver.committed = false
+ c.driver.rolledBack = false
+ return &oracleManualTx{driver: c.driver}, nil
+}
+
+var (
+ _ driver.ConnBeginTx = (*oracleManualTxConn)(nil)
+ _ driver.ExecerContext = (*oracleManualTxConn)(nil)
+ _ driver.QueryerContext = (*oracleManualTxConn)(nil)
+)
+
+func (c *oracleManualTxConn) ExecContext(_ context.Context, query string, _ []driver.NamedValue) (driver.Result, error) {
+ c.driver.mu.Lock()
+ defer c.driver.mu.Unlock()
+ c.driver.execs = append(c.driver.execs, query)
+ return driver.RowsAffected(1), nil
+}
+
+func (c *oracleManualTxConn) QueryContext(_ context.Context, query string, _ []driver.NamedValue) (driver.Rows, error) {
+ c.driver.mu.Lock()
+ defer c.driver.mu.Unlock()
+ c.driver.queries = append(c.driver.queries, query)
+ return &oracleManualTxRows{
+ columns: []string{"A"},
+ values: [][]driver.Value{{int64(1)}},
+ }, nil
+}
+
+type oracleManualTx struct {
+ driver *oracleManualTxDriver
+ done bool
+}
+
+func (t *oracleManualTx) Commit() error {
+ t.driver.mu.Lock()
+ defer t.driver.mu.Unlock()
+ if t.done {
+ return errors.New("transaction already finished")
+ }
+ t.done = true
+ t.driver.committed = true
+ return nil
+}
+
+func (t *oracleManualTx) Rollback() error {
+ t.driver.mu.Lock()
+ defer t.driver.mu.Unlock()
+ if t.done {
+ return errors.New("transaction already finished")
+ }
+ t.done = true
+ t.driver.rolledBack = true
+ return nil
+}
+
+// database/sql routes Exec/Query on *sql.Tx through the underlying conn while the
+// transaction is open when the driver implements ExecerContext/QueryerContext.
+// Keep explicit Tx methods unused; Conn methods above serve sticky TX execution.
+
+type oracleManualTxRows struct {
+ columns []string
+ values [][]driver.Value
+ next int
+}
+
+func (r *oracleManualTxRows) Columns() []string { return r.columns }
+func (r *oracleManualTxRows) Close() error { return nil }
+func (r *oracleManualTxRows) Next(dest []driver.Value) error {
+ if r.next >= len(r.values) {
+ return io.EOF
+ }
+ copy(dest, r.values[r.next])
+ r.next++
+ return nil
+}
diff --git a/apps/desktop/src/components/layout/EditorToolbar.vue b/apps/desktop/src/components/layout/EditorToolbar.vue
index 542eff96d7..8995f44761 100644
--- a/apps/desktop/src/components/layout/EditorToolbar.vue
+++ b/apps/desktop/src/components/layout/EditorToolbar.vue
@@ -150,9 +150,12 @@ function toggleSqlSemanticDiagnostics() {
sqlSemanticDiagnosticsMode: sqlSemanticDiagnosticsEnabled.value ? "disabled" : "enabled",
});
}
+const isTransactionActive = computed(() => !!props.txnSessionId);
+const isManualTransactionMode = computed(() => props.autoCommit === false || isTransactionActive.value);
+const transactionModeBadge = computed(() => (isManualTransactionMode.value ? "M" : "A"));
const transactionTooltip = computed(() => {
const isAgent = (props.activeConnection?.db_type as string) === "agent";
- const isManual = props.autoCommit === false;
+ const isManual = isManualTransactionMode.value;
if (isAgent && isManual) return t("toolbar.manualTransactionAgent");
if (isAgent) return t("toolbar.autoCommitAgent");
return isManual ? t("toolbar.manualTransaction") : t("toolbar.autoCommit");
@@ -162,7 +165,6 @@ const executeButtonClass = computed(() => {
return activeProductionContext.value.active ? "bg-red-500/10 text-red-700 hover:bg-red-500/20 hover:text-red-800 dark:text-red-300 dark:hover:text-red-200" : "bg-emerald-500/10 text-emerald-700 hover:bg-emerald-500/20 hover:text-emerald-800 dark:text-emerald-300 dark:hover:text-emerald-200";
});
-const isTransactionActive = computed(() => !!props.txnSessionId);
const canMultiExecute = computed(() => {
if (!supportsQueryExecution(props.activeConnection?.db_type)) return false;
if (props.activeTab.isExecuting || props.activeTab.isExplaining || props.activeTab.isCancelling) return false;
@@ -306,41 +308,6 @@ async function changeCatalog(selectedCatalog: string) {
{{ explainAnalyzeTooltip }}
-
-
-
-
-
- {{ transactionTooltip }}
-
-
-
-
-
-
- {{ t("toolbar.commit") }}
-
-
-
-
-
-
-
- {{ t("toolbar.rollback") }}
-
{{ t("toolbar.multiDbExecute") }}
+
+
+
+
+
+
+ {{ transactionTooltip }}
+
+
+
+
+
+
+ {{ t("toolbar.commit") }}
+
+
+
+
+
+
+
+ {{ t("toolbar.rollback") }}
+
+
diff --git a/apps/desktop/src/components/layout/__tests__/EditorToolbar.transactionMode.spec.ts b/apps/desktop/src/components/layout/__tests__/EditorToolbar.transactionMode.spec.ts
new file mode 100644
index 0000000000..aaf3a8c44f
--- /dev/null
+++ b/apps/desktop/src/components/layout/__tests__/EditorToolbar.transactionMode.spec.ts
@@ -0,0 +1,34 @@
+import { readFileSync } from "node:fs";
+import { describe, expect, it } from "vitest";
+
+const toolbarSource = readFileSync(new URL("../EditorToolbar.vue", import.meta.url), "utf8");
+
+describe("EditorToolbar transaction mode control", () => {
+ it("shows an A or M badge next to Tx instead of relying on color alone", () => {
+ expect(toolbarSource).toContain('const transactionModeBadge = computed(() => (isManualTransactionMode.value ? "M" : "A"));');
+ expect(toolbarSource).toContain('Tx:');
+ expect(toolbarSource).toContain(">{{ transactionModeBadge }}");
+ });
+
+ it("exposes the transaction mode to assistive technology", () => {
+ expect(toolbarSource).toContain('class="ml-1 flex items-center gap-0.5 border-l border-border/60 pl-1" role="group" :aria-label="transactionTooltip"');
+ expect(toolbarSource).toContain(':aria-label="transactionTooltip"');
+ expect(toolbarSource).toContain(':aria-pressed="isManualTransactionMode"');
+ expect(toolbarSource).toContain(":aria-label=\"t('toolbar.commit')\"");
+ expect(toolbarSource).toContain(":aria-label=\"t('toolbar.rollback')\"");
+ });
+
+ it("places the transaction controls at the end in mode, commit, rollback order", () => {
+ const multiExecuteIndex = toolbarSource.indexOf("@click=\"emit('multiExecute')\"");
+ const transactionToggleIndex = toolbarSource.indexOf("@click=\"emit('update:autoCommit', autoCommit === false)\"");
+ const commitIndex = toolbarSource.indexOf("@click=\"emit('commit')\"");
+ const rollbackIndex = toolbarSource.indexOf("@click=\"emit('rollback')\"");
+ const actionGroupEndIndex = toolbarSource.indexOf('');
+
+ expect(multiExecuteIndex).toBeGreaterThan(-1);
+ expect(transactionToggleIndex).toBeGreaterThan(multiExecuteIndex);
+ expect(commitIndex).toBeGreaterThan(transactionToggleIndex);
+ expect(rollbackIndex).toBeGreaterThan(commitIndex);
+ expect(actionGroupEndIndex).toBeGreaterThan(rollbackIndex);
+ });
+});
diff --git a/apps/desktop/src/lib/__tests__/database/databaseFeatureSupport.spec.ts b/apps/desktop/src/lib/__tests__/database/databaseFeatureSupport.spec.ts
index 3c31cb363e..d15a56b5f1 100644
--- a/apps/desktop/src/lib/__tests__/database/databaseFeatureSupport.spec.ts
+++ b/apps/desktop/src/lib/__tests__/database/databaseFeatureSupport.spec.ts
@@ -3,6 +3,7 @@ import { connectionNamespaceCreationTarget, databaseNodeNamespaceCreationTarget
import { editableDatabasePropertyGroups, editableSchemaPropertyGroups } from "@/lib/database/databasePropertyEditing";
import { buildGetDatabaseCommentSql } from "@/lib/database/dbAdminSql";
import {
+ defaultAutoCommitForDbType,
isSchemaAware,
supportsConnectionScopedQueryExecution,
supportsConnectionDatabaseBrowser,
@@ -89,6 +90,7 @@ describe("supportsTransaction", () => {
it("returns true for supported database types", () => {
expect(supportsTransaction("postgres")).toBe(true);
expect(supportsTransaction("mysql")).toBe(true);
+ expect(supportsTransaction("oracle")).toBe(true);
});
it("returns false for unsupported database types", () => {
@@ -101,7 +103,6 @@ describe("supportsTransaction", () => {
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);
@@ -112,6 +113,19 @@ describe("supportsTransaction", () => {
});
});
+describe("defaultAutoCommitForDbType", () => {
+ it("defaults Oracle query tabs to manual transactions", () => {
+ expect(defaultAutoCommitForDbType("oracle")).toBe(false);
+ });
+
+ it("defaults other databases to auto-commit", () => {
+ expect(defaultAutoCommitForDbType("mysql")).toBe(true);
+ expect(defaultAutoCommitForDbType("postgres")).toBe(true);
+ expect(defaultAutoCommitForDbType("dameng")).toBe(true);
+ expect(defaultAutoCommitForDbType(undefined)).toBe(true);
+ });
+});
+
describe("supportsSqlInListPaste", () => {
it("allows generic and SQL-like editors", () => {
expect(supportsSqlInListPaste(undefined)).toBe(true);
diff --git a/apps/desktop/src/lib/app/openTabsPersistence.ts b/apps/desktop/src/lib/app/openTabsPersistence.ts
index 1602bff1dd..4a32024dd0 100644
--- a/apps/desktop/src/lib/app/openTabsPersistence.ts
+++ b/apps/desktop/src/lib/app/openTabsPersistence.ts
@@ -42,6 +42,7 @@ export interface SavedOpenTab {
whereInput?: string;
pinned?: boolean;
mode?: QueryTab["mode"];
+ autoCommit?: boolean;
mqTenant?: string;
mqInitialTab?: QueryTab["mqInitialTab"];
nacosNamespace?: string;
@@ -117,6 +118,7 @@ export function serializeOpenTabs(tabs: QueryTab[]): SavedOpenTab[] {
...(tab.whereInput !== undefined ? { whereInput: tab.whereInput } : {}),
pinned: tab.pinned,
mode: tab.mode,
+ ...(tab.mode === "query" && tab.autoCommit !== undefined ? { autoCommit: tab.autoCommit } : {}),
...(tab.mqTenant !== undefined ? { mqTenant: tab.mqTenant } : {}),
...(tab.mqInitialTab !== undefined ? { mqInitialTab: tab.mqInitialTab } : {}),
...(tab.nacosNamespace !== undefined ? { nacosNamespace: tab.nacosNamespace } : {}),
diff --git a/apps/desktop/src/lib/database/databaseFeatureSupport.ts b/apps/desktop/src/lib/database/databaseFeatureSupport.ts
index a3d22f3b9c..7fea054dca 100644
--- a/apps/desktop/src/lib/database/databaseFeatureSupport.ts
+++ b/apps/desktop/src/lib/database/databaseFeatureSupport.ts
@@ -229,7 +229,7 @@ export function usesPostgresLikeStructureCopy(dbType?: DatabaseType): boolean {
return !!dbType && PG_LIKE_STRUCTURE_TYPES.has(dbType);
}
-const TRANSACTION_SUPPORTED_TYPES: readonly string[] = ["postgres", "mysql"];
+const TRANSACTION_SUPPORTED_TYPES: readonly string[] = ["postgres", "mysql", "oracle"];
/**
* Returns true if the given database type supports explicit transaction control
@@ -238,3 +238,11 @@ const TRANSACTION_SUPPORTED_TYPES: readonly string[] = ["postgres", "mysql"];
export function supportsTransaction(dbType?: string): boolean {
return !!dbType && TRANSACTION_SUPPORTED_TYPES.includes(dbType);
}
+
+/**
+ * Default auto-commit mode when opening a query tab for the given database type.
+ * Oracle defaults to manual transactions so edits require an explicit Commit.
+ */
+export function defaultAutoCommitForDbType(dbType?: string): boolean {
+ return dbType !== "oracle";
+}
diff --git a/apps/desktop/src/stores/__tests__/queryStore.hiddenPrimaryKey.spec.ts b/apps/desktop/src/stores/__tests__/queryStore.hiddenPrimaryKey.spec.ts
index d91857d832..66653bf6b8 100644
--- a/apps/desktop/src/stores/__tests__/queryStore.hiddenPrimaryKey.spec.ts
+++ b/apps/desktop/src/stores/__tests__/queryStore.hiddenPrimaryKey.spec.ts
@@ -408,6 +408,8 @@ describe("queryStore hidden primary key editing", () => {
const { useQueryStore } = await import("@/stores/queryStore");
const store = useQueryStore();
const tabId = store.createTab("oracle-1", "ORCL", "Query");
+ // Exercise auto-commit execute-multi path (Oracle tabs default to manual TX).
+ store.setAutoCommit(tabId, true);
await store.executeTabSql(tabId, "SELECT t.* FROM SH_SMCVDMS_OVERSEAS_DRSSITB.TT_PLATFORM_CARS t WHERE t.PLATFORM = '轻卡'");
@@ -461,6 +463,7 @@ describe("queryStore hidden primary key editing", () => {
const { useQueryStore } = await import("@/stores/queryStore");
const store = useQueryStore();
const tabId = store.createTab("oracle-1", "ORCL", "Query");
+ store.setAutoCommit(tabId, true);
await store.executeTabSql(tabId, "SELECT * FROM CUSTOMERS");
@@ -494,6 +497,8 @@ describe("queryStore hidden primary key editing", () => {
const { useQueryStore } = await import("@/stores/queryStore");
const store = useQueryStore();
const tabId = store.createTab("oracle-1", "ORCL", "Query");
+ // Exercise auto-commit execute-multi path (Oracle tabs default to manual TX).
+ store.setAutoCommit(tabId, true);
await store.executeTabSql(tabId, "SELECT * FROM aa");
@@ -532,6 +537,7 @@ describe("queryStore hidden primary key editing", () => {
const { useQueryStore } = await import("@/stores/queryStore");
const store = useQueryStore();
const tabId = store.createTab("oracle-1", "ncdb", "Query");
+ store.setAutoCommit(tabId, true);
const sql = "SELECT * FROM VITF_PRODUCT_INFOOA WHERE PK_MATERIAL IN ('1', '2')";
const execution = store.executeTabSql(tabId, sql);
@@ -568,6 +574,8 @@ describe("queryStore hidden primary key editing", () => {
const { useQueryStore } = await import("@/stores/queryStore");
const store = useQueryStore();
const tabId = store.createTab("oracle-1", "ORCL", "Query");
+ // Exercise auto-commit execute-multi path (Oracle tabs default to manual TX).
+ store.setAutoCommit(tabId, true);
const execution = store.executeTabSql(tabId, "SELECT t.* FROM APP.WIDE_TABLE t");
await vi.waitFor(() => expect(executeMulti).toHaveBeenCalled());
@@ -618,6 +626,8 @@ describe("queryStore hidden primary key editing", () => {
const { useQueryStore } = await import("@/stores/queryStore");
const store = useQueryStore();
const tabId = store.createTab("oracle-1", "ORCL", "Query");
+ // Exercise auto-commit execute-multi path (Oracle tabs default to manual TX).
+ store.setAutoCommit(tabId, true);
await store.executeTabSql(tabId, sql);
@@ -665,6 +675,8 @@ describe("queryStore hidden primary key editing", () => {
const { useQueryStore } = await import("@/stores/queryStore");
const store = useQueryStore();
const tabId = store.createTab("oracle-1", "ORCL", "Query");
+ // Exercise auto-commit execute-multi path (Oracle tabs default to manual TX).
+ store.setAutoCommit(tabId, true);
const tab = store.tabs.find((item) => item.id === tabId)!;
await store.executeTabSql(tabId, "SELECT * FROM APP.PLATFORM_VIEW");
@@ -695,6 +707,7 @@ describe("queryStore hidden primary key editing", () => {
const { useQueryStore } = await import("@/stores/queryStore");
const store = useQueryStore();
const tabId = store.createTab("oracle-1", "ORCL", "Query");
+ store.setAutoCommit(tabId, true);
await store.executeTabSql(tabId, "SELECT PAYLOAD FROM APP.DOCUMENTS");
@@ -723,6 +736,7 @@ describe("queryStore hidden primary key editing", () => {
const { useQueryStore } = await import("@/stores/queryStore");
const store = useQueryStore();
const tabId = store.createTab("oracle-1", "ORCL", "Query");
+ store.setAutoCommit(tabId, true);
await store.executeTabSql(tabId, "SELECT * FROM APP.DOCUMENT_VIEW");
@@ -752,6 +766,7 @@ describe("queryStore hidden primary key editing", () => {
const { useQueryStore } = await import("@/stores/queryStore");
const store = useQueryStore();
const tabId = store.createTab("oracle-1", "ORCL", "Query");
+ store.setAutoCommit(tabId, true);
await store.executeTabSql(tabId, "SELECT * FROM DOCUMENTS");
@@ -780,6 +795,7 @@ describe("queryStore hidden primary key editing", () => {
const { useQueryStore } = await import("@/stores/queryStore");
const store = useQueryStore();
const tabId = store.createTab("oracle-1", "ORCL", "Query");
+ store.setAutoCommit(tabId, true);
await store.executeTabSql(tabId, "SELECT NAME FROM APP.CUSTOMERS");
@@ -817,6 +833,7 @@ describe("queryStore hidden primary key editing", () => {
const { useQueryStore } = await import("@/stores/queryStore");
const store = useQueryStore();
const tabId = store.createTab("oracle-1", "ORCL", "Query");
+ store.setAutoCommit(tabId, true);
await store.executeTabSql(tabId, "SELECT * FROM CUSTOMERS");
@@ -959,6 +976,8 @@ describe("queryStore hidden primary key editing", () => {
const { useQueryStore } = await import("@/stores/queryStore");
const store = useQueryStore();
const tabId = store.createTab("oracle-1", "XEPDB1", "Query");
+ // Exercise auto-commit execute-multi path (Oracle tabs default to manual TX).
+ store.setAutoCommit(tabId, true);
await store.executeTabSql(tabId, "SELECT NAME FROM DBX_HIDDEN_PK_EDIT_TEST");
diff --git a/apps/desktop/src/stores/__tests__/queryStore.multiStatementError.spec.ts b/apps/desktop/src/stores/__tests__/queryStore.multiStatementError.spec.ts
index 614d14e63a..b116cdf453 100644
--- a/apps/desktop/src/stores/__tests__/queryStore.multiStatementError.spec.ts
+++ b/apps/desktop/src/stores/__tests__/queryStore.multiStatementError.spec.ts
@@ -386,6 +386,9 @@ describe("queryStore multi-statement errors", () => {
const store = useQueryStore();
const tabA = store.createTab("oracle-1", "ORCL", "Tab A");
const tabB = store.createTab("oracle-1", "ORCL", "Tab B");
+ // Exercise auto-commit execute-multi path (Oracle tabs default to manual TX).
+ store.setAutoCommit(tabA, true);
+ store.setAutoCommit(tabB, true);
await store.executeTabSql(tabA, "ALTER SESSION SET CURRENT_SCHEMA = REPORTING");
expect(store.tabs.find((tab) => tab.id === tabA)?.completionContextVersion).toBe(1);
diff --git a/apps/desktop/src/stores/connectionStore.ts b/apps/desktop/src/stores/connectionStore.ts
index 5cf0e9572f..d049734826 100644
--- a/apps/desktop/src/stores/connectionStore.ts
+++ b/apps/desktop/src/stores/connectionStore.ts
@@ -371,7 +371,19 @@ export const useConnectionStore = defineStore("connection", () => {
let savedSqlFilesByDatabase = indexSavedSqlFilesByDatabase(savedSqlStore.allFiles);
const connections = ref([]);
const isDesktop = isTauriRuntime();
- const activeConnectionId = ref(localStorage.getItem(ACTIVE_CONNECTION_STORAGE_KEY));
+ // Prefer a safe read so Node/vitest (missing or partial storage mocks) do not
+ // throw when query tabs resolve connection db_type defaults via this store.
+ const activeConnectionId = ref(
+ (() => {
+ try {
+ const storage = globalThis.localStorage;
+ if (!storage || typeof storage.getItem !== "function") return null;
+ return storage.getItem(ACTIVE_CONNECTION_STORAGE_KEY);
+ } catch {
+ return null;
+ }
+ })(),
+ );
const selectedTreeNodeId = ref(null);
const selectedTreeNodeIds = ref([]);
// O(1) membership set — rebuilds only when selectedTreeNodeIds changes.
@@ -386,8 +398,14 @@ export const useConnectionStore = defineStore("connection", () => {
const treeClipboard = ref(null);
watch(activeConnectionId, (id) => {
- if (id) localStorage.setItem(ACTIVE_CONNECTION_STORAGE_KEY, id);
- else localStorage.removeItem(ACTIVE_CONNECTION_STORAGE_KEY);
+ try {
+ const storage = globalThis.localStorage;
+ if (!storage || typeof storage.setItem !== "function" || typeof storage.removeItem !== "function") return;
+ if (id) storage.setItem(ACTIVE_CONNECTION_STORAGE_KEY, id);
+ else storage.removeItem(ACTIVE_CONNECTION_STORAGE_KEY);
+ } catch {
+ // Ignore storage failures in non-browser / partial mock environments.
+ }
});
const treeNodes = ref([]);
const sidebarDatabaseStorageCache = new Map();
diff --git a/apps/desktop/src/stores/queryStore.ts b/apps/desktop/src/stores/queryStore.ts
index 23ade64681..466c9b5bc3 100644
--- a/apps/desktop/src/stores/queryStore.ts
+++ b/apps/desktop/src/stores/queryStore.ts
@@ -35,7 +35,7 @@ import { redisCommandResultToQueryResult } from "@/lib/redis/redisQueryResult";
import { nextRedisCommandDb } from "@/lib/redis/redisCommandSession";
import { isRedisMutatingCommand } from "@/lib/redis/redisCommandTable";
import { usesAgentCursorForQuery } from "@/lib/database/databaseDriverManifest";
-import { supportsClearableQuerySchema } from "@/lib/database/databaseFeatureSupport";
+import { defaultAutoCommitForDbType, supportsClearableQuerySchema } from "@/lib/database/databaseFeatureSupport";
import { canInsertTableRows, canUseKeylessRowPredicate, DBX_ROWID_COLUMN, editablePrimaryKeys, usesSyntheticRowIdKey } from "@/lib/table/tableEditing";
import { TABLE_DATA_EXPORT_PAGE_SIZE } from "@/lib/table/tableDataExport";
import { tableMetaForDataTab } from "@/lib/table/tableDataTabMeta";
@@ -1509,9 +1509,12 @@ export const useQueryStore = defineStore("query", () => {
function applyRestoredOpenTabs(restored: { tabs: QueryTab[]; activeTabId: string | null }) {
const connectionStore = useConnectionStore();
for (const tab of restored.tabs) {
- if (tab.mode !== "data") continue;
const connection = connectionStore.getConfig(tab.connectionId);
- if (connection) tab.schema = connectionObjectTreeNodeSchema(connection, tab.database, tab.schema);
+ if (tab.mode === "query" && tab.autoCommit === undefined) {
+ tab.autoCommit = defaultAutoCommitForDbType(connection?.db_type);
+ } else if (tab.mode === "data" && connection) {
+ tab.schema = connectionObjectTreeNodeSchema(connection, tab.database, tab.schema);
+ }
}
tabs.value = restored.tabs;
activeTabId.value = restored.activeTabId;
@@ -1657,6 +1660,7 @@ export const useQueryStore = defineStore("query", () => {
}
const id = uuid();
+ const dbType = useConnectionStore().getConfig(connectionId)?.db_type;
const tab: QueryTab = {
id,
title: title || `query_${tabs.value.length + 1}`,
@@ -1671,6 +1675,7 @@ export const useQueryStore = defineStore("query", () => {
isCancelling: false,
isExplaining: false,
mode,
+ ...(mode === "query" ? { autoCommit: defaultAutoCommitForDbType(dbType) } : {}),
};
if (mode === "query") tab.originalSql = initialSql ?? "";
const activeIndex = options.insertAfterActive ? tabs.value.findIndex((item) => item.id === activeTabId.value) : -1;
@@ -1748,6 +1753,7 @@ export const useQueryStore = defineStore("query", () => {
// File-backed tabs are identified by their full path, not their basename.
// Bypassing createTab avoids overwriting another file with the same name.
const id = uuid();
+ const dbType = useConnectionStore().getConfig(connectionId)?.db_type;
const tab: QueryTab = {
id,
title: "",
@@ -1763,6 +1769,7 @@ export const useQueryStore = defineStore("query", () => {
isCancelling: false,
isExplaining: false,
mode: "query",
+ autoCommit: defaultAutoCommitForDbType(dbType),
};
tabs.value.push(tab);
activeTabId.value = id;
@@ -3022,7 +3029,10 @@ export const useQueryStore = defineStore("query", () => {
function rollbackTabTransaction(tab: QueryTab, options?: { resetAutoCommit?: boolean }) {
if (tab.txnSessionId) void rollbackTransaction(tab.id);
- if (options?.resetAutoCommit) tab.autoCommit = true;
+ if (options?.resetAutoCommit) {
+ const dbType = useConnectionStore().getConfig(tab.connectionId)?.db_type;
+ tab.autoCommit = defaultAutoCommitForDbType(dbType);
+ }
tab.txnAutoRolledBack = false;
}
@@ -3161,6 +3171,7 @@ export const useQueryStore = defineStore("query", () => {
const id = uuid();
const restoredPosition = restoreSavedSqlEditorPosition(file.id, file.sql);
+ const dbType = useConnectionStore().getConfig(target.connectionId)?.db_type;
const tab: QueryTab = {
id,
title: file.name,
@@ -3176,6 +3187,7 @@ export const useQueryStore = defineStore("query", () => {
isCancelling: false,
isExplaining: false,
mode: "query",
+ autoCommit: defaultAutoCommitForDbType(dbType),
editorSelection: restoredPosition.selection,
editorViewport: restoredPosition.viewport,
};
@@ -5175,13 +5187,17 @@ export const useQueryStore = defineStore("query", () => {
queryExecutionLog("error", "error", { traceId, elapsed: elapsed(), error: e });
// Sync connection state if the error indicates a lost connection
useConnectionStore().recordConnectionLostError(executionConnectionId ?? tab.connectionId, e);
- // Handle manual transaction auto-rollback (e.g. deadlock detected by server,
- // statement error inside a manual transaction, or idle timeout).
+ // Handle manual transaction auto-rollback (idle timeout only for the banner;
+ // other statement failures still clear the session without the 5-minute notice).
if (tab.autoCommit === false) {
const errMsg: string = e?.message ?? String(e);
- if (/rolled.?back/i.test(errMsg) || errMsg.includes("已自动回滚")) {
+ const idleTimeout = /5 minutes of inactivity/i.test(errMsg) || errMsg.includes("5 分钟无操作") || errMsg.includes("已自动回滚");
+ if (idleTimeout) {
tab.txnSessionId = undefined;
tab.txnAutoRolledBack = true;
+ } else if (/rolled.?back/i.test(errMsg) || /transaction session not found/i.test(errMsg) || /agent runtime terminated/i.test(errMsg)) {
+ tab.txnSessionId = undefined;
+ tab.txnAutoRolledBack = false;
}
}
const current = findExecutionTab(id);
@@ -5881,6 +5897,7 @@ export const useQueryStore = defineStore("query", () => {
function openResultArchiveTab(archive: DecodedQueryResultArchive): string | undefined {
const id = uuid();
const title = archive.tab.title.trim() || t("tabs.importedResultArchive");
+ const dbType = useConnectionStore().getConfig(archive.tab.connectionId)?.db_type;
const tab: QueryTab = {
id,
title,
@@ -5897,6 +5914,7 @@ export const useQueryStore = defineStore("query", () => {
isCancelling: false,
isExplaining: false,
mode: "query",
+ autoCommit: defaultAutoCommitForDbType(dbType),
};
if (!restoreCachedResultPayload(tab, archive.snapshot)) return undefined;
const activeRun = tab.resultRuns?.find((run) => run.id === tab.activeResultRunId) ?? tab.resultRuns?.[0];
diff --git a/crates/dbx-core/assets/agent-protocol-v1.json b/crates/dbx-core/assets/agent-protocol-v1.json
index dc9b4d087b..250bed89cf 100644
--- a/crates/dbx-core/assets/agent-protocol-v1.json
+++ b/crates/dbx-core/assets/agent-protocol-v1.json
@@ -36,6 +36,9 @@
"get_explain_info",
"execute_batch",
"execute_transaction",
+ "begin_manual_transaction",
+ "commit_manual_transaction",
+ "rollback_manual_transaction",
"disconnect",
"shutdown"
],
diff --git a/crates/dbx-core/assets/agent-protocol-v2.json b/crates/dbx-core/assets/agent-protocol-v2.json
index 6492173590..55b86252ee 100644
--- a/crates/dbx-core/assets/agent-protocol-v2.json
+++ b/crates/dbx-core/assets/agent-protocol-v2.json
@@ -40,6 +40,9 @@
"get_explain_info",
"execute_batch",
"execute_transaction",
+ "begin_manual_transaction",
+ "commit_manual_transaction",
+ "rollback_manual_transaction",
"disconnect",
"shutdown"
],
diff --git a/crates/dbx-core/src/connection.rs b/crates/dbx-core/src/connection.rs
index e2492d999b..29e988e259 100644
--- a/crates/dbx-core/src/connection.rs
+++ b/crates/dbx-core/src/connection.rs
@@ -178,6 +178,14 @@ enum ConnectionDatabaseInfoSource {
pub enum TxnConnection {
Postgres(Box),
Mysql(mysql_async::Conn),
+ /// Dedicated agent multi_session workload client with an open sticky TX.
+ Agent {
+ client: Arc,
+ /// Client-session id used when opening the dedicated agent pool.
+ client_session_id: String,
+ database: Option,
+ cleanup_guard: ClientSessionPoolCleanupGuard,
+ },
}
pub struct TransactionSession {
@@ -376,7 +384,7 @@ struct PoolRoutingControl {
task_supervisor: TaskSupervisor,
}
-pub(crate) struct ClientSessionPoolCleanupGuard {
+pub struct ClientSessionPoolCleanupGuard {
pool_key: String,
routing: PoolRoutingControl,
armed: bool,
@@ -435,12 +443,20 @@ impl PoolRoutingControl {
let mut removed = vec![(pool_key.to_string(), pool)];
if replace_agent_runtime {
let sibling_keys = shared_runtime_sibling_keys(&connections, &removed[0].1);
- for key in sibling_keys {
- if let Some(pool) = connections.remove(&key) {
- removed.push((key, pool));
+ let protects_manual_txn = sibling_keys.iter().any(|key| is_manual_transaction_pool_key(key))
+ || is_manual_transaction_pool_key(pool_key);
+ if protects_manual_txn {
+ log::warn!(
+ "Skipping shared Agent runtime kill for '{pool_key}' because a manual-transaction session is active on the same runtime"
+ );
+ } else {
+ for key in sibling_keys {
+ if let Some(pool) = connections.remove(&key) {
+ removed.push((key, pool));
+ }
}
+ fail_stop_removed_agent_pool(pool_key, &removed[0].1);
}
- fail_stop_removed_agent_pool(pool_key, &removed[0].1);
}
removed
};
@@ -472,14 +488,30 @@ impl PoolRoutingControl {
_ => None,
})
.collect::>();
- let removed = runtime_keys
- .into_iter()
- .filter_map(|key| connections.remove(&key).map(|pool| (key, pool)))
- .collect::>();
- if !expected_client.fail_stop() {
- log::warn!("Failed to terminate the shared Agent runtime while detaching pool '{pool_key}'");
+ // Sticky manual-transaction sessions pin work on the shared runtime.
+ // Prefer quarantining only the failed route over killing every sibling TX.
+ let protects_manual_txn = runtime_keys.iter().any(|key| is_manual_transaction_pool_key(key))
+ || is_manual_transaction_pool_key(pool_key);
+ if protects_manual_txn {
+ log::warn!(
+ "Agent pool '{pool_key}' requested runtime replacement, but a manual-transaction session shares the runtime; detaching only this pool"
+ );
+ if is_current {
+ let pool = connections.remove(pool_key).expect("current Agent pool must still exist");
+ vec![(pool_key.to_string(), pool)]
+ } else {
+ Vec::new()
+ }
+ } else {
+ let removed = runtime_keys
+ .into_iter()
+ .filter_map(|key| connections.remove(&key).map(|pool| (key, pool)))
+ .collect::>();
+ if !expected_client.fail_stop() {
+ log::warn!("Failed to terminate the shared Agent runtime while detaching pool '{pool_key}'");
+ }
+ removed
}
- removed
} else {
let pool = connections.remove(pool_key).expect("current Agent pool must still exist");
vec![(pool_key.to_string(), pool)]
@@ -511,20 +543,40 @@ impl PoolRoutingControl {
PoolKind::Agent(client) => Some(client.clone()),
_ => None,
};
+ let protects_manual_txn = match agent_client.as_ref() {
+ Some(client) if client.uses_shared_runtime() => {
+ let connections = self.connections.read().await;
+ connections.iter().any(|(key, pool)| {
+ is_manual_transaction_pool_key(key)
+ && matches!(pool, PoolKind::Agent(sibling) if client.shares_runtime_with(sibling))
+ })
+ }
+ _ => false,
+ };
match tokio::time::timeout(Duration::from_secs(POOL_CLOSE_TIMEOUT_SECS), close_pool_kind(pool)).await {
Ok(Ok(())) => {}
Ok(Err(error)) => {
log::warn!("Failed to close connection pool '{pool_key}': {error}");
- if let Some(client) = agent_client.filter(|_| should_replace_agent_runtime(&error)) {
+ if protects_manual_txn {
+ log::warn!(
+ "Leaving shared Agent runtime running after close failure for '{pool_key}' to protect sibling sessions"
+ );
+ } else if let Some(client) = agent_client.filter(|_| should_replace_agent_runtime(&error)) {
self.replace_runtime_after_close_failure(&pool_key, &client).await;
}
}
Err(_) => {
- log::warn!(
- "Timed out closing connection pool '{pool_key}' after {POOL_CLOSE_TIMEOUT_SECS}s; replacing a shared Agent runtime when present."
- );
- if let Some(client) = agent_client {
- self.replace_runtime_after_close_failure(&pool_key, &client).await;
+ if protects_manual_txn {
+ log::warn!(
+ "Timed out closing shared Agent pool '{pool_key}' after {POOL_CLOSE_TIMEOUT_SECS}s; leaving the runtime running so sibling sessions (e.g. manual transactions) stay alive"
+ );
+ } else {
+ log::warn!(
+ "Timed out closing connection pool '{pool_key}' after {POOL_CLOSE_TIMEOUT_SECS}s; replacing a shared Agent runtime when present."
+ );
+ if let Some(client) = agent_client {
+ self.replace_runtime_after_close_failure(&pool_key, &client).await;
+ }
}
}
}
@@ -1500,13 +1552,19 @@ impl AppState {
// candidate never reaches this point, so the existing route keeps its state.
routing.stop_keepalive(&pool_key);
activity.insert(pool_key.clone(), PoolActivity::now());
- self.start_keepalive_task(
- &pool_key,
- &pool,
- config,
- #[cfg(feature = "mq-admin")]
- mq_keepalive_adapter.clone(),
- );
+ // Manual-transaction sessions pin a sticky connection/TX. Keepalive
+ // detach on these pools was able to tear down the shared agent runtime
+ // (and any open TX) when close timed out — skip probes for them.
+ let skip_keepalive = pool_key.contains(":session:manual-txn-");
+ if !skip_keepalive {
+ self.start_keepalive_task(
+ &pool_key,
+ &pool,
+ config,
+ #[cfg(feature = "mq-admin")]
+ mq_keepalive_adapter.clone(),
+ );
+ }
break Ok(connections.insert(pool_key.clone(), pool));
};
let previous = match previous {
@@ -3626,6 +3684,21 @@ impl AppState {
.await
}
+ pub(crate) async fn workload_session_pool_cleanup_guard(
+ &self,
+ connection_id: &str,
+ database: Option<&str>,
+ client_session_id: &str,
+ ) -> Option {
+ self.client_session_pool_cleanup_guard_for_role(
+ connection_id,
+ database,
+ client_session_id,
+ AgentSessionRole::Workload,
+ )
+ .await
+ }
+
async fn client_session_pool_cleanup_guard_for_role(
&self,
connection_id: &str,
@@ -4986,6 +5059,10 @@ fn is_session_scoped_pool_key(pool_key: &str) -> bool {
pool_key.contains(":session:")
}
+fn is_manual_transaction_pool_key(pool_key: &str) -> bool {
+ pool_key.contains(":session:manual-txn-")
+}
+
pub(crate) fn config_for_pool_key<'a>(
pool_key: &str,
configs: &'a HashMap,
@@ -5493,7 +5570,7 @@ mod tests {
redis_sentinel_transport_id, redis_sentinel_transport_prefix, sqlserver_legacy_agent_config,
sqlserver_legacy_driver_error, sqlserver_uses_legacy_driver, task_client_session_id,
upsert_connection_url_param, uses_bare_mysql_pool, uses_tcp_probe, validate_connection_url_params,
- validate_h2_database_path, AppState, MysqlMode, PoolKind, GAUSSDB_M_JDBC_DRIVER_CLASS,
+ validate_h2_database_path, AppState, MysqlMode, PoolKind, TxnConnection, GAUSSDB_M_JDBC_DRIVER_CLASS,
GAUSSDB_M_JDBC_DRIVER_PROFILE, PRESTOSQL_JDBC_DRIVER_CLASS,
};
use crate::agent_connection::{
@@ -7662,6 +7739,39 @@ for line in sys.stdin:
let _ = std::fs::remove_dir_all(dir);
}
+ #[tokio::test]
+ async fn metadata_close_replace_runtime_preserves_manual_transaction_sibling() {
+ let (state, dir) = test_app_state().await;
+ let (runtime, metadata_client, manual_txn_client) =
+ replace_runtime_on_error_clients(&dir, "close_session").await;
+ let mut config = mysql_config(Some("analytics"));
+ config.id = "conn".to_string();
+ config.db_type = DatabaseType::Dameng;
+ state.configs.write().await.insert(config.id.clone(), config);
+ let metadata_pool_key = "conn:analytics:session:metadata-session:role:metadata";
+ let manual_txn_pool_key = "conn:analytics:session:manual-txn-test";
+ {
+ let mut connections = state.connections.write().await;
+ connections.insert(metadata_pool_key.to_string(), PoolKind::Agent(metadata_client));
+ connections.insert(manual_txn_pool_key.to_string(), PoolKind::Agent(manual_txn_client));
+ }
+ {
+ let mut activity = state.pool_activity.write().await;
+ activity.insert(metadata_pool_key.to_string(), super::PoolActivity::now());
+ activity.insert(manual_txn_pool_key.to_string(), super::PoolActivity::now());
+ }
+
+ assert!(state.close_metadata_session_pool("conn", Some("analytics"), "metadata-session").await.unwrap());
+
+ assert!(!state.connections.read().await.contains_key(metadata_pool_key));
+ assert!(state.connections.read().await.contains_key(manual_txn_pool_key));
+ assert!(state.pool_activity.read().await.contains_key(manual_txn_pool_key));
+ assert!(!runtime.is_failed());
+
+ state.shutdown(Duration::from_secs(1)).await;
+ let _ = std::fs::remove_dir_all(dir);
+ }
+
#[tokio::test]
async fn reclaim_close_replace_runtime_never_restores_failed_pool() {
let (state, dir) = test_app_state().await;
@@ -8151,6 +8261,43 @@ for line in sys.stdin:
let _ = std::fs::remove_dir_all(dir);
}
+ #[tokio::test]
+ async fn manual_transaction_connection_drop_detaches_session_pool() {
+ let (state, dir) = test_app_state().await;
+ let (runtime, manual_txn_client, _sibling_client) = replace_runtime_on_error_clients(&dir, "unused").await;
+ let mut config = mysql_config(Some("analytics"));
+ config.id = "conn".to_string();
+ config.db_type = DatabaseType::Dameng;
+ state.configs.write().await.insert(config.id.clone(), config);
+
+ let client_session_id = "manual-txn-test";
+ let pool_key = "conn:analytics:session:manual-txn-test";
+ state.connections.write().await.insert(pool_key.to_string(), PoolKind::Agent(manual_txn_client.clone()));
+ state.pool_activity.write().await.insert(pool_key.to_string(), super::PoolActivity::now());
+ let cleanup_guard =
+ state.workload_session_pool_cleanup_guard("conn", Some("analytics"), client_session_id).await.unwrap();
+
+ drop(TxnConnection::Agent {
+ client: manual_txn_client,
+ client_session_id: client_session_id.to_string(),
+ database: Some("analytics".to_string()),
+ cleanup_guard,
+ });
+
+ for _ in 0..100 {
+ if !state.connections.read().await.contains_key(pool_key) {
+ break;
+ }
+ tokio::time::sleep(Duration::from_millis(10)).await;
+ }
+ assert!(!state.connections.read().await.contains_key(pool_key));
+ assert!(!state.pool_activity.read().await.contains_key(pool_key));
+ assert!(!runtime.is_failed());
+
+ state.shutdown(Duration::from_secs(1)).await;
+ let _ = std::fs::remove_dir_all(dir);
+ }
+
#[tokio::test]
async fn closing_oracle_table_tab_keeps_connection_scoped_pool() {
let (state, dir) = test_app_state().await;
diff --git a/crates/dbx-core/src/db/agent_driver.rs b/crates/dbx-core/src/db/agent_driver.rs
index 9f6254c24a..cbd02a83ee 100644
--- a/crates/dbx-core/src/db/agent_driver.rs
+++ b/crates/dbx-core/src/db/agent_driver.rs
@@ -880,6 +880,11 @@ impl PooledAgentClient {
self.shared_runtime.as_ref().is_none_or(|runtime| !runtime.is_failed())
}
+ /// True when this client is attached to a multi-session shared agent process.
+ pub fn uses_shared_runtime(&self) -> bool {
+ self.shared_runtime.is_some()
+ }
+
/// Terminates a protocol-v2 shared runtime without waiting for the logical session lock.
/// Legacy single-session clients can only be killed immediately when they are not busy.
pub fn fail_stop(&self) -> bool {
@@ -1151,12 +1156,15 @@ pub enum AgentMethod {
GetExplainInfo,
ExecuteBatch,
ExecuteTransaction,
+ BeginManualTransaction,
+ CommitManualTransaction,
+ RollbackManualTransaction,
Disconnect,
Shutdown,
}
impl AgentMethod {
- pub const ALL: [Self; 36] = [
+ pub const ALL: [Self; 39] = [
Self::Handshake,
Self::Connect,
Self::OpenSession,
@@ -1191,6 +1199,9 @@ impl AgentMethod {
Self::GetExplainInfo,
Self::ExecuteBatch,
Self::ExecuteTransaction,
+ Self::BeginManualTransaction,
+ Self::CommitManualTransaction,
+ Self::RollbackManualTransaction,
Self::Disconnect,
Self::Shutdown,
];
@@ -1232,6 +1243,9 @@ impl AgentMethod {
Self::GetExplainInfo => "get_explain_info",
Self::ExecuteBatch => "execute_batch",
Self::ExecuteTransaction => "execute_transaction",
+ Self::BeginManualTransaction => "begin_manual_transaction",
+ Self::CommitManualTransaction => "commit_manual_transaction",
+ Self::RollbackManualTransaction => "rollback_manual_transaction",
Self::Disconnect => "disconnect",
Self::Shutdown => "shutdown",
}
@@ -2448,6 +2462,24 @@ impl AgentDriverClient {
.await
}
+ pub async fn begin_manual_transaction(
+ &mut self,
+ schema: Option<&str>,
+ ) -> Result {
+ self.invalidate_cached_query();
+ self.call_method(AgentMethod::BeginManualTransaction, agent_manual_transaction_params(schema)).await
+ }
+
+ pub async fn commit_manual_transaction(&mut self) -> Result {
+ self.invalidate_cached_query();
+ self.call_method(AgentMethod::CommitManualTransaction, serde_json::json!({})).await
+ }
+
+ pub async fn rollback_manual_transaction(&mut self) -> Result {
+ self.invalidate_cached_query();
+ self.call_method(AgentMethod::RollbackManualTransaction, serde_json::json!({})).await
+ }
+
pub async fn execute_batch(
&mut self,
database: Option<&str>,
@@ -2853,6 +2885,14 @@ pub fn agent_transaction_params(database: Option<&str>, statements: &[String], s
})
}
+pub fn agent_manual_transaction_params(schema: Option<&str>) -> Value {
+ let schema = schema.map(str::trim).filter(|schema| !schema.is_empty());
+ match schema {
+ Some(schema) => serde_json::json!({ "schema": schema }),
+ None => serde_json::json!({}),
+ }
+}
+
pub fn mongo_database_params(database: &str) -> Value {
serde_json::json!({ "database": database })
}
@@ -4476,6 +4516,9 @@ for line in sys.stdin:
assert_eq!(AgentMethod::CloseTableReadSession.as_str(), "close_table_read_session");
assert_eq!(AgentMethod::ExecuteBatch.as_str(), "execute_batch");
assert_eq!(AgentMethod::ExecuteTransaction.as_str(), "execute_transaction");
+ assert_eq!(AgentMethod::BeginManualTransaction.as_str(), "begin_manual_transaction");
+ assert_eq!(AgentMethod::CommitManualTransaction.as_str(), "commit_manual_transaction");
+ assert_eq!(AgentMethod::RollbackManualTransaction.as_str(), "rollback_manual_transaction");
assert_eq!(AgentMethod::Disconnect.as_str(), "disconnect");
assert_eq!(AgentMethod::Shutdown.as_str(), "shutdown");
}
diff --git a/crates/dbx-core/src/query.rs b/crates/dbx-core/src/query.rs
index 37da4a642a..48843da5e0 100644
--- a/crates/dbx-core/src/query.rs
+++ b/crates/dbx-core/src/query.rs
@@ -4315,24 +4315,28 @@ async fn begin_transaction_session(
) -> Result {
let mysql_catalog_dialect = connection_mysql_catalog_dialect(state, connection_id).await;
let pool_database = query_pool_database(database, catalog);
- let pool_key = state.get_or_create_pool(connection_id, pool_database).await?;
+ // Probe the primary pool to learn the backend kind. Agent manual TX opens a
+ // dedicated multi_session workload pool so sticky TX state is isolated.
+ let probe_pool_key = state.get_or_create_pool(connection_id, pool_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),
+ Agent,
}
let pool_handle = {
let connections = state.connections.read().await;
- match connections.get(&pool_key).ok_or("Connection not found")? {
+ match connections.get(&probe_pool_key).ok_or("Connection not found")? {
PoolKind::Postgres(pg) => TxnPoolHandle::Postgres(pg.clone()),
PoolKind::Mysql(mp, _) => TxnPoolHandle::Mysql(mp.clone()),
+ PoolKind::Agent(_) if !consistent_snapshot => TxnPoolHandle::Agent,
_ => return Err("Manual transaction is not supported for this database type".to_string()),
}
}; // connections lock released here
- let txn_conn = match pool_handle {
+ let (txn_conn, pool_key) = match pool_handle {
TxnPoolHandle::Postgres(pg_pool) => {
let conn = pg_pool.get().await.map_err(|e| format!("Failed to get Postgres connection: {e}"))?;
let begin_sql = postgres_transaction_begin_sql(consistent_snapshot);
@@ -4347,7 +4351,7 @@ async fn begin_transaction_session(
.await
.map_err(|e| format!("SET search_path failed: {e}"))?;
}
- TxnConnection::Postgres(Box::new(conn))
+ (TxnConnection::Postgres(Box::new(conn)), probe_pool_key.clone())
}
TxnPoolHandle::Mysql(mysql_pool) => {
let mut conn = mysql_pool.get_conn().await.map_err(|e| format!("Failed to get MySQL connection: {e}"))?;
@@ -4371,7 +4375,45 @@ async fn begin_transaction_session(
if !syntax_errors.is_empty() {
return Err(format!("START TRANSACTION failed for all compatible forms: {}", syntax_errors.join("; ")));
}
- TxnConnection::Mysql(conn)
+ (TxnConnection::Mysql(conn), probe_pool_key.clone())
+ }
+ TxnPoolHandle::Agent => {
+ let client_session_id = format!("manual-txn-{}", uuid::Uuid::new_v4());
+ let agent_pool_key =
+ state.get_or_create_pool_for_session(connection_id, pool_database, Some(&client_session_id)).await?;
+ let client = {
+ let connections = state.connections.read().await;
+ match connections.get(&agent_pool_key) {
+ Some(PoolKind::Agent(client)) => client.clone(),
+ _ => {
+ let _ = state.close_client_session_pool(connection_id, pool_database, &client_session_id).await;
+ return Err("Agent connection not found for manual transaction".to_string());
+ }
+ }
+ };
+ let begin_result = {
+ let mut locked = client.lock().await;
+ locked.begin_manual_transaction::(schema).await
+ };
+ if let Err(error) = begin_result {
+ let _ = state.close_client_session_pool(connection_id, pool_database, &client_session_id).await;
+ return Err(format!("BEGIN manual transaction failed: {error}"));
+ }
+ let Some(cleanup_guard) =
+ state.workload_session_pool_cleanup_guard(connection_id, pool_database, &client_session_id).await
+ else {
+ let _ = state.close_client_session_pool(connection_id, pool_database, &client_session_id).await;
+ return Err("Manual transaction requires a dedicated Agent session".to_string());
+ };
+ (
+ TxnConnection::Agent {
+ client,
+ client_session_id,
+ database: pool_database.map(str::to_string),
+ cleanup_guard,
+ },
+ agent_pool_key,
+ )
}
};
@@ -4503,16 +4545,17 @@ pub async fn execute_in_manual_transaction(
}
if session.last_activity.elapsed() > TXN_IDLE_TIMEOUT {
let session = sessions.remove(txn_session_id).expect("session exists");
- Some(session.connection)
+ Some(session)
} else {
session.busy = true;
session.last_activity = std::time::Instant::now();
None
}
};
- if let Some(connection) = connection {
- let mut conn = connection.lock().await;
+ if let Some(session) = connection {
+ let mut conn = session.connection.lock().await;
let _ = rollback_manual_txn_connection(&mut conn).await;
+ release_manual_txn_agent_pool(state, &session.connection_id, &mut conn).await;
return Err("Transaction was auto-rolled back due to 5 minutes of inactivity".to_string());
}
@@ -4533,6 +4576,9 @@ pub async fn execute_in_manual_transaction(
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,
+ TxnConnection::Agent { client, .. } => {
+ execute_manual_txn_agent_statement(client, db_type, statement, _database, _schema, row_limit).await
+ }
};
match result {
Ok(query_result) => results.push(query_result),
@@ -4545,8 +4591,9 @@ pub async fn execute_in_manual_transaction(
};
if should_rollback {
let _ = rollback_manual_txn_connection(&mut conn).await;
+ release_manual_txn_agent_pool(state, &connection_id, &mut conn).await;
}
- return Err(format!("Statement {} failed: {}. Transaction was auto-rolled back.", i + 1, e));
+ return Err(format!("Statement {} failed: {}. The manual transaction was rolled back.", i + 1, e));
}
}
}
@@ -4679,12 +4726,19 @@ where
},
Err(err) => Err(format!("Query failed: {err}")),
},
+ TxnConnection::Agent { .. } => {
+ Err("Streaming rows inside an agent manual transaction is not supported".to_string())
+ }
};
if let Err(err) = &stream_result {
- let should_rollback = state.transaction_sessions.write().await.remove(txn_session_id).is_some();
- if should_rollback {
+ let removed = {
+ let mut sessions = state.transaction_sessions.write().await;
+ sessions.remove(txn_session_id)
+ };
+ if let Some(session) = removed {
let _ = rollback_manual_txn_connection(&mut conn).await;
+ release_manual_txn_agent_pool(state, &session.connection_id, &mut conn).await;
}
return Err(format!("{err}. Transaction was auto-rolled back."));
}
@@ -4714,10 +4768,51 @@ async fn rollback_manual_txn_connection(conn: &mut TxnConnection) -> Result<(),
TxnConnection::Mysql(conn) => {
conn.query_drop("ROLLBACK").await.map_err(|e| format!("ROLLBACK failed: {e}"))?;
}
+ TxnConnection::Agent { client, .. } => {
+ let mut locked = client.lock().await;
+ match locked.rollback_manual_transaction::().await {
+ Ok(_) => {}
+ Err(error) if error.to_ascii_lowercase().contains("no manual transaction") => {}
+ Err(error) => return Err(format!("ROLLBACK failed: {error}")),
+ }
+ // Close the dedicated agent session so Oracle resources are released even if
+ // the pool map entry is cleaned up later.
+ let _ = locked.disconnect().await;
+ }
}
Ok(())
}
+async fn release_manual_txn_agent_pool(state: &AppState, connection_id: &str, conn: &mut TxnConnection) {
+ let TxnConnection::Agent { client_session_id, database, cleanup_guard, .. } = conn else {
+ return;
+ };
+ if state.detach_client_session_pool(connection_id, database.as_deref(), client_session_id).await.unwrap_or(false) {
+ cleanup_guard.disarm();
+ }
+}
+
+async fn execute_manual_txn_agent_statement(
+ client: &Arc,
+ db_type: Option,
+ statement: &str,
+ database: &str,
+ schema: Option<&str>,
+ row_limit: usize,
+) -> Result {
+ let sql = sql_for_execution_context(db_type, statement, schema);
+ let execution_schema = schema_for_execution_context(db_type, schema);
+ let options = QueryExecutionOptions { max_rows: Some(row_limit.max(1)), ..QueryExecutionOptions::default() };
+ let params =
+ agent_execute_query_params(&sql, Some(database).filter(|value| !value.is_empty()), execution_schema, options);
+ let mut locked = client.lock().await;
+ locked
+ .execute_query_typed_with_timeout::(params, None)
+ .await
+ .map(|result| truncate_result_with_max_rows(result, Some(row_limit.max(1))))
+ .map_err(|error| error.into_legacy_string())
+}
+
/// 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,
@@ -4865,7 +4960,16 @@ pub async fn commit_manual_transaction(state: &AppState, txn_session_id: &str) -
TxnConnection::Mysql(conn) => {
conn.query_drop("COMMIT").await.map_err(|e| format!("COMMIT failed: {e}"))?;
}
+ TxnConnection::Agent { client, .. } => {
+ let mut locked = client.lock().await;
+ locked
+ .commit_manual_transaction::()
+ .await
+ .map_err(|error| format!("COMMIT failed: {error}"))?;
+ let _ = locked.disconnect().await;
+ }
}
+ release_manual_txn_agent_pool(state, &session.connection_id, &mut conn).await;
log::info!("[query][manual_txn:commit] session_id={}", txn_session_id);
Ok(db::QueryResult {
@@ -4894,6 +4998,7 @@ pub async fn rollback_manual_transaction(state: &AppState, txn_session_id: &str)
let mut conn = session.connection.lock().await;
rollback_manual_txn_connection(&mut conn).await?;
+ release_manual_txn_agent_pool(state, &session.connection_id, &mut conn).await;
log::info!("[query][manual_txn:rollback] session_id={}", txn_session_id);
Ok(db::QueryResult {
diff --git a/packages/app-tests/openTabsPersistence.test.ts b/packages/app-tests/openTabsPersistence.test.ts
index 5e5d29135c..8f51e24304 100644
--- a/packages/app-tests/openTabsPersistence.test.ts
+++ b/packages/app-tests/openTabsPersistence.test.ts
@@ -42,6 +42,14 @@ test("serializes unsaved query tabs with editor context", () => {
]);
});
+test("round-trips query transaction mode", () => {
+ const saved = serializeOpenTabs([queryTab({ autoCommit: false })]);
+ const restored = restoreOpenTabsState(JSON.stringify(saved), "tab-1");
+
+ assert.equal(saved[0]?.autoCommit, false);
+ assert.equal(restored.tabs[0]?.autoCommit, false);
+});
+
test("serializes object source query tabs with save context", () => {
const saved = serializeOpenTabs([
queryTab({
diff --git a/packages/app-tests/queryStore.test.ts b/packages/app-tests/queryStore.test.ts
index edfe6d5a55..637959c4a0 100644
--- a/packages/app-tests/queryStore.test.ts
+++ b/packages/app-tests/queryStore.test.ts
@@ -296,6 +296,33 @@ test("external SQL file paths persist with open query tabs", async () => {
}
});
+test("legacy Oracle query tabs restore with the manual transaction default", async () => {
+ const restoreStorage = installMemoryStorage();
+ try {
+ setActivePinia(createPinia());
+ let connectionStore = useConnectionStore();
+ connectionStore.connections = [oracleConn("oracle-1")];
+ let store = useQueryStore();
+ const tabId = store.createTab("oracle-1", "ORCL", "Query");
+ await store.flushPendingPersist();
+
+ const storageKey = "dbx-app-state:open_tabs";
+ const saved = JSON.parse(localStorage.getItem(storageKey) ?? "null");
+ delete saved.tabs[0].autoCommit;
+ localStorage.setItem(storageKey, JSON.stringify(saved));
+
+ setActivePinia(createPinia());
+ connectionStore = useConnectionStore();
+ connectionStore.connections = [oracleConn("oracle-1")];
+ store = useQueryStore();
+ await store.initOpenTabs();
+
+ assert.equal(store.tabs.find((tab) => tab.id === tabId)?.autoCommit, false);
+ } finally {
+ restoreStorage();
+ }
+});
+
test("clean saved SQL tabs persist without duplicating SQL text", async () => {
const restoreStorage = installMemoryStorage();
try {
@@ -2008,6 +2035,7 @@ test("normalizes unquoted Oracle query identifiers before loading editable metad
try {
const tabId = store.createTab("oracle-1", "ORCL", "Query 1", "query", "app");
+ store.setAutoCommit(tabId, true);
await store.executeTabSql(tabId, "select id, name from users");
const tab = store.tabs.find((item) => item.id === tabId);
@@ -4418,6 +4446,7 @@ for (const pageSize of [1_000, 25_000, 100_000]) {
settingsStore.updateEditorSettings({ pageSize });
connectionStore.addEphemeralConnection(oracleConn("oracle-1"));
const tabId = store.createTab("oracle-1", "ORCL", "Query", "query", "APP");
+ store.setAutoCommit(tabId, true);
globalThis.fetch = withConnectionHealthMock(async (input, init) => {
const url = String(input);
@@ -6678,6 +6707,7 @@ test("query execution waits for a cleared schema client session to close", async
connectionStore.addEphemeralConnection(oracleConn("oracle-1"));
const tabId = store.createTab("oracle-1", "ORCL", "Query", "query", "REPORTING");
+ store.setAutoCommit(tabId, true);
let resolveClientSessionClose: ((response: Response) => void) | undefined;
let executeRequests = 0;
@@ -6740,7 +6770,9 @@ test("failed schema session reset blocks query and Oracle explain execution", as
connectionStore.addEphemeralConnection(oracleConn("oracle-1"));
const queryTabId = store.createTab("oracle-1", "ORCL", "Query", "query", "REPORTING");
+ store.setAutoCommit(queryTabId, true);
const explainTabId = store.createTab("oracle-1", "ORCL", "Explain", "query", "REPORTING");
+ store.setAutoCommit(explainTabId, true);
globalThis.fetch = async (input) => {
const url = String(input);
@@ -7181,6 +7213,7 @@ for (const resultState of [
settingsStore.updateEditorSettings({ pageSize: 100_000, autoCalculateTotalRows: true });
connectionStore.addEphemeralConnection(oracleConn("oracle-1"));
const tabId = store.createTab("oracle-1", "ORCL", "Query", "query", "APP");
+ store.setAutoCommit(tabId, true);
const tab = store.tabs.find((item) => item.id === tabId);
assert.ok(tab);
@@ -7255,6 +7288,8 @@ for (const paginationMode of [
settingsStore.updateEditorSettings({ autoCalculateTotalRows: true });
connectionStore.addEphemeralConnection(paginationMode.useAgentResultSession ? oracleConn("conn-1") : conn("conn-1"));
const tabId = store.createTab("conn-1", "db", "Query", "query", "public");
+ // Oracle defaults to manual TX; these cases exercise auto-commit pagination.
+ store.setAutoCommit(tabId, true);
const tab = store.tabs.find((item) => item.id === tabId);
assert.ok(tab);