feat(execution): multi-database SQL execution with scoped targets

This commit is contained in:
miracle authored and GitHub committed 2026-08-09 14:24:10 +08:00
1 parent 328a2aa107
commit 71ccb3d97d
46 files changed
+4421 -182

No files matched your search

@@ -18,6 +18,7 @@ const mocks = vi.hoisted(() => ({
tabs: [] as QueryTab[],
setTableMeta: vi.fn(),
clearInvalidDataTabSort: vi.fn(),
activeResultExecutionTarget: vi.fn(),
}));
vi.mock("vue-i18n", () => ({
@@ -45,6 +46,7 @@ vi.mock("@/stores/connectionStore", () => ({
vi.mock("@/stores/queryStore", () => ({
useQueryStore: () => ({
executeTabSql: mocks.executeTabSql,
activeResultExecutionTarget: mocks.activeResultExecutionTarget,
setExecuting: mocks.setExecuting,
updateSql: mocks.updateSql,
tabs: mocks.tabs,
@@ -120,6 +122,7 @@ describe("useDataGridActions", () => {
mocks.buildTableSelectSql.mockResolvedValue("SELECT * FROM public.users LIMIT 100 OFFSET 0");
mocks.buildSortedQuerySql.mockResolvedValue({ ok: true, sql: "SELECT sorted" });
mocks.ensureConnected.mockResolvedValue(undefined);
mocks.activeResultExecutionTarget.mockReturnValue(undefined);
mocks.getColumns.mockResolvedValue([{ name: "id", data_type: "integer", is_nullable: false, column_default: null, is_primary_key: true, extra: null }]);
mocks.listIndexes.mockResolvedValue([]);
});
@@ -405,4 +408,38 @@ describe("useDataGridActions", () => {
}),
);
});
it("uses the active multi-database result target for pagination", async () => {
const tab = {
id: "tab-1",
connectionId: "source-1",
database: "source_db",
title: "Query",
sql: "SELECT * FROM users",
resultBaseSql: "SELECT * FROM users",
resultPageLimit: 100,
resultPageOffset: 0,
result: { columns: ["id"], rows: [[1]], affected_rows: 0, execution_time_ms: 1 },
mode: "query",
isDirty: false,
isExecuting: false,
isExplaining: false,
} as QueryTab;
const target = { connectionId: "target-2", database: "reporting", schema: "audit" };
mocks.activeResultExecutionTarget.mockReturnValue(target);
mocks.getConfig.mockImplementation((id: string) => ({ id, db_type: "postgres" }));
const actions = useDataGridActions(computed(() => tab));
await actions.onPaginate(100, 100);
expect(mocks.executeTabSql).toHaveBeenCalledWith(
"tab-1",
"SELECT * FROM users",
expect.objectContaining({
executionTarget: target,
targetContext: { scope: "database", database: "reporting", schema: "audit" },
pagination: { offset: 100, limit: 100, sessionId: undefined },
}),
);
});
});
@@ -0,0 +1,188 @@
import { describe, expect, it, vi } from "vitest";
import { ref } from "vue";
import { useMultiDbExecution } from "@/composables/useMultiDbExecution";
const targets = [
{ connectionId: "conn-1", database: "app" },
{ connectionId: "conn-2", database: "report" },
{ connectionId: "conn-3", database: "audit" },
] as const;
describe("useMultiDbExecution", () => {
it("executes serially and continues after a target failure", async () => {
const executionOrder: string[] = [];
const executor = useMultiDbExecution(
{
executeTarget: async ({ target }) => {
executionOrder.push(target.connectionId);
return target.connectionId === "conn-1" ? { status: "failed", errorMessage: "boom" } : { status: "success" };
},
},
{ sourceTabId: ref("source") },
);
const batch = await executor.start("ALTER TABLE users ADD COLUMN active BOOLEAN", targets);
expect(executionOrder).toEqual(["conn-1", "conn-2", "conn-3"]);
expect(batch?.items.map((item) => item.status)).toEqual(["failed", "success", "success"]);
expect(batch?.status).toBe("completed");
});
it("executes all targets concurrently in parallel mode", async () => {
let release!: () => void;
const gate = new Promise<void>((resolve) => {
release = resolve;
});
const started: string[] = [];
let running = 0;
let maxRunning = 0;
const executor = useMultiDbExecution(
{
executeTarget: async ({ target }) => {
started.push(target.connectionId);
running += 1;
maxRunning = Math.max(maxRunning, running);
await gate;
running -= 1;
return { status: "success" as const, durationMs: 12 };
},
},
{ sourceTabId: "source" },
);
const run = executor.start("UPDATE users SET active = TRUE", targets, {}, "parallel");
await Promise.resolve();
expect(started).toEqual(["conn-1", "conn-2", "conn-3"]);
expect(maxRunning).toBe(3);
release();
const batch = await run;
expect(batch?.mode).toBe("parallel");
expect(batch?.items.map((item) => item.status)).toEqual(["success", "success", "success"]);
expect(batch?.items.every((item) => item.durationMs === 12)).toBe(true);
expect(batch?.durationMs).toEqual(expect.any(Number));
});
it("takes a target snapshot before execution starts", async () => {
let release!: () => void;
const gate = new Promise<void>((resolve) => {
release = resolve;
});
const executor = useMultiDbExecution(
{
executeTarget: async () => {
await gate;
return { status: "success" };
},
},
{ sourceTabId: "source" },
);
const mutableTargets = targets.map((target) => ({ ...target }));
const run = executor.start("SELECT 1", mutableTargets);
await Promise.resolve();
mutableTargets[0].database = "changed-after-confirmation";
expect(executor.batch.value?.items[0].target.database).toBe("app");
release();
await run;
});
it("keeps the source offset in the immutable batch context", async () => {
let receivedOffset: number | undefined;
const executor = useMultiDbExecution(
{
executeTarget: async ({ context }) => {
receivedOffset = context.sourceOffset;
return { status: "success" };
},
},
{ sourceTabId: "source" },
);
await executor.start("SELECT 1", [targets[0]], { sourceOffset: 17 });
expect(receivedOffset).toBe(17);
expect(Object.isFrozen(executor.batch.value?.context)).toBe(true);
});
it("cancels the current target and marks remaining targets as not executed", async () => {
let release!: () => void;
const gate = new Promise<void>((resolve) => {
release = resolve;
});
const cancelTarget = vi.fn(async () => {
release();
});
const executor = useMultiDbExecution(
{
executeTarget: async () => {
await gate;
return { status: "cancelled" };
},
cancelTarget,
},
{ sourceTabId: "source" },
);
const run = executor.start("DROP TABLE users", targets);
await Promise.resolve();
await executor.cancel();
await run;
expect(cancelTarget).toHaveBeenCalledWith("source", expect.any(String));
expect(executor.batch.value?.items.map((item) => item.status)).toEqual(["cancelled", "not_executed", "not_executed"]);
expect(executor.batch.value?.status).toBe("cancelled");
});
it("enters cancelling before waiting for the current target to settle", async () => {
let release!: () => void;
const gate = new Promise<void>((resolve) => {
release = resolve;
});
const executor = useMultiDbExecution(
{
executeTarget: async () => {
await gate;
return { status: "cancelled" };
},
cancelTarget: async () => release(),
},
{ sourceTabId: "source" },
);
const run = executor.start("DROP TABLE users", targets);
await Promise.resolve();
const cancel = executor.cancel();
expect(executor.batch.value?.status).toBe("cancelling");
await cancel;
await run;
expect(executor.batch.value?.status).toBe("cancelled");
});
it("does not create a target tab when cancellation arrives during validation", async () => {
let releaseValidation!: () => void;
const validationFinished = new Promise<void>((resolve) => {
releaseValidation = resolve;
});
const executor = useMultiDbExecution(
{
validateTarget: async () => {
await validationFinished;
return { valid: true };
},
executeTarget: async () => ({ status: "success" as const }),
},
{ sourceTabId: "source" },
);
const run = executor.start("ALTER TABLE users ADD COLUMN active BOOLEAN", targets);
await Promise.resolve();
await executor.cancel();
releaseValidation();
await run;
expect(executor.batch.value?.items.map((item) => item.status)).toEqual(["cancelled", "not_executed", "not_executed"]);
});
});
@@ -17,6 +17,7 @@ import { uuid } from "@/lib/common/utils";
import { simpleDataGridOrderByReferencesMissingColumn, type DataGridSortMode } from "@/lib/dataGrid/dataGridSort";
import type { DataGridReloadIntent } from "@/lib/dataGrid/dataGridToolbar";
import { queryResultBaseSql, queryResultExecutionSql } from "@/lib/tabs/tabPresentation";
import { sqlExecutionTargetCapabilities } from "@/lib/database/sqlExecutionTargetCapabilities";
const DATA_TAB_METADATA_TTL_MS = TABLE_METADATA_CACHE_TTL_MS;
@@ -40,6 +41,14 @@ export function useDataGridActions(activeTab: ComputedRef<QueryTab | undefined>)
const queryStore = useQueryStore();
const settingsStore = useSettingsStore();
function activeQueryTargetOptions(tab: QueryTab) {
const executionTarget = queryStore.activeResultExecutionTarget(tab.id);
if (!executionTarget) return {};
const connection = connectionStore.getConfig(executionTarget.connectionId);
const capabilities = sqlExecutionTargetCapabilities(connection);
return capabilities ? { executionTarget, targetContext: capabilities.provider.toExecutionContext(executionTarget, connection!) } : { executionTarget };
}
function quoteIdent(tab: QueryTab, name: string): string {
const config = connectionStore.getConfig(tab.connectionId);
return quoteTableDataIdentifier(effectiveDatabaseTypeForConnection(config), name, connectionStore.connectionIdentifierQuote?.(tab.connectionId));
@@ -211,6 +220,7 @@ export function useDataGridActions(activeTab: ComputedRef<QueryTab | undefined>)
tab.resultSortMode = undefined;
tab.resultSortedSql = undefined;
await queryStore.executeTabSql(tab.id, resultGroupSql, {
...activeQueryTargetOptions(tab),
resultBaseSql: resultGroupSql,
resultSortedSql: undefined,
preserveResultDuringExecution: true,
@@ -222,6 +232,7 @@ export function useDataGridActions(activeTab: ComputedRef<QueryTab | undefined>)
const sortColumns = visibleQuerySortColumns(tab.result?.columns ?? [], tab.result?.hidden_column_indexes, tab.resultSortColumnIndex ?? -1);
const rebuildHiddenKeySort = !!tab.result?.hidden_column_indexes?.length && tab.resultSortMode === "database" && !!tab.resultSortDirection && !!tab.resultSortColumn && !!sortColumns;
await queryStore.executeTabSql(tab.id, rebuildHiddenKeySort ? (tab.resultBaseSql ?? tab.sql) : tab.resultSortedSql, {
...activeQueryTargetOptions(tab),
resultBaseSql: tab.resultBaseSql ?? tab.sql,
...(rebuildHiddenKeySort
? {
@@ -240,6 +251,7 @@ export function useDataGridActions(activeTab: ComputedRef<QueryTab | undefined>)
}
if (sql?.trim()) {
await queryStore.executeTabSql(tab.id, sql, {
...(tab.mode === "query" ? activeQueryTargetOptions(tab) : {}),
resultBaseSql: sql,
resultSortedSql: undefined,
preserveResultDuringExecution: true,
@@ -264,6 +276,7 @@ export function useDataGridActions(activeTab: ComputedRef<QueryTab | undefined>)
const sessionId = tab.result?.has_more && tab.result?.session_id && continuesResultSession ? tab.result.session_id : undefined;
const resultBaseSql = queryResultBaseSql(tab);
await queryStore.executeTabSql(tab.id, baseSql, {
...activeQueryTargetOptions(tab),
resultBaseSql,
resultSortedSql: tab.resultSortedSql,
...(hasDatabaseSort
@@ -276,7 +289,7 @@ export function useDataGridActions(activeTab: ComputedRef<QueryTab | undefined>)
},
}
: {}),
pagination: { offset, limit, sessionId },
pagination: { offset, limit, sessionId, clientSessionId: sessionId ? tab.resultClientSessionId : undefined },
...appendOptions,
preserveResultDuringExecution: true,
preserveTotalRowCountDuringExecution: true,
@@ -331,6 +344,7 @@ export function useDataGridActions(activeTab: ComputedRef<QueryTab | undefined>)
if (!direction) {
await queryStore.executeTabSql(tab.id, baseSql, {
...activeQueryTargetOptions(tab),
resultBaseSql: baseSql,
resultSortedSql: undefined,
preserveResultDuringExecution: true,
@@ -340,7 +354,8 @@ export function useDataGridActions(activeTab: ComputedRef<QueryTab | undefined>)
return;
}
const config = connectionStore.getConfig(tab.connectionId);
const executionTarget = queryStore.activeResultExecutionTarget(tab.id);
const config = connectionStore.getConfig(executionTarget?.connectionId ?? tab.connectionId);
if (effectiveDatabaseTypeForConnection(config) === "mongodb") {
const sortedSql = applyMongoFindSort(baseSql, column, direction);
if (!sortedSql) {
@@ -349,6 +364,7 @@ export function useDataGridActions(activeTab: ComputedRef<QueryTab | undefined>)
}
queryStore.updateSql(tab.id, sortedSql);
await queryStore.executeTabSql(tab.id, sortedSql, {
...activeQueryTargetOptions(tab),
resultBaseSql: baseSql,
resultSortedSql: sortedSql,
preserveResultDuringExecution: true,
@@ -377,6 +393,7 @@ export function useDataGridActions(activeTab: ComputedRef<QueryTab | undefined>)
return;
}
await queryStore.executeTabSql(tab.id, built.sql, {
...activeQueryTargetOptions(tab),
resultBaseSql: baseSql,
resultSortedSql: built.sql,
preserveResultDuringExecution: true,
@@ -386,6 +403,7 @@ export function useDataGridActions(activeTab: ComputedRef<QueryTab | undefined>)
return;
}
await queryStore.executeTabSql(tab.id, baseSql, {
...activeQueryTargetOptions(tab),
resultBaseSql: baseSql,
querySort: {
resultColumns: sortColumns.resultColumns,
@@ -1,7 +1,7 @@
import { ref } from "vue";
import { useConnectionStore } from "@/stores/connectionStore";
import { filterDatabaseNamesForConnection, filterSchemaNamesForConnection } from "@/lib/database/visibleDatabases";
import { isDorisFamilyCatalogCapable } from "@/lib/database/databaseFeatureSupport";
import { isDorisFamilyCatalogCapable, supportsQueryTargetDatabaseListing } from "@/lib/database/databaseFeatureSupport";
import { usesTreeSchemaMode } from "@/lib/database/databaseCapabilities";
import { isInternalDorisCatalog } from "@/lib/database/databaseFeatureSupport";
import type { CatalogInfo, ConnectionConfig } from "@/types/database";
@@ -78,43 +78,54 @@ export function useDatabaseOptions() {
const loadingCatalogOptions = ref<Record<string, boolean>>({});
const catalogDatabaseOptions = ref<Record<string, string[]>>({});
const loadingCatalogDatabaseOptions = ref<Record<string, boolean>>({});
const databaseRequests = new Map<string, Promise<void>>();
const catalogRequests = new Map<string, Promise<CatalogInfo[]>>();
const catalogDatabaseRequests = new Map<string, Promise<string[]>>();
async function loadDatabaseOptions(connectionId: string, catalog?: string) {
async function loadDatabaseOptions(connectionId: string, catalog?: string): Promise<void> {
const connection = connectionStore.getConfig(connectionId);
if (!connection || loadingDatabaseOptions.value[connectionId]) return;
if (!connection) return;
loadingDatabaseOptions.value[connectionId] = true;
try {
await connectionStore.ensureConnected(connectionId);
if (connection.db_type === "redis") {
const dbs = await api.redisListDatabases(connectionId);
databaseOptions.value[connectionId] = databaseOptionsForConnection(
dbs.map((db) => String(db.db)),
connection,
);
} else if (connection.db_type === "mongodb") {
databaseOptions.value[connectionId] = filterDatabaseNamesForConnection(await api.mongoListDatabases(connectionId), connection);
} else if (connection.db_type === "dameng") {
// 达梦的"数据库"概念对应 schema,使用 listSchemas 获取 schema 列表
databaseOptions.value[connectionId] = await fetchNamespaceOptionsForConnection(connectionId, connection);
} else if (catalog && isDorisFamilyCatalogCapable(connection?.db_type, connection?.driver_profile)) {
const dbs = await api.listDorisCatalogDatabases(connectionId, catalog);
databaseOptions.value[connectionId] = databaseOptionsForConnection(
dbs.map((db) => db.name),
connection,
);
} else {
const dbs = await api.listDatabases(connectionId);
databaseOptions.value[connectionId] = databaseOptionsForConnection(
dbs.map((db) => db.name),
connection,
);
const requestKey = `${connectionId}:${catalog ?? ""}`;
const pending = databaseRequests.get(requestKey);
if (pending) return pending;
const request = (async () => {
loadingDatabaseOptions.value[connectionId] = true;
try {
await connectionStore.ensureConnected(connectionId);
if (connection.db_type === "redis") {
const dbs = await api.redisListDatabases(connectionId);
databaseOptions.value[connectionId] = databaseOptionsForConnection(
dbs.map((db) => String(db.db)),
connection,
);
} else if (connection.db_type === "mongodb") {
databaseOptions.value[connectionId] = filterDatabaseNamesForConnection(await api.mongoListDatabases(connectionId), connection);
} else if (connection.db_type === "dameng") {
databaseOptions.value[connectionId] = await fetchNamespaceOptionsForConnection(connectionId, connection);
} else if (supportsQueryTargetDatabaseListing(connection.db_type)) {
databaseOptions.value[connectionId] = filterDatabaseNamesForConnection(await api.documentListDatabases(connectionId), connection);
} else if (catalog && isDorisFamilyCatalogCapable(connection?.db_type, connection?.driver_profile)) {
const dbs = await api.listDorisCatalogDatabases(connectionId, catalog);
databaseOptions.value[connectionId] = databaseOptionsForConnection(
dbs.map((db) => db.name),
connection,
);
} else {
const dbs = await api.listDatabases(connectionId);
databaseOptions.value[connectionId] = databaseOptionsForConnection(
dbs.map((db) => db.name),
connection,
);
}
} finally {
loadingDatabaseOptions.value[connectionId] = false;
databaseRequests.delete(requestKey);
}
} finally {
loadingDatabaseOptions.value[connectionId] = false;
}
})();
databaseRequests.set(requestKey, request);
return request;
}
async function loadCatalogOptions(connectionId: string): Promise<CatalogInfo[]> {
@@ -2,7 +2,7 @@ import { reactive, computed } from "vue";
import * as api from "@/lib/backend/api";
import { isTerminalTransferProgress } from "@/lib/backend/transferProgress";
export type BackgroundTaskKind = "table-export" | "database-export" | "sql-file" | "data-transfer";
export type BackgroundTaskKind = "table-export" | "database-export" | "sql-file" | "data-transfer" | "multi-db-execution";
export type BackgroundTaskStatus = "Running" | "Writing" | "Done" | "Error" | "Cancelled";
export type DatabaseExportSource = "manual" | "scheduled";
@@ -46,6 +46,31 @@ export interface ExportTask {
targetTables?: string[];
transferFailures?: DataTransferFailure[];
transferFailuresOmitted?: number;
multiDbSourceTabId?: string;
multiDbTotal?: number;
multiDbCompleted?: number;
multiDbSuccessCount?: number;
multiDbFailureCount?: number;
multiDbSkippedCount?: number;
multiDbNotExecutedCount?: number;
currentTarget?: { connectionId: string; catalog?: string; database: string; schema?: string };
onOpen?: () => void;
}
export interface MultiDbExecutionTaskProgress {
sourceTabId: string;
total: number;
completed: number;
successCount: number;
failureCount: number;
skippedCount: number;
notExecutedCount: number;
status: "running" | "completed" | "cancelled";
startedAt: number;
finishedAt?: number;
elapsedMs?: number;
currentTarget?: { connectionId: string; catalog?: string; database: string; schema?: string };
errorMessage?: string;
}
export const MAX_TRANSFER_FAILURE_DETAILS = 100;
@@ -325,6 +350,48 @@ export function useExportTracker() {
return task;
}
function addMultiDbExecutionTask(batchId: string, label: string, sourceTabId: string, onOpen?: () => void): ExportTask {
const task = reactive<ExportTask>({
exportId: batchId,
kind: "multi-db-execution",
tableName: label,
format: "sql",
filePath: "",
rowsExported: 0,
totalRows: null,
status: "Running",
errorMessage: null,
multiDbSourceTabId: sourceTabId,
onOpen,
startedAt: Date.now(),
});
taskMap.set(batchId, task);
return task;
}
function updateMultiDbExecutionTask(batchId: string, progress: MultiDbExecutionTaskProgress): void {
const task = taskMap.get(batchId);
if (!task) return;
task.multiDbSourceTabId = progress.sourceTabId;
task.multiDbTotal = progress.total;
task.multiDbCompleted = progress.completed;
task.multiDbSuccessCount = progress.successCount;
task.multiDbFailureCount = progress.failureCount;
task.multiDbSkippedCount = progress.skippedCount;
task.multiDbNotExecutedCount = progress.notExecutedCount;
task.rowsExported = progress.completed;
task.totalRows = progress.total;
task.currentTarget = progress.currentTarget;
task.errorMessage = progress.errorMessage ?? null;
task.startedAt = progress.startedAt;
task.finishedAt = progress.finishedAt;
task.elapsedMs = progress.elapsedMs;
if (progress.status === "running") task.status = "Running";
else if (progress.status === "cancelled") task.status = "Cancelled";
else task.status = progress.failureCount > 0 ? "Error" : "Done";
if (task.status === "Done" || task.status === "Error" || task.status === "Cancelled") finishExportTask(task);
}
function startDataTransferTask(
request: api.TransferRequest,
label: string,
@@ -505,6 +572,8 @@ export function useExportTracker() {
addDatabaseExportTask,
addSqlFileTask,
addDataTransferTask,
addMultiDbExecutionTask,
updateMultiDbExecutionTask,
startDataTransferTask,
updateTableExportTask,
updateDatabaseExportTask,
@@ -0,0 +1,215 @@
import { computed, reactive, ref, type ComputedRef, type Ref } from "vue";
import type { MultiDbExecutionTarget, MultiDbExecutionItemStatus, MultiDbTargetExecutionResult } from "@/types/sqlExecution";
export interface MultiDbExecutionItem {
id: string;
target: Readonly<MultiDbExecutionTarget>;
status: MultiDbExecutionItemStatus;
errorMessage?: string;
startedAt?: number;
completedAt?: number;
durationMs?: number;
}
export interface MultiDbExecutionBatch {
id: string;
sourceTabId: string;
sql: string;
items: MultiDbExecutionItem[];
status: "running" | "cancelling" | "completed" | "cancelled";
cancelRequested: boolean;
context: MultiDbExecutionContext;
mode: "serial" | "parallel";
startedAt: number;
completedAt?: number;
durationMs?: number;
}
export interface MultiDbExecutionAdapter {
executeTarget: (input: { target: MultiDbExecutionTarget; sourceTabId: string; sql: string; scopeId: string; context: Readonly<MultiDbExecutionContext>; isCancellationRequested: () => boolean }) => Promise<MultiDbTargetExecutionResult>;
validateTarget?: (target: MultiDbExecutionTarget) => Promise<{ valid: boolean; errorMessage?: string }>;
cancelTarget?: (sourceTabId: string, scopeId: string) => Promise<void>;
cancelPending?: (scopeId: string) => void | Promise<void>;
}
export interface MultiDbExecutionContext {
readonly batchId: string;
readonly sourceTabId: string;
readonly sql: string;
readonly targets: readonly MultiDbExecutionTarget[];
/** Offset of the submitted SQL in the source editor, captured at confirmation time. */
readonly sourceOffset?: number;
}
export type MultiDbExecutionContextOverrides = Pick<MultiDbExecutionContext, "sourceOffset">;
export interface MultiDbExecutionOptions {
sourceTabId: Ref<string> | ComputedRef<string> | string;
}
export type MultiDbExecutionMode = "serial" | "parallel";
function executionId(): string {
return `multi-${Date.now()}-${Math.random().toString(36).slice(2, 10)}`;
}
function normalizeError(error: unknown): string {
if (error instanceof Error) return error.message;
return typeof error === "string" ? error : String(error);
}
export function useMultiDbExecution(adapter: MultiDbExecutionAdapter, options: MultiDbExecutionOptions) {
const batch = ref<MultiDbExecutionBatch>();
const isRunning = computed(() => batch.value?.status === "running" || batch.value?.status === "cancelling");
function sourceTabId(): string {
return typeof options.sourceTabId === "string" ? options.sourceTabId : options.sourceTabId.value;
}
function markPendingNotExecuted(): void {
const current = batch.value;
if (!current) return;
for (const item of current.items) {
if (item.status === "pending") {
item.status = "not_executed";
item.completedAt ??= Date.now();
item.durationMs ??= 0;
}
}
}
async function executeItem(current: MultiDbExecutionBatch, item: MultiDbExecutionItem): Promise<void> {
if (current.cancelRequested) {
markPendingNotExecuted();
return;
}
item.status = "running";
item.startedAt = Date.now();
try {
if (adapter.validateTarget) {
const validation = await adapter.validateTarget(item.target);
if (!validation.valid) {
item.status = current.cancelRequested ? "cancelled" : "failed";
item.errorMessage = validation.errorMessage;
item.completedAt = Date.now();
item.durationMs = item.completedAt - (item.startedAt ?? item.completedAt);
return;
}
}
if (current.cancelRequested) {
item.status = "cancelled";
markPendingNotExecuted();
item.completedAt = Date.now();
item.durationMs = item.completedAt - (item.startedAt ?? item.completedAt);
return;
}
const result = await adapter.executeTarget({
target: item.target,
sourceTabId: current.sourceTabId,
sql: current.sql,
scopeId: current.id,
context: current.context,
isCancellationRequested: () => current.cancelRequested,
});
item.status = current.cancelRequested && result.status === "failed" ? "cancelled" : result.status;
item.errorMessage = result.errorMessage;
item.durationMs = result.durationMs;
} catch (error) {
// One target is intentionally isolated from the queue. Adapter errors
// become target failures so later targets keep running.
item.status = current.cancelRequested ? "cancelled" : "failed";
item.errorMessage = normalizeError(error);
} finally {
item.completedAt = Date.now();
item.durationMs ??= item.completedAt - (item.startedAt ?? item.completedAt);
}
}
async function executeBatch(current: MultiDbExecutionBatch): Promise<void> {
if (current.mode === "parallel") {
await Promise.all(current.items.map((item) => executeItem(current, item)));
} else {
for (const item of current.items) {
await executeItem(current, item);
if (current.cancelRequested) break;
}
}
if (current.cancelRequested) markPendingNotExecuted();
current.status = current.cancelRequested ? "cancelled" : "completed";
current.completedAt = Date.now();
current.durationMs = current.completedAt - current.startedAt;
}
async function start(sql: string, targets: readonly MultiDbExecutionTarget[], context: MultiDbExecutionContextOverrides = {}, mode: MultiDbExecutionMode = "serial"): Promise<MultiDbExecutionBatch | undefined> {
if (isRunning.value || !sql.trim() || targets.length === 0) return undefined;
const sourceId = sourceTabId();
const id = executionId();
const targetSnapshot = targets.map((target) => Object.freeze({ ...target }));
const current = reactive<MultiDbExecutionBatch>({
id,
sourceTabId: sourceId,
sql,
items: targetSnapshot.map((target, index) => ({
id: `${index}-${executionId()}`,
target: Object.freeze({ ...target }),
status: "pending",
})),
status: "running",
cancelRequested: false,
mode,
context: Object.freeze({
batchId: id,
sourceTabId: sourceId,
sql,
targets: Object.freeze(targetSnapshot),
...context,
}),
startedAt: Date.now(),
});
batch.value = current;
await executeBatch(current);
return current;
}
async function cancel(): Promise<void> {
const current = batch.value;
if (!current || current.status !== "running") return;
current.cancelRequested = true;
current.status = "cancelling";
try {
await adapter.cancelPending?.(current.id);
} catch {
// Cancellation is best-effort. Continue to the active target even when
// a confirmation store has already settled concurrently.
}
const running = current.items.find((item) => item.status === "running");
if (running && adapter.cancelTarget) {
try {
await adapter.cancelTarget(current.sourceTabId, current.id);
} catch {
// The target execution will settle the batch when its request returns.
}
}
if (!running) {
markPendingNotExecuted();
current.status = "cancelled";
current.completedAt = Date.now();
current.durationMs = current.completedAt - current.startedAt;
}
}
function reset(): void {
if (isRunning.value) return;
batch.value = undefined;
}
return {
batch,
isRunning,
start,
cancel,
reset,
};
}
@@ -0,0 +1,336 @@
import { computed, ref, type ComputedRef, type Ref } from "vue";
import { useConnectionStore } from "@/stores/connectionStore";
import { useDatabaseOptions, catalogDatabaseOptionsKey } from "@/composables/useDatabaseOptions";
import { schemaOptionsCacheKey, useSchemaOptions } from "@/composables/useSchemaOptions";
import { isInternalDorisCatalog } from "@/lib/database/databaseFeatureSupport";
import { effectiveDatabaseTypeForConnection } from "@/lib/database/jdbcDialect";
import { supportsQueryExecution } from "@/lib/database/databaseFeatureSupport";
import { sqlExecutionTargetCapabilities, targetCanUseDefaultWhenDatabaseListEmpty, targetDefaultDatabase, targetIsSingleDatabase, targetSupportsCatalog, targetSupportsSchema, targetUsesConnectionOnlyScope } from "@/lib/database/sqlExecutionTargetCapabilities";
import type { CatalogInfo, ConnectionConfig, DatabaseType } from "@/types/database";
import type { MultiDbExecutionTarget, SqlExecutionTargetValidation } from "@/types/sqlExecution";
export type SqlExecutionTargetValidationReason = "targetMissingConnection" | "targetTypeMismatch" | "targetCatalogMissing" | "targetDatabaseMissing" | "targetSchemaMissing" | "targetPermissionDenied" | "targetValidationFailed";
export interface MultiDbTargetCatalogOption {
name: string;
targetCatalog?: string;
isInternal: boolean;
}
export interface MultiDbTargetResourceState {
status: "idle" | "loading" | "loaded" | "error";
loading: boolean;
loaded: boolean;
error?: string;
}
type DatabaseTypeSource = Ref<DatabaseType | undefined> | ComputedRef<DatabaseType | undefined> | DatabaseType | undefined;
function sourceValue(source: DatabaseTypeSource): DatabaseType | undefined {
return typeof source === "string" || source === undefined ? source : source.value;
}
function errorMessage(error: unknown): string {
return error instanceof Error ? error.message : String(error);
}
function catalogTargetValue(catalog: CatalogInfo): MultiDbTargetCatalogOption {
const isInternal = isInternalDorisCatalog(catalog.catalog_type, catalog.name);
return {
name: catalog.name,
targetCatalog: isInternal ? undefined : catalog.name,
isInternal,
};
}
export function useMultiDbTargetSelection(databaseTypeSource: DatabaseTypeSource) {
const connectionStore = useConnectionStore();
const databaseOptionsApi = useDatabaseOptions();
const schemaOptionsApi = useSchemaOptions();
const resourceStates = ref<Record<string, MultiDbTargetResourceState>>({});
const pendingLoads = new Map<string, Promise<void>>();
const databaseType = computed(() => sourceValue(databaseTypeSource));
const compatibleConnections = computed(() =>
connectionStore.connections
.filter((connection) => {
const effectiveType = effectiveDatabaseTypeForConnection(connection);
return effectiveType === databaseType.value && supportsQueryExecution(connection.db_type);
})
.sort((a, b) => (a.name || a.id).localeCompare(b.name || b.id, undefined, { numeric: true, sensitivity: "base" })),
);
function stateFor(key: string): MultiDbTargetResourceState {
return resourceStates.value[key] ?? { status: "idle", loading: false, loaded: false };
}
function setState(key: string, state: MultiDbTargetResourceState): void {
resourceStates.value[key] = state;
}
async function loadOnce(key: string, loader: () => Promise<void>): Promise<void> {
const existing = pendingLoads.get(key);
if (existing) return existing;
const request = (async () => {
setState(key, { status: "loading", loading: true, loaded: false });
try {
await loader();
setState(key, { status: "loaded", loading: false, loaded: true });
} catch (error) {
setState(key, { status: "error", loading: false, loaded: false, error: errorMessage(error) });
throw error;
} finally {
pendingLoads.delete(key);
}
})();
pendingLoads.set(key, request);
return request;
}
function connection(connectionId: string): ConnectionConfig | undefined {
return connectionStore.getConfig(connectionId);
}
function isCatalogCapable(connectionId: string): boolean {
return targetSupportsCatalog(connection(connectionId));
}
function catalogsForConnection(connectionId: string): MultiDbTargetCatalogOption[] {
return (databaseOptionsApi.catalogOptions.value[connectionId] ?? []).map(catalogTargetValue);
}
function catalogOption(connectionId: string, targetCatalog?: string): MultiDbTargetCatalogOption | undefined {
return catalogsForConnection(connectionId).find((catalog) => catalog.targetCatalog === targetCatalog);
}
function databaseCacheKey(connectionId: string, targetCatalog?: string): string {
return targetCatalog ? catalogDatabaseOptionsKey(connectionId, targetCatalog) : connectionId;
}
function databaseNamesForConnection(connectionId: string, targetCatalog?: string): string[] {
const config = connection(connectionId);
if (config && targetUsesConnectionOnlyScope(config)) return [];
const cached = targetCatalog ? databaseOptionsApi.catalogDatabaseOptions.value[databaseCacheKey(connectionId, targetCatalog)] : databaseOptionsApi.databaseOptions.value[connectionId];
const cache = targetCatalog ? databaseOptionsApi.catalogDatabaseOptions.value : databaseOptionsApi.databaseOptions.value;
const cacheKey = databaseCacheKey(connectionId, targetCatalog);
if (stateFor(`databases:${cacheKey}`).status === "error") return [];
const hasLoadedMetadata = Object.prototype.hasOwnProperty.call(cache, cacheKey);
const loadedNames = cached ?? [];
if (loadedNames.length > 0) return loadedNames;
if (config && targetCanUseDefaultWhenDatabaseListEmpty(config)) {
const defaultDatabase = targetDefaultDatabase(config);
if (defaultDatabase) return [defaultDatabase];
}
// An empty loaded response is authoritative. Returning no synthetic
// database here lets the selector expose an explicit connection fallback
// only for capabilities that registered one, without turning an empty
// metadata response into an arbitrary database list.
if (hasLoadedMetadata) return [];
return [];
}
function schemaDatabaseName(target: MultiDbExecutionTarget, config: ConnectionConfig): string {
if (target.database) return target.database;
if (targetIsSingleDatabase(config)) return "_";
return "";
}
function schemaNamesForTarget(target: MultiDbExecutionTarget): string[] {
const config = connection(target.connectionId);
if (!config || !targetSupportsSchema(config)) return [];
const database = schemaDatabaseName(target, config);
if (stateFor(`schemas:${schemaOptionsCacheKey(target.connectionId, database, config.show_system_schemas === true)}`).status === "error") return [];
return schemaOptionsApi.getSchemaOptionsForDb(target.connectionId, database);
}
async function loadConnection(connectionId: string): Promise<void> {
const config = connection(connectionId);
if (!config) return;
if (targetUsesConnectionOnlyScope(config)) return;
if (isCatalogCapable(connectionId)) {
await loadCatalogs(connectionId);
const catalogs = catalogsForConnection(connectionId);
const internalCatalog = catalogs.find((catalog) => catalog.isInternal);
await loadDatabase(connectionId, internalCatalog?.targetCatalog);
return;
}
await loadDatabase(connectionId);
}
async function loadCatalogs(connectionId: string): Promise<void> {
if (!isCatalogCapable(connectionId)) return;
const config = connection(connectionId);
const provider = sqlExecutionTargetCapabilities(config)?.provider;
if (!config || !provider) return;
await loadOnce(`catalogs:${connectionId}`, async () => {
await provider.listTargets(
config,
{ level: "catalog" },
{
listCatalogs: async () => {
await databaseOptionsApi.loadCatalogOptions(connectionId);
return catalogsForConnection(connectionId).map((catalog) => catalog.name);
},
},
);
});
}
async function loadDatabase(connectionId: string, targetCatalog?: string): Promise<void> {
const config = connection(connectionId);
if (!config) return;
if (targetUsesConnectionOnlyScope(config)) return;
const provider = sqlExecutionTargetCapabilities(config)?.provider;
if (!provider) return;
const key = databaseCacheKey(connectionId, targetCatalog);
await loadOnce(`databases:${key}`, async () => {
await provider.listTargets(
config,
{ level: "database", catalog: targetCatalog },
{
listDatabases: async (catalog) => {
if (catalog) await databaseOptionsApi.loadCatalogDatabaseOptions(connectionId, catalog);
else await databaseOptionsApi.loadDatabaseOptions(connectionId);
return databaseNamesForConnection(connectionId, catalog);
},
},
);
});
}
async function loadSchemas(target: MultiDbExecutionTarget): Promise<void> {
const config = connection(target.connectionId);
if (!config || !targetSupportsSchema(config)) return;
const provider = sqlExecutionTargetCapabilities(config)?.provider;
if (!provider) return;
const database = schemaDatabaseName(target, config);
const key = schemaOptionsCacheKey(target.connectionId, database, config.show_system_schemas === true);
await loadOnce(`schemas:${key}`, async () => {
await provider.listTargets(
config,
{ level: "schema", database },
{
listSchemas: async (requestedDatabase) => {
await schemaOptionsApi.loadSchemaOptions(target.connectionId, requestedDatabase);
return schemaOptionsApi.getSchemaOptionsForDb(target.connectionId, requestedDatabase);
},
},
);
});
}
function resourceState(key: string): MultiDbTargetResourceState {
return stateFor(key);
}
function databaseState(connectionId: string, targetCatalog?: string): MultiDbTargetResourceState {
return resourceState(`databases:${databaseCacheKey(connectionId, targetCatalog)}`);
}
function catalogState(connectionId: string): MultiDbTargetResourceState {
return resourceState(`catalogs:${connectionId}`);
}
function schemaState(target: MultiDbExecutionTarget): MultiDbTargetResourceState {
const config = connection(target.connectionId);
if (!config) return { status: "idle", loading: false, loaded: false };
const database = schemaDatabaseName(target, config);
return resourceState(`schemas:${schemaOptionsCacheKey(target.connectionId, database, config.show_system_schemas === true)}`);
}
async function validateTarget(target: MultiDbExecutionTarget): Promise<SqlExecutionTargetValidation> {
const config = connection(target.connectionId);
if (!config) return { target, state: "invalid", reason: "targetMissingConnection" };
if (!databaseType.value) return { target, state: "invalid", reason: "targetValidationFailed" };
if (effectiveDatabaseTypeForConnection(config) !== databaseType.value) return { target, state: "invalid", reason: "targetTypeMismatch" };
const capabilities = sqlExecutionTargetCapabilities(config);
if (!capabilities) return { target, state: "invalid", reason: "targetValidationFailed" };
try {
await capabilities.provider.validateTarget(target, config);
} catch {
return { target, state: "invalid", reason: "targetValidationFailed" };
}
if (capabilities.scope === "connection") {
if (target.database.trim() || target.catalog !== undefined || target.schema !== undefined) return { target, state: "invalid", reason: "targetDatabaseMissing" };
return { target: { ...target, database: "" }, state: "valid" };
}
try {
if (isCatalogCapable(target.connectionId)) {
await loadCatalogs(target.connectionId);
// An omitted catalog is the normalized representation of the
// driver's internal catalog. It must still be present in the loaded
// catalog metadata; otherwise an empty/partial catalog response must
// not silently validate a target against a fabricated default.
if (!catalogOption(target.connectionId, target.catalog)) {
return { target, state: "invalid", reason: "targetCatalogMissing" };
}
} else if (target.catalog !== undefined) {
return { target, state: "invalid", reason: "targetCatalogMissing" };
}
await loadDatabase(target.connectionId, target.catalog);
if (databaseState(target.connectionId, target.catalog).status !== "loaded") {
return { target, state: "invalid", reason: "targetValidationFailed" };
}
const databases = databaseNamesForConnection(target.connectionId, target.catalog);
if (target.database) {
if (databases.length === 0) {
if (!capabilities.allowDefaultWhenDatabaseListEmpty || target.database !== capabilities.defaultDatabase) {
return { target, state: "invalid", reason: "targetDatabaseMissing" };
}
} else if (!databases.includes(target.database)) {
return { target, state: "invalid", reason: "targetDatabaseMissing" };
}
} else if (capabilities.databaseRequired || !capabilities.allowsEmptyDatabaseTarget) {
return { target, state: "invalid", reason: "targetDatabaseMissing" };
}
if (target.schema !== undefined) {
if (!capabilities.supportsSchema) return { target, state: "invalid", reason: "targetSchemaMissing" };
await loadSchemas(target);
if (schemaState(target).status !== "loaded") {
return { target, state: "invalid", reason: "targetValidationFailed" };
}
const schemas = schemaNamesForTarget(target);
if (schemas.length === 0 || !schemas.includes(target.schema)) return { target, state: "invalid", reason: "targetSchemaMissing" };
}
// Keep the provider as the final authority for the context that will be
// consumed by execution. This also prevents a future namespace provider
// from being considered valid merely because a list item was present.
capabilities.provider.toExecutionContext(target, config);
return { target, state: "valid" };
} catch (error) {
return { target, state: "invalid", reason: error instanceof Error && /permission|denied|forbidden|unauthorized/i.test(error.message) ? "targetPermissionDenied" : "targetValidationFailed" };
}
}
async function validateTargets(targets: readonly MultiDbExecutionTarget[]): Promise<SqlExecutionTargetValidation[]> {
// Target validation is independent. Keep the shared loadOnce/pendingLoads
// deduplication semantics, but allow different connections/databases to
// load metadata concurrently instead of making the dialog wait for a
// full serial pass.
return Promise.all(targets.map((target) => validateTarget(target)));
}
return {
databaseType,
compatibleConnections,
resourceStates,
connection,
isCatalogCapable,
catalogsForConnection,
databaseNamesForConnection,
schemaNamesForTarget,
loadConnection,
loadCatalogs,
loadDatabase,
loadSchemas,
resourceState,
catalogState,
databaseState,
schemaState,
validateTarget,
validateTargets,
};
}
@@ -24,6 +24,7 @@ export function useSchemaOptions() {
const schemaOptions = ref<Record<string, string[]>>({});
const loadingSchemaOptions = ref<Record<string, boolean>>({});
const schemaRequests = new Map<string, Promise<void>>();
function cacheKey(connectionId: string, database: string) {
return schemaOptionsCacheKey(connectionId, database, connectionStore.getConfig(connectionId)?.show_system_schemas === true);
@@ -40,18 +41,24 @@ export function useSchemaOptions() {
if (!database && !isSingleDatabase(dbType) && !usesTreeSchemaMode(dbType)) return;
const key = cacheKey(connectionId, database);
if (hasSchemaOptionsCacheEntry(schemaOptions.value, key)) return;
if (loadingSchemaOptions.value[key]) return;
const pending = schemaRequests.get(key);
if (pending) return pending;
loadingSchemaOptions.value[key] = true;
try {
await connectionStore.ensureConnected(connectionId);
schemaOptions.value[key] = schemaOptionsForConnection(await api.listSchemas(connectionId, database), connection, database);
} catch (e) {
schemaOptions.value[key] = [];
throw e;
} finally {
loadingSchemaOptions.value[key] = false;
}
const request = (async () => {
loadingSchemaOptions.value[key] = true;
try {
await connectionStore.ensureConnected(connectionId);
schemaOptions.value[key] = schemaOptionsForConnection(await api.listSchemas(connectionId, database), connection, database);
} catch (e) {
schemaOptions.value[key] = [];
throw e;
} finally {
loadingSchemaOptions.value[key] = false;
schemaRequests.delete(key);
}
})();
schemaRequests.set(key, request);
return request;
}
function getSchemaOptionsForDb(connectionId: string, database: string): string[] {
+202 -5
View File
@@ -6,6 +6,7 @@ import { useConnectionStore } from "@/stores/connectionStore";
import { useSettingsStore } from "@/stores/settingsStore";
import { useToast } from "@/composables/useToast";
import { isSingleDatabase, usesTreeSchemaMode } from "@/lib/database/databaseCapabilities";
import { supportsConnectionScopedQueryExecution } from "@/lib/database/databaseFeatureSupport";
import { supportsConnectionLevelSqlExecution } from "@/lib/connection/connectionLevelDatabaseBootstrap";
import { classifySqlActivityKind } from "@/lib/history/historyActivityKind";
import { sqlMetadataRefreshTarget } from "@/lib/sql/sqlMetadataRefresh";
@@ -19,7 +20,11 @@ import { expandSqlVariables } from "@/lib/sql/sqlVariables";
import { enabledSqlParameterSyntaxes, resolveSqlVariableSyntaxToggles } from "@/lib/sql/sqlVariableSyntax";
import { assessProductionSql } from "@/lib/database/productionSafety";
import { useProductionSafetyStore } from "@/stores/productionSafetyStore";
import type { SqlExecutionDangerRequest } from "@/stores/sqlExecutionDangerStore";
import type { ConnectionConfig, DatabaseType, QueryTab } from "@/types/database";
import type { MultiDbExecutionTarget, MultiDbResultRunExecution, MultiDbTargetExecutionResult } from "@/types/sqlExecution";
import { effectiveDatabaseTypeForConnection } from "@/lib/database/jdbcDialect";
import type { SqlExecutionTargetContext } from "@/lib/database/sqlExecutionTargetRegistry";
const DANGER_RE = /^\s*(DROP|DELETE|TRUNCATE|ALTER|UPDATE|MERGE|REPLACE)\b/i;
@@ -27,6 +32,24 @@ interface SqlExecutionOptions {
openInNewResultTab?: boolean;
}
interface TargetSqlExecutionInput {
tab: QueryTab;
connection: ConnectionConfig;
sql: string;
executionTarget?: MultiDbExecutionTarget;
resultRun?: {
batchId: string;
title: string;
target: MultiDbExecutionTarget;
};
sourceOffset?: number;
blockDangerousRedisCommands?: boolean;
targetLabel?: string;
scopeId?: string;
isCancellationRequested?: () => boolean;
targetContext?: SqlExecutionTargetContext;
}
export function stripSqlComments(sql: string): string {
return sql
.replace(/\/\*[\s\S]*?\*\//g, " ")
@@ -80,6 +103,7 @@ export function useSqlExecution(deps: {
activeOutputView: Ref<"result" | "summary" | "explain" | "chart" | "messages">;
blockDangerousRedisCommands?: Ref<boolean>;
onMissingDatabase?: () => void;
requestDangerConfirmation?: (request: SqlExecutionDangerRequest) => Promise<boolean>;
}) {
const { t } = useI18n();
const queryStore = useQueryStore();
@@ -103,6 +127,7 @@ export function useSqlExecution(deps: {
const pendingDangerKind = ref<"sql" | "redis">("sql");
const pendingDangerSourceOffset = ref<number | undefined>();
const pendingOpenInNewResultTab = ref(false);
let pendingSqlParameterContinuation: ((sql: string, sourceOffset?: number) => Promise<void> | void) | undefined;
async function resolvedExecutableSql(source?: SqlExecutionOverride): Promise<{ sql: string; sourceOffset?: number }> {
const atSetEnabled = resolveSqlVariableSyntaxToggles(settingsStore.editorSettings.sqlVariableSyntaxOverrides, deps.activeConnection.value?.db_type).atSet;
@@ -192,7 +217,7 @@ export function useSqlExecution(deps: {
}
}
function prepareSqlParameterDialog(sql: string, sourceOffset?: number, options: SqlExecutionOptions = {}): boolean {
function prepareSqlParameterDialog(sql: string, sourceOffset?: number, options: SqlExecutionOptions = {}, continuation?: (sql: string, sourceOffset?: number) => Promise<void> | void): boolean {
const databaseType = deps.activeConnection.value?.db_type;
const toggles = resolveSqlVariableSyntaxToggles(settingsStore.editorSettings.sqlVariableSyntaxOverrides, databaseType);
const enabledSyntaxes = enabledSqlParameterSyntaxes(toggles);
@@ -204,10 +229,20 @@ export function useSqlExecution(deps: {
sqlParameterEnabledSyntaxes.value = enabledSyntaxes;
pendingSourceOffset.value = sourceOffset;
pendingOpenInNewResultTab.value = options.openInNewResultTab === true;
pendingSqlParameterContinuation = continuation;
showSqlParameterDialog.value = true;
return true;
}
async function prepareMultiExecute(onReady: (sql: string, sourceOffset?: number) => Promise<void> | void): Promise<boolean> {
const tab = deps.activeTab.value;
const { sql, sourceOffset } = await resolvedExecutableSql();
if (!tab || !sql.trim()) return false;
if (supportsSqlTemplateParameters(deps.activeConnection.value, sql) && prepareSqlParameterDialog(sql, sourceOffset, {}, onReady)) return true;
await onReady(sql, sourceOffset);
return false;
}
async function doExecute(sql?: string, sourceOffset?: number, options: SqlExecutionOptions = {}) {
if (sql === undefined) ({ sql, sourceOffset } = await resolvedExecutableSql());
const tab = deps.activeTab.value;
@@ -263,6 +298,161 @@ export function useSqlExecution(deps: {
}
}
async function executeTargetSql(input: TargetSqlExecutionInput): Promise<MultiDbTargetExecutionResult> {
const { tab, connection, sql, sourceOffset, targetLabel } = input;
const startedAt = Date.now();
const executionTab = input.executionTarget
? {
...tab,
connectionId: input.executionTarget.connectionId,
catalog: input.executionTarget.catalog,
database: input.executionTarget.database,
schema: input.executionTarget.schema,
}
: tab;
const finish = (result: MultiDbTargetExecutionResult): MultiDbTargetExecutionResult => ({
...result,
durationMs: Date.now() - startedAt,
});
const cancelRequested = () => input.isCancellationRequested?.() === true;
const tabCancelRequested = (count: number) => (tab.cancelRequestCount ?? 0) !== count;
if (cancelRequested()) return finish({ status: "cancelled" });
if (!sql.trim()) return finish({ status: "failed", errorMessage: t("explain.emptySql") });
if (requiresDatabaseSelection(executionTab, connection, sql)) {
return finish({ status: "failed", errorMessage: t("editor.selectDatabaseRequired") });
}
const blockRedisCommands = input.blockDangerousRedisCommands !== false;
if (connection.db_type === "redis" && blockRedisCommands) {
const commands = sql
.split("\n")
.map((line) => line.trim())
.filter((line) => line.length > 0);
let highestSafety: "allowed" | "confirm" | "blocked" = "allowed";
for (const command of commands) {
const safety = classifyRedisCommandSafety(command);
if (safety === "blocked") {
highestSafety = "blocked";
break;
}
if (safety === "confirm") highestSafety = "confirm";
}
if (highestSafety === "blocked") {
return finish({ status: "skipped", errorMessage: t("redis.blockedCommand", { command: "Redis" }) });
}
if (highestSafety === "confirm") {
const confirmed = await deps.requestDangerConfirmation?.({
sql,
kind: "redis",
connectionName: connection.name,
database: executionTab.database,
targetLabel,
databaseType: connection.db_type,
scopeId: input.scopeId,
});
if (cancelRequested()) return finish({ status: "cancelled" });
if (!confirmed) return finish({ status: "skipped", errorMessage: t("dangerDialog.cancel") });
}
}
const productionAssessment = assessProductionSql(sql, connection, executionTab.database);
if (productionAssessment.active && productionAssessment.isMutation) {
const confirmed = await productionSafetyStore.requestConfirmation({
sql,
connectionName: connection.name,
database: executionTab.database,
productionDatabases: productionAssessment.databases,
source: t("production.sourceMultiDbSql"),
scopeId: input.scopeId,
});
if (cancelRequested()) return finish({ status: "cancelled" });
if (!confirmed) return finish({ status: "skipped", errorMessage: t("dangerDialog.cancel") });
}
if (isDangerousSql(sql, connection.db_type) && settingsStore.editorSettings.confirmDangerousSqlExecution) {
const confirmed = await deps.requestDangerConfirmation?.({
sql,
kind: "sql",
connectionName: connection.name,
database: executionTab.database,
targetLabel,
databaseType: connection.db_type,
scopeId: input.scopeId,
});
if (cancelRequested()) return finish({ status: "cancelled" });
if (!confirmed) return finish({ status: "skipped", errorMessage: t("dangerDialog.cancel") });
}
if (cancelRequested()) return finish({ status: "cancelled" });
const cancelRequestCount = tab.cancelRequestCount ?? 0;
const workerId = input.executionTarget ? queryStore.createMultiDbExecutionWorker(tab.id, input.executionTarget, input.scopeId ?? "") : undefined;
if (input.executionTarget && !workerId) return finish({ status: "failed", errorMessage: t("multiDbExecute.targetMissingConnection") });
const executionTabId = workerId ?? tab.id;
const captureWorkerResult = (status: MultiDbResultRunExecution["status"], errorMessage?: string): string | undefined => {
if (!workerId || !input.resultRun) return undefined;
return queryStore.captureMultiDbExecutionWorkerResult(tab.id, workerId, sql, {
kind: "multi-db",
batchId: input.resultRun.batchId,
target: input.resultRun.target,
title: input.resultRun.title,
status,
durationMs: Date.now() - startedAt,
errorMessage,
});
};
try {
await queryStore.executeTabSql(executionTabId, sql, {
resultBaseSql: sql,
...(input.targetContext ? { targetContext: input.targetContext } : {}),
...(sourceOffset !== undefined ? { sourceOffset } : {}),
...(connection.db_type === "redis" ? { skipRedisSafetyCheck: !blockRedisCommands } : {}),
});
const latest = queryStore.getExecutionTab(executionTabId) ?? tab;
if (cancelRequested() || tabCancelRequested(cancelRequestCount)) {
return finish({ status: "cancelled" });
}
const failure = firstQueryExecutionError(latest);
const errorMessage = failure ? String(failure.rows?.[0]?.[0] ?? t("common.failed")) : undefined;
const success = !failure;
const resultStatus = success ? "success" : "failed";
captureWorkerResult(resultStatus, errorMessage);
historyStore.add({
connection_id: executionTab.connectionId,
connection_name: connection.name || "",
database: executionTab.database,
sql,
execution_time_ms: Date.now() - startedAt,
success,
error: errorMessage,
activity_kind: classifySqlActivityKind(sql),
operation: primarySqlOperation(sql),
affected_rows: success ? latest.result?.affected_rows : undefined,
});
if (success) {
const refreshTarget = sqlMetadataRefreshTarget(sql, executionTab.schema);
if (refreshTarget.scope === "connection") {
await connectionStore.loadDatabases(executionTab.connectionId, { force: true });
} else if (refreshTarget.scope === "database") {
await connectionStore.refreshObjectListTreeNode(executionTab.connectionId, executionTab.database, refreshTarget.schema);
}
}
// 多库 worker 路径(workerId 存在)下不切主编辑器输出视图:并行 Promise.all
// 会让多个 worker 几乎同时改这个共享 ref,导致主视图在 result/summary 间反复
// 跳动(闪烁/竞态)。worker 结果已由 captureMultiDbExecutionWorkerResult 记录
// 到 source tab 的 result run 并通过 projectResultRun 投影显示,无需再切主视图。
if (!workerId) {
deps.activeOutputView.value = success && (latest.result?.columns.length || latest.results?.some((result) => result.columns.length)) ? "result" : "summary";
}
return finish(success ? { status: "success", errorMessage } : { status: "failed", errorMessage });
} catch (error) {
const errorMessage = error instanceof Error ? error.message : String(error);
captureWorkerResult("failed", errorMessage);
return finish({ status: "failed", errorMessage });
} finally {
if (workerId) await queryStore.removeMultiDbExecutionWorker(workerId, input.scopeId);
}
}
function cancelActiveExecution() {
const tab = deps.activeTab.value;
if (!tab) return;
@@ -321,7 +511,10 @@ export function useSqlExecution(deps: {
const sourceOffset = pendingSourceOffset.value;
pendingSourceOffset.value = undefined;
pendingOpenInNewResultTab.value = false;
await continueExecute(sql, sourceOffset, { openInNewResultTab });
const continuation = pendingSqlParameterContinuation;
pendingSqlParameterContinuation = undefined;
if (continuation) await continuation(sql, sourceOffset);
else await continueExecute(sql, sourceOffset, { openInNewResultTab });
}
watch(showSqlParameterDialog, (open) => {
@@ -332,6 +525,7 @@ export function useSqlExecution(deps: {
sqlParameterEnabledSyntaxes.value = [];
pendingSourceOffset.value = undefined;
pendingOpenInNewResultTab.value = false;
pendingSqlParameterContinuation = undefined;
});
watch(showDangerDialog, (open) => {
@@ -360,6 +554,8 @@ export function useSqlExecution(deps: {
sqlParameterDatabaseType,
sqlParameterEnabledSyntaxes,
onSqlParametersConfirm,
prepareMultiExecute,
executeTargetSql,
explainMode,
};
}
@@ -373,11 +569,12 @@ export function supportsSqlTemplateParameters(connection: Pick<ConnectionConfig,
export function requiresDatabaseSelection(tab: QueryTab, connection: ConnectionConfig | undefined, _sql = ""): boolean {
if (tab.mode !== "query") return false;
if (!connection) return false;
const databaseType = effectiveDatabaseTypeForConnection(connection) ?? connection.db_type;
if (tab.database) return false;
if (tab.database === "" && usesTreeSchemaMode(connection.db_type)) return false;
if (isSingleDatabase(connection.db_type)) return false;
if (tab.database === "" && usesTreeSchemaMode(databaseType)) return false;
if (isSingleDatabase(databaseType)) return false;
// MySQL-compatible servers decide per statement whether a default database is required.
// Keep interactive execution connection-scoped instead of rejecting valid qualified or constant queries.
if (supportsConnectionLevelSqlExecution(connection)) return false;
return !["elasticsearch", "easysearch", "qdrant", "milvus", "weaviate", "chromadb", "zookeeper"].includes(connection.db_type);
return !supportsConnectionScopedQueryExecution(databaseType);
}