🐛 fix(query-editor): 修复挂起事务内查询结果不一致

2026-08-09 16:23:27 +08:00 · 2026-06-24 11:37:34 +08:00
parent 2b4beae9df
commit d2c4160c6d
8 changed files with 281 additions and 8 deletions
--- a/frontend/src/components/QueryEditor.external-sql-save.test.tsx
+++ b/frontend/src/components/QueryEditor.external-sql-save.test.tsx
@@ -100,6 +100,7 @@ const backendApp = vi.hoisted(() => ({
  DBQuery: vi.fn(),
  DBQueryWithCancel: vi.fn(),
  DBQueryMulti: vi.fn(),
+  DBQueryMultiInTransaction: vi.fn(),
  DBQueryMultiTransactional: vi.fn(),
  DBCommitTransaction: vi.fn(),
  DBRollbackTransaction: vi.fn(),
@@ -673,6 +674,7 @@ describe('QueryEditor external SQL save', () => {
    backendApp.ExportSQLFile.mockResolvedValue({ success: true });
    backendApp.DBQueryWithCancel.mockResolvedValue({ success: true, data: [] });
    backendApp.DBQueryMulti.mockResolvedValue({ success: true, data: [] });
+    backendApp.DBQueryMultiInTransaction.mockResolvedValue({ success: true, data: [] });
    backendApp.DBQueryMultiTransactional.mockResolvedValue({ success: true, data: [] });
    backendApp.DBCommitTransaction.mockResolvedValue({ success: true, message: '事务已提交' });
    backendApp.DBRollbackTransaction.mockResolvedValue({ success: true, message: '事务已回滚' });
@@ -4760,6 +4762,62 @@ describe('QueryEditor external SQL save', () => {
    expect(textContent(renderer!.root)).not.toContain('未提交');
  });

+  it('reuses the pending managed transaction for follow-up read-only SQL in the same tab', async () => {
+    backendApp.DBQueryMultiTransactional.mockResolvedValueOnce({
+      success: true,
+      transactionId: 'tx-1',
+      transactionPending: true,
+      data: [
+        { columns: ['affectedRows'], rows: [{ affectedRows: 1 }], statementIndex: 1 },
+      ],
+    });
+    backendApp.DBQueryMultiInTransaction.mockResolvedValueOnce({
+      success: true,
+      transactionId: 'tx-1',
+      transactionPending: true,
+      data: [
+        { columns: ['name'], rows: [{ name: 'new' }], statementIndex: 1 },
+      ],
+    });
+
+    let renderer!: ReactTestRenderer;
+    await act(async () => {
+      renderer = create(<QueryEditor tab={createTab({ query: "UPDATE users SET name = 'new' WHERE id = 1" })} />);
+    });
+
+    await act(async () => {
+      await findButton(renderer!, '运行').props.onClick();
+    });
+    await act(async () => {
+      await Promise.resolve();
+      await Promise.resolve();
+    });
+
+    await act(async () => {
+      renderer.update(<QueryEditor tab={createTab({ query: 'SELECT name FROM users WHERE id = 1' })} />);
+    });
+
+    await act(async () => {
+      await findButton(renderer!, '运行').props.onClick();
+    });
+    await act(async () => {
+      await Promise.resolve();
+      await Promise.resolve();
+    });
+
+    expect(backendApp.DBQueryMultiTransactional).toHaveBeenCalledTimes(1);
+    expect(backendApp.DBQueryMultiInTransaction).toHaveBeenCalledWith(
+      'tx-1',
+      expect.stringContaining('SELECT name FROM users'),
+      'query-1',
+    );
+    expect(backendApp.DBQueryMulti).not.toHaveBeenCalled();
+    expect(dataGridState.latestProps?.columnNames).toEqual(['name']);
+    expect(dataGridState.latestProps?.data?.[0]).toMatchObject({ name: 'new' });
+    expect(textContent(renderer!.root)).toContain('提交');
+    expect(textContent(renderer!.root)).toContain('回滚');
+  });
+
  it('runs SQL editor WITH DML through a pending managed transaction', async () => {
    const sql = 'WITH target AS (SELECT id FROM users WHERE active = 1) UPDATE users SET synced = 1 WHERE id IN (SELECT id FROM target)';
    backendApp.DBQueryMultiTransactional.mockResolvedValueOnce({
--- a/frontend/src/components/QueryEditor.tsx
+++ b/frontend/src/components/QueryEditor.tsx
@@ -6,7 +6,7 @@ import { format } from 'sql-formatter';
 import { v4 as uuidv4 } from 'uuid';
 import { TabData, ColumnDefinition } from '../types';
 import { useStore } from '../store';
-import { DBQuery, DBQueryWithCancel, DBQueryMulti, DBQueryMultiTransactional, DBGetTables, DBGetAllColumns, DBGetDatabases, DBGetColumns, CancelQuery, GenerateQueryID, WriteSQLFile, ExportSQLFile } from '../../wailsjs/go/app/App';
+import { DBQuery, DBQueryWithCancel, DBQueryMulti, DBQueryMultiInTransaction, DBQueryMultiTransactional, DBGetTables, DBGetAllColumns, DBGetDatabases, DBGetColumns, CancelQuery, GenerateQueryID, WriteSQLFile, ExportSQLFile } from '../../wailsjs/go/app/App';
 import { GONAVI_ROW_KEY } from './DataGrid';
 import { findConnectionMutatingStatements } from '../utils/connectionReadOnly';
 import { getDataSourceCapabilities, shouldShowOceanBaseRowNumberColumn } from '../utils/dataSourceCapabilities';
@@ -25,7 +25,7 @@ import {
 import { extractQueryResultTableRef, type QueryResultTableRef } from '../utils/queryResultTable';
 import { quoteIdentPart, quoteQualifiedIdent } from '../utils/sql';
 import { formatSqlExecutionError, hasLocalizedSqlTimeoutKeyword } from '../utils/sqlErrorSemantics';
-import { shouldUseSqlEditorManagedTransactionForType } from '../utils/sqlEditorTransaction';
+import { canReusePendingSqlEditorTransactionForType, shouldUseSqlEditorManagedTransactionForType } from '../utils/sqlEditorTransaction';
 import { findSqlStatementRanges, resolveCurrentSqlStatementRange, resolveExecutableSql } from '../utils/sqlStatementSelection';
 import { isMacLikePlatform } from '../utils/appearance';
 import { splitSidebarQualifiedName } from '../utils/sidebarLocate';
@@ -2874,6 +2874,20 @@ const QueryEditor: React.FC<{ tab: TabData; isActive?: boolean }> = ({ tab, isAc
      }
  };

+  const executeSqlEditorMultiQuery = useCallback((
+      config: Record<string, any>,
+      dbName: string,
+      sql: string,
+      queryId: string,
+      sourceStatements: string[],
+  ) => {
+      const pendingTransaction = pendingSqlTransactionRef.current;
+      if (pendingTransaction && canReusePendingSqlEditorTransactionForType(String(config.type || ''), sourceStatements)) {
+          return DBQueryMultiInTransaction(pendingTransaction.id, sql, queryId);
+      }
+      return DBQueryMulti(buildRpcConnectionConfig(config) as any, dbName, sql, queryId);
+  }, []);
+
  // 精准重查询单个结果集（提交事务 / 刷新按钮使用），不会重跑整个编辑器 SQL
  const handleReloadResult = async (resultKey: string, sql: string) => {
      if (!sql?.trim() || !currentDb) return;
@@ -2893,14 +2907,20 @@ const QueryEditor: React.FC<{ tab: TabData; isActive?: boolean }> = ({ tab, isAc

      try {
          setLoading(true);
-          // 使用 DBQueryMulti 保持和首次查询一致的后端路径
+          // 保持与首次执行一致的后端路径，必要时复用挂起事务
          let queryId: string;
          try {
              queryId = await GenerateQueryID();
          } catch {
              queryId = 'reload-' + Date.now();
          }
-          const res = await DBQueryMulti(buildRpcConnectionConfig(config) as any, currentDb, sql, queryId);
+          const res = await executeSqlEditorMultiQuery(
+              config,
+              currentDb,
+              sql,
+              queryId,
+              splitSQLStatements(sql),
+          );
          if (!res?.success) {
              message.error(translate('query_editor.message.refresh_failed', {
                  error: formatSqlExecutionError(res?.message || translate('common.unknown'), { translate }),
@@ -2997,7 +3017,13 @@ const QueryEditor: React.FC<{ tab: TabData; isActive?: boolean }> = ({ tab, isAc
          } catch {
              queryId = 'query-page-' + Date.now();
          }
-          const res = await DBQueryMulti(buildRpcConnectionConfig(config) as any, currentDb, pageSql, queryId);
+          const res = await executeSqlEditorMultiQuery(
+              config,
+              currentDb,
+              pageSql,
+              queryId,
+              splitSQLStatements(pageSql),
+          );
          if (!res?.success) {
              message.error(translate('query_editor.message.page_query_failed', {
                  error: formatSqlExecutionError(res?.message || translate('common.unknown'), { translate }),
@@ -3417,7 +3443,8 @@ const QueryEditor: React.FC<{ tab: TabData; isActive?: boolean }> = ({ tab, isAc
                if (result.applied) anyLimitApplied = true;
                return { ...plan, executedSql: result.sql };
            });
-            const fullSQL = executablePlans.map((plan) => plan.executedSql).join(';\n');
+            const executableStatements = executablePlans.map((plan) => plan.executedSql);
+            const fullSQL = executableStatements.join(';\n');

            const startTime = Date.now();
            let queryId: string;
@@ -3429,8 +3456,15 @@ const QueryEditor: React.FC<{ tab: TabData; isActive?: boolean }> = ({ tab, isAc
            }
            setQueryId(queryId);

-            const queryExecutor = useManagedTransaction ? DBQueryMultiTransactional : DBQueryMulti;
-            const res = await queryExecutor(buildRpcConnectionConfig(config) as any, currentDb, fullSQL, queryId);
+            const res = useManagedTransaction
+                ? await DBQueryMultiTransactional(buildRpcConnectionConfig(config) as any, currentDb, fullSQL, queryId)
+                : await executeSqlEditorMultiQuery(
+                    config,
+                    currentDb,
+                    fullSQL,
+                    queryId,
+                    executableStatements,
+                );
            const duration = Date.now() - startTime;

            addSqlLog({
--- a/frontend/src/utils/sqlEditorTransaction.test.ts
+++ b/frontend/src/utils/sqlEditorTransaction.test.ts
@@ -1,6 +1,7 @@
 import { describe, expect, it } from 'vitest';

 import {
+  canReusePendingSqlEditorTransactionForType,
  resolveSqlEditorOperationKeyword,
  shouldUseSqlEditorManagedTransaction,
  shouldUseSqlEditorManagedTransactionForType,
@@ -51,4 +52,19 @@ describe('sqlEditorTransaction', () => {
      'UPDATE hive.default.orders SET status = \'done\'',
    ])).toBe(false);
  });
+
+  it('reuses a pending managed transaction only for read-only follow-up SQL', () => {
+    expect(canReusePendingSqlEditorTransactionForType('mysql', [
+      'SELECT * FROM users WHERE id = 1',
+    ])).toBe(true);
+    expect(canReusePendingSqlEditorTransactionForType('mysql', [
+      'WITH target AS (SELECT id FROM users) SELECT * FROM target',
+    ])).toBe(true);
+    expect(canReusePendingSqlEditorTransactionForType('mysql', [
+      'UPDATE users SET name = "n" WHERE id = 1',
+    ])).toBe(false);
+    expect(canReusePendingSqlEditorTransactionForType('mysql', [
+      'COMMIT',
+    ])).toBe(false);
+  });
 });
--- a/frontend/src/utils/sqlEditorTransaction.ts
+++ b/frontend/src/utils/sqlEditorTransaction.ts
@@ -274,3 +274,26 @@ export const shouldUseSqlEditorManagedTransactionForType = (

 export const shouldUseSqlEditorManagedTransaction = (statements: string[]): boolean =>
    shouldUseSqlEditorManagedTransactionForType('', statements);
+
+export const canReusePendingSqlEditorTransactionForType = (
+    type: string,
+    statements: string[],
+): boolean => {
+    if (String(type || '').trim().toLowerCase() === 'trino') {
+        return false;
+    }
+    let hasReadStatement = false;
+    for (const statement of statements) {
+        const trimmed = String(statement || '').trim();
+        if (!trimmed) continue;
+        if (isSqlEditorTransactionControlStatement(trimmed)) return false;
+        if (sqlEditorStatementHasManagedWrite(trimmed)) return false;
+        const keyword = resolveSqlEditorOperationKeyword(trimmed);
+        if (!SQL_EDITOR_READ_KEYWORDS.has(keyword)) return false;
+        hasReadStatement = true;
+    }
+    return hasReadStatement;
+};
+
+export const canReusePendingSqlEditorTransaction = (statements: string[]): boolean =>
+    canReusePendingSqlEditorTransactionForType('', statements);
--- a/frontend/wailsjs/go/app/App.d.ts
+++ b/frontend/wailsjs/go/app/App.d.ts
@@ -60,6 +60,8 @@ export function DBQueryIsolated(arg1:connection.ConnectionConfig,arg2:string,arg

 export function DBQueryMulti(arg1:connection.ConnectionConfig,arg2:string,arg3:string,arg4:string):Promise<connection.QueryResult>;

+export function DBQueryMultiInTransaction(arg1:string,arg2:string,arg3:string):Promise<connection.QueryResult>;
+
 export function DBQueryMultiTransactional(arg1:connection.ConnectionConfig,arg2:string,arg3:string,arg4:string):Promise<connection.QueryResult>;

 export function DBQueryWithCancel(arg1:connection.ConnectionConfig,arg2:string,arg3:string,arg4:string):Promise<connection.QueryResult>;
--- a/frontend/wailsjs/go/app/App.js
+++ b/frontend/wailsjs/go/app/App.js
@@ -110,6 +110,10 @@ export function DBQueryMulti(arg1, arg2, arg3, arg4) {
  return window['go']['app']['App']['DBQueryMulti'](arg1, arg2, arg3, arg4);
 }

+export function DBQueryMultiInTransaction(arg1, arg2, arg3) {
+  return window['go']['app']['App']['DBQueryMultiInTransaction'](arg1, arg2, arg3);
+}
+
 export function DBQueryMultiTransactional(arg1, arg2, arg3, arg4) {
  return window['go']['app']['App']['DBQueryMultiTransactional'](arg1, arg2, arg3, arg4);
 }
--- a/internal/app/methods_db_multi_test.go
+++ b/internal/app/methods_db_multi_test.go
@@ -687,6 +687,84 @@ func TestDBQueryMultiTransactionalKeepsDMLTransactionOpenUntilCommit(t *testing.
 	}
 }

+func TestDBQueryMultiInTransactionReusesPendingManagedSessionForReadQueries(t *testing.T) {
+	originalNewDatabaseFunc := newDatabaseFunc
+	t.Cleanup(func() {
+		newDatabaseFunc = originalNewDatabaseFunc
+	})
+
+	updateStmt := "UPDATE users SET name = 'new' WHERE id = 1"
+	readStmt := "SELECT name FROM users WHERE id = 1"
+	fakeDB := &fakeTransactionalDB{
+		fakeBatchWriteDB: fakeBatchWriteDB{
+			execAffected: map[string]int64{
+				updateStmt: 1,
+			},
+			queryMap: map[string][]map[string]interface{}{
+				readStmt: {
+					{"name": "new"},
+				},
+			},
+			fieldMap: map[string][]string{
+				readStmt: {"name"},
+			},
+		},
+	}
+	newDatabaseFunc = func(dbType string) (db.Database, error) {
+		return fakeDB, nil
+	}
+
+	app := NewAppWithSecretStore(secretstore.NewUnavailableStore("test"))
+	config := connection.ConnectionConfig{Type: "mysql", Host: "127.0.0.1", Port: 3306, User: "root"}
+
+	startResult := app.DBQueryMultiTransactional(config, "main", updateStmt, "tx-query")
+	if !startResult.Success {
+		t.Fatalf("expected transactional update success, got failure: %s", startResult.Message)
+	}
+	if startResult.TransactionID == "" || !startResult.TransactionPending {
+		t.Fatalf("expected pending transaction metadata, got id=%q pending=%v", startResult.TransactionID, startResult.TransactionPending)
+	}
+	if fakeDB.txSession == nil {
+		t.Fatal("expected transaction provider session to be opened")
+	}
+	if fakeDB.txSession.closed {
+		t.Fatal("expected transaction session to stay open before follow-up read")
+	}
+
+	readResult := app.DBQueryMultiInTransaction(startResult.TransactionID, readStmt, "tx-query-read")
+	if !readResult.Success {
+		t.Fatalf("expected in-transaction read success, got failure: %s", readResult.Message)
+	}
+	if readResult.TransactionID != startResult.TransactionID || !readResult.TransactionPending {
+		t.Fatalf("expected follow-up read to preserve pending transaction metadata, got id=%q pending=%v", readResult.TransactionID, readResult.TransactionPending)
+	}
+	if fakeDB.txSession.queryCalls == 0 {
+		t.Fatal("expected follow-up read to execute on the pinned transaction session")
+	}
+	if fakeDB.txSession.closed {
+		t.Fatal("expected transaction session to remain open after follow-up read")
+	}
+
+	resultSets, ok := readResult.Data.([]connection.ResultSetData)
+	if !ok {
+		t.Fatalf("expected []connection.ResultSetData from in-transaction read, got %T", readResult.Data)
+	}
+	if len(resultSets) != 1 {
+		t.Fatalf("expected one read result set, got %#v", resultSets)
+	}
+	if got := resultSets[0].Rows[0]["name"]; got != "new" {
+		t.Fatalf("expected in-transaction read to return updated value, got %#v", got)
+	}
+
+	rollbackResult := app.DBRollbackTransaction(startResult.TransactionID)
+	if !rollbackResult.Success {
+		t.Fatalf("expected rollback success after follow-up read, got failure: %s", rollbackResult.Message)
+	}
+	if !fakeDB.txSession.closed {
+		t.Fatal("expected transaction session to close after rollback")
+	}
+}
+
 func TestDBQueryMultiTransactionalUsesImplicitSessionTransactionForOracle(t *testing.T) {
 	originalNewDatabaseFunc := newDatabaseFunc
 	t.Cleanup(func() {
--- a/internal/app/methods_db_transaction.go
+++ b/internal/app/methods_db_transaction.go
@@ -203,6 +203,64 @@ func (a *App) DBQueryMultiTransactional(config connection.ConnectionConfig, dbNa
 	}
 }

+// DBQueryMultiInTransaction executes follow-up SQL in an existing SQL editor managed transaction.
+// The transaction remains open until DBCommitTransaction or DBRollbackTransaction is called.
+func (a *App) DBQueryMultiInTransaction(transactionID string, query string, queryID string) connection.QueryResult {
+	transactionID = strings.TrimSpace(transactionID)
+	if transactionID == "" {
+		return connection.QueryResult{Success: false, Message: a.appText("db.backend.error.transaction_id_required", nil), QueryID: queryID}
+	}
+	if queryID == "" {
+		queryID = generateQueryID()
+	}
+
+	a.sqlTransactionMu.Lock()
+	tx, ok := a.sqlTransactions[transactionID]
+	a.sqlTransactionMu.Unlock()
+	if !ok || tx == nil || tx.execer == nil {
+		return connection.QueryResult{Success: false, Message: a.appText("db.backend.error.transaction_not_found", nil), QueryID: queryID}
+	}
+
+	runConfig := connection.ConnectionConfig{Type: tx.dbType}
+	query = sanitizeSQLForPgLike(tx.dbType, query)
+	statements := splitSQLStatements(query)
+
+	ctx, cancel := newQueryExecutionContext(runConfig)
+	defer cancel()
+
+	a.queryMu.Lock()
+	a.runningQueries[queryID] = queryContext{
+		cancel:  cancel,
+		started: time.Now(),
+	}
+	a.queryMu.Unlock()
+	defer func() {
+		a.queryMu.Lock()
+		delete(a.runningQueries, queryID)
+		a.queryMu.Unlock()
+	}()
+
+	resultSets, err := executeManagedSQLTransactionStatements(ctx, tx.execer, runConfig, statements, a.appText)
+	if err != nil {
+		logger.Error(err, "DBQueryMultiInTransaction 执行失败：id=%s dbType=%s SQL片段=%q", transactionID, tx.dbType, sqlSnippet(query))
+		return connection.QueryResult{
+			Success:            false,
+			Message:            err.Error(),
+			QueryID:            queryID,
+			TransactionID:      transactionID,
+			TransactionPending: true,
+		}
+	}
+
+	return connection.QueryResult{
+		Success:            true,
+		Data:               resultSets,
+		QueryID:            queryID,
+		TransactionID:      transactionID,
+		TransactionPending: true,
+	}
+}
+
 func executeManagedSQLTransactionStatements(ctx context.Context, session db.StatementExecer, runConfig connection.ConnectionConfig, statements []string, text func(string, map[string]any) string) ([]connection.ResultSetData, error) {
 	if text == nil {
 		text = defaultDBBackendText