xiaoxue_admin
/
dbx
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix(mysql): use database-specific catalog switching

This commit is contained in:
jischeng 2026-07-29 20:41:09 +08:00 committed by GitHub
parent 9571519179
commit 2152f8cde8
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
31 changed files with 818 additions and 182 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
apps/desktop/src/App.vue
View File

@ -1525,9 +1525,14 @@ function changeActiveDatabase(database: string) {
}
}
function changeActiveCatalog(catalog: string | undefined, database: string) {
const tab = activeTab.value;
if (tab) queryStore.updateCatalog(tab.id, catalog, database);
}
async function setActiveDatabaseAsDefault() {
const tab = activeTab.value;
if (!tab || !tab.connectionId || !tab.database) return;
if (!tab || !tab.connectionId || !tab.database || tab.catalog) return;
await connectionStore.setDefaultDatabase(tab.connectionId, tab.database);
}
@ -1557,9 +1562,11 @@ function ensureQueryTab(): string {
const tab = activeTab.value;
if (tab && tab.mode === "query") return tab.id;
const connId = connectionStore.activeConnectionId || connectionStore.connections[0]?.id || "";
const db = tab?.connectionId === connId ? tab.database : connectionStore.getConfig(connId)?.database || "";
const schema = tab?.connectionId === connId ? tab.schema : undefined;
return queryStore.createTab(connId, db, undefined, "query", schema);
const sameConnectionTab = tab?.connectionId === connId ? tab : undefined;
const db = sameConnectionTab?.database || connectionStore.getConfig(connId)?.database || "";
const schema = sameConnectionTab?.schema ?? sameConnectionTab?.objectBrowser?.schema ?? sameConnectionTab?.tableMeta?.schema;
const catalog = sameConnectionTab?.catalog ?? sameConnectionTab?.objectBrowser?.catalog ?? sameConnectionTab?.tableMeta?.catalog;
return queryStore.createTab(connId, db, undefined, "query", schema, undefined, catalog);
}
function routeAiRedisCommand(command: string, execute: boolean): boolean {
@ -2263,6 +2270,7 @@ onUnmounted(() => {
@paste-sql-in-condition="pasteClipboardAsSqlInCondition"
@change-connection="changeActiveConnection"
@change-database="changeActiveDatabase"
@change-catalog="changeActiveCatalog"
@change-schema="changeActiveSchema"
@set-default-database="setActiveDatabaseAsDefault"
@clear-default-database="clearActiveDefaultDatabase"

82
apps/desktop/src/components/layout/EditorToolbar.vue
View File

@ -9,13 +9,14 @@ import TruncatedTextTooltip from "@/components/ui/TruncatedTextTooltip.vue";
import DatabaseIcon from "@/components/icons/DatabaseIcon.vue";
import ProductionContextBadge from "@/components/common/ProductionContextBadge.vue";
import { useConnectionStore } from "@/stores/connectionStore";
import { useDatabaseOptions } from "@/composables/useDatabaseOptions";
import { catalogDatabaseOptionsKey, databaseAfterCatalogChange, normalizedQueryTabCatalog, queryCatalogSelectorVisible, selectedQueryCatalogName, useDatabaseOptions } from "@/composables/useDatabaseOptions";
import { useSchemaOptions } from "@/composables/useSchemaOptions";
import { connectionIconType } from "@/lib/connection/connectionPresentation";
import { formatDatabaseLabel, isDefaultDatabase } from "@/lib/database/defaultDatabase";
import { connectionDisplayName } from "@/lib/tabs/tabPresentation";
import { useConnectionGroupLabel } from "@/composables/useConnectionGroupLabel";
import { isSingleDatabase, supportsClearableQuerySchema, supportsSqlInListPaste, supportsTransaction as supportsTransactionFeature } from "@/lib/database/databaseCapabilities";
import { connectionIsDorisFamilyCatalogCapable } from "@/lib/database/databaseFeatureSupport";
import { hexToRgba } from "@/lib/common/color";
import { productionContextForDatabase } from "@/lib/database/productionSafety";
import type { QueryTab, ConnectionConfig } from "@/types/database";
@ -46,6 +47,7 @@ const emit = defineEmits<{
importResultArchive: [];
pasteSqlInCondition: [];
changeConnection: [connectionId: string];
changeCatalog: [catalog: string | undefined, database: string];
changeDatabase: [database: string];
changeSchema: [schema: string | undefined];
setDefaultDatabase: [];
@ -59,13 +61,30 @@ const emit = defineEmits<{
const { t } = useI18n();
const connectionStore = useConnectionStore();
const { databaseOptions, loadingDatabaseOptions, loadDatabaseOptions } = useDatabaseOptions();
const { databaseOptions, loadingDatabaseOptions, loadDatabaseOptions, catalogOptions, loadingCatalogOptions, loadCatalogOptions, catalogDatabaseOptions, loadingCatalogDatabaseOptions, loadCatalogDatabaseOptions } = useDatabaseOptions();
const { loadSchemaOptions, getSchemaOptionsForDb, isLoadingSchemas, isSchemaAware } = useSchemaOptions();
const activeCatalogs = computed(() => {
const connection = props.activeConnection;
return connection ? (catalogOptions.value[connection.id] ?? []) : [];
});
const activeCatalogNames = computed(() => activeCatalogs.value.map((catalog) => catalog.name));
const showCatalogSelector = computed(() => connectionIsDorisFamilyCatalogCapable(props.activeConnection) && queryCatalogSelectorVisible(activeCatalogs.value));
const activeCatalogValue = computed(() => selectedQueryCatalogName(activeCatalogs.value, props.activeTab.catalog));
const activeCatalogDatabaseKey = computed(() => (props.activeConnection && props.activeTab.catalog ? catalogDatabaseOptionsKey(props.activeConnection.id, props.activeTab.catalog) : ""));
const activeDatabaseOptions = computed(() => {
const connection = props.activeConnection;
return connection ? (databaseOptions.value[connection.id] ?? []) : [];
if (!connection) return [];
if (props.activeTab.catalog) return catalogDatabaseOptions.value[activeCatalogDatabaseKey.value] ?? [];
return databaseOptions.value[connection.id] ?? [];
});
const loadingActiveDatabaseOptions = computed(() => {
const connection = props.activeConnection;
if (!connection) return false;
if (props.activeTab.catalog) return loadingCatalogDatabaseOptions.value[activeCatalogDatabaseKey.value] ?? false;
return loadingDatabaseOptions.value[connection.id] ?? false;
});
const switchingCatalog = ref(false);
const connectionOptionIds = computed(() => connectionStore.connections.map((connection) => connection.id));
const { connectionGroupLabel } = useConnectionGroupLabel();
@ -135,6 +154,18 @@ watchEffect(() => {
loadSchemaOptions(connection.id, schemaDatabaseKey.value).catch(() => {});
}
});
watchEffect(() => {
const connection = props.activeConnection;
if (!connection || !connectionIsDorisFamilyCatalogCapable(connection)) return;
void loadCatalogOptions(connection.id).catch(() => {});
});
watchEffect(() => {
const connection = props.activeConnection;
const catalog = props.activeTab.catalog;
if (!connection || !catalog) return;
void loadCatalogDatabaseOptions(connection.id, catalog).catch(() => {});
});
const isActiveDatabaseDefault = computed(() => isDefaultDatabase(props.activeConnection, activeDatabaseValue.value));
const toolbarStyle = computed(() => {
@ -161,6 +192,18 @@ function databaseOptionIsProduction(database: string): boolean {
if (!database || props.activeConnection?.is_production) return false;
return productionContextForDatabase(props.activeConnection, database).reason === "database";
}
async function changeCatalog(selectedCatalog: string) {
const connection = props.activeConnection;
if (!connection) return;
switchingCatalog.value = true;
try {
const catalog = normalizedQueryTabCatalog(activeCatalogs.value, selectedCatalog);
const databases = catalog ? await loadCatalogDatabaseOptions(connection.id, selectedCatalog) : await loadDatabaseOptions(connection.id).then(() => databaseOptions.value[connection.id] ?? []);
emit("changeCatalog", catalog, databaseAfterCatalogChange(props.activeTab.database, databases));
} finally {
switchingCatalog.value = false;
}
}
</script>
<template>
@ -364,6 +407,31 @@ function databaseOptionIsProduction(database: string): boolean {
</template>
</SearchableSelect>
</div>
<div v-if="showCatalogSelector" class="flex items-center gap-1">
<SearchableSelect
:model-value="activeCatalogValue"
:options="activeCatalogNames"
:placeholder="t('editor.selectCatalog')"
:search-placeholder="t('editor.searchCatalog')"
:empty-text="t('grid.noSearchResults')"
:loading-text="t('common.loading')"
:loading="loadingCatalogOptions[activeConnection?.id || ''] || switchingCatalog"
trigger-variant="ghost"
trigger-class="gap-1.5"
trigger-icon-class="h-3 w-3"
@update:model-value="changeCatalog"
@update:open="
(open: boolean) => {
if (open && activeConnection) loadCatalogOptions(activeConnection.id).catch(() => {});
}
"
>
<template #trigger-label="{ label, loading }">
<Layers class="h-3.5 w-3.5 shrink-0" />
<span class="truncate">{{ loading ? t("common.loading") : label }}</span>
</template>
</SearchableSelect>
</div>
<div
v-if="activeConnection?.db_type !== 'elasticsearch' && activeConnection?.db_type !== 'qdrant' && activeConnection?.db_type !== 'milvus' && activeConnection?.db_type !== 'weaviate' && activeConnection?.db_type !== 'chromadb' && activeConnection?.db_type !== 'zookeeper' && !isSingleDb"
class="flex items-center gap-1"
@ -376,7 +444,7 @@ function databaseOptionIsProduction(database: string): boolean {
:search-placeholder="t('editor.searchDatabase')"
:empty-text="t('grid.noSearchResults')"
:loading-text="t('common.loading')"
:loading="loadingDatabaseOptions[activeConnection?.id || '']"
:loading="loadingActiveDatabaseOptions"
:display-name="databaseDisplayName"
trigger-variant="ghost"
trigger-class="gap-1.5"
@ -384,7 +452,9 @@ function databaseOptionIsProduction(database: string): boolean {
@update:model-value="(database) => emit('changeDatabase', database)"
@update:open="
(open: boolean) => {
if (open && activeConnection) loadDatabaseOptions(activeConnection.id).catch(() => {});
if (!open || !activeConnection) return;
if (activeTab.catalog) loadCatalogDatabaseOptions(activeConnection.id, activeTab.catalog).catch(() => {});
else loadDatabaseOptions(activeConnection.id).catch(() => {});
}
"
>
@ -408,7 +478,7 @@ function databaseOptionIsProduction(database: string): boolean {
</TooltipTrigger>
<TooltipContent>{{ t("editor.clearDatabase") }}</TooltipContent>
</Tooltip>
<Button v-if="activeDatabaseValue" variant="ghost" size="sm" class="h-6 px-2 text-[11px]" @click="isActiveDatabaseDefault ? emit('clearDefaultDatabase') : emit('setDefaultDatabase')">
<Button v-if="activeDatabaseValue && !activeTab.catalog" variant="ghost" size="sm" class="h-6 px-2 text-[11px]" @click="isActiveDatabaseDefault ? emit('clearDefaultDatabase') : emit('setDefaultDatabase')">
<Check v-if="isActiveDatabaseDefault" class="h-3 w-3" />
{{ isActiveDatabaseDefault ? t("editor.defaultDatabase") : t("editor.setDefaultDatabase") }}
</Button>

11
apps/desktop/src/components/objects/ObjectBrowser.vue
View File

@ -1142,13 +1142,16 @@ async function openSource(row: ObjectBrowserRow) {
}
async function openNewQuery(row: ObjectBrowserRow) {
const tabId = queryStore.createTab(props.connection.id, props.database, row.name);
const schema = row.schema || selectedSchema.value;
const tabId = queryStore.createTab(props.connection.id, props.database, row.name, "query", schema, undefined, props.catalog);
queryStore.updateSql(
tabId,
await buildTableSelectSql({
databaseType: effectiveDatabaseType.value,
identifierQuote: connectionStore.connectionIdentifierQuote?.(props.connection.id),
schema: row.schema || selectedSchema.value,
catalog: props.catalog,
database: props.database,
schema,
tableName: row.name,
limit: 100,
}),
@ -1165,7 +1168,7 @@ function openProcedureExecutionSql(sql: string) {
const row = procedureExecutionTarget.value;
if (!row || !sql) return;
const schema = row.schema || selectedSchema.value;
const tabId = queryStore.createTab(props.connection.id, props.database, `Execute - ${row.name}`, "query", schema);
const tabId = queryStore.createTab(props.connection.id, props.database, `Execute - ${row.name}`, "query", schema, undefined, props.catalog);
queryStore.updateSql(tabId, sql);
}
@ -1173,7 +1176,7 @@ async function executeProcedureSql(sql: string) {
const row = procedureExecutionTarget.value;
if (!row || !sql) return;
const schema = row.schema || selectedSchema.value;
const tabId = queryStore.createTab(props.connection.id, props.database, `Execute - ${row.name}`, "query", schema);
const tabId = queryStore.createTab(props.connection.id, props.database, `Execute - ${row.name}`, "query", schema, undefined, props.catalog);
queryStore.updateSql(tabId, sql);
await queryStore.executeTabSql(tabId, sql);
}

4
apps/desktop/src/components/sidebar/ConnectionTree.vue
View File

@ -1316,14 +1316,14 @@ function clearSidebarTableNameFilters() {
function openSidebarProcedureSql(sql: string) {
const target = sidebarProcedureTarget.value;
if (!target?.connectionId || !target.database || !sql) return;
const tabId = queryStore.createTab(target.connectionId, target.database, `Execute - ${target.label}`, "query", target.schema);
const tabId = queryStore.createTab(target.connectionId, target.database, `Execute - ${target.label}`, "query", target.schema, undefined, target.catalog);
queryStore.updateSql(tabId, sql);
}
async function executeSidebarProcedureSql(sql: string) {
const target = sidebarProcedureTarget.value;
if (!target?.connectionId || !target.database || !sql) return;
const tabId = queryStore.createTab(target.connectionId, target.database, `Execute - ${target.label}`, "query", target.schema);
const tabId = queryStore.createTab(target.connectionId, target.database, `Execute - ${target.label}`, "query", target.schema, undefined, target.catalog);
queryStore.updateSql(tabId, sql);
await queryStore.executeTabSql(tabId, sql);
}

30
apps/desktop/src/components/sidebar/SidebarTreeRuntimeHost.vue
View File

@ -1186,7 +1186,7 @@ async function loadTemplateContext(allowView = false) {
let columns: ColumnInfo[] = [];
try {
const querySchema = connectionObjectTreeQuerySchema(config, node.database, tableSchema);
columns = await api.getColumns(node.connectionId, node.database, querySchema, node.label);
columns = await api.getColumns(node.connectionId, node.database, querySchema, node.label, node.catalog);
} catch (e) {
console.warn("[DBX][tableSqlTemplate:getColumns:error]", e);
}
@ -1195,7 +1195,7 @@ async function loadTemplateContext(allowView = false) {
if (dbType === "tdengine") {
try {
const querySchema = connectionObjectTreeQuerySchema(config, node.database, tableSchema);
const tables = await api.listTables(node.connectionId, node.database, querySchema, node.label, 200);
const tables = await api.listTables(node.connectionId, node.database, querySchema, node.label, 200, undefined, undefined, node.catalog);
const matched = tables.find((table) => table.name.toLowerCase() === node.label.toLowerCase());
if (matched?.table_type) tableType = matched.table_type;
} catch (e) {
@ -1206,8 +1206,8 @@ async function loadTemplateContext(allowView = false) {
return { node, dbType, tableSchema, columns, tableType };
}
function openSqlTemplateTab(connectionId: string, database: string, schema: string | undefined, sql: string, title?: string) {
const tabId = queryStore.createTab(connectionId, database, title, "query", schema);
function openSqlTemplateTab(connectionId: string, database: string, schema: string | undefined, catalog: string | undefined, sql: string, title?: string) {
const tabId = queryStore.createTab(connectionId, database, title, "query", schema, undefined, catalog);
queryStore.updateSql(tabId, sql);
}
@ -1217,11 +1217,13 @@ async function newSelectTemplate() {
if (!context) return;
const sql = buildTableSelectTemplate({
databaseType: context.dbType,
catalog: context.node.catalog,
database: context.node.database,
schema: context.tableSchema,
tableName: context.node.label,
columns: context.columns,
});
openSqlTemplateTab(context.node.connectionId!, context.node.database!, context.node.schema, sql);
openSqlTemplateTab(context.node.connectionId!, context.node.database!, context.node.schema, context.node.catalog, sql);
} catch (e: any) {
toast(t("connection.connectFailed", { message: translateBackendError(t, e?.message || String(e)) }), 5000);
}
@ -1233,12 +1235,14 @@ async function newInsertTemplate() {
if (!context) return;
const sql = buildTableInsertTemplate({
databaseType: context.dbType,
catalog: context.node.catalog,
database: context.node.database,
schema: context.tableSchema,
tableName: context.node.label,
columns: context.columns,
tableType: context.tableType,
});
openSqlTemplateTab(context.node.connectionId!, context.node.database!, context.node.schema, sql);
openSqlTemplateTab(context.node.connectionId!, context.node.database!, context.node.schema, context.node.catalog, sql);
} catch (e: any) {
toast(t("connection.connectFailed", { message: translateBackendError(t, e?.message || String(e)) }), 5000);
}
@ -1250,11 +1254,13 @@ async function newUpdateTemplate() {
if (!context) return;
const sql = buildTableUpdateTemplate({
databaseType: context.dbType,
catalog: context.node.catalog,
database: context.node.database,
schema: context.tableSchema,
tableName: context.node.label,
columns: context.columns,
});
openSqlTemplateTab(context.node.connectionId!, context.node.database!, context.node.schema, sql);
openSqlTemplateTab(context.node.connectionId!, context.node.database!, context.node.schema, context.node.catalog, sql);
} catch (e: any) {
toast(t("connection.connectFailed", { message: translateBackendError(t, e?.message || String(e)) }), 5000);
}
@ -1266,11 +1272,13 @@ async function newDeleteTemplate() {
if (!context) return;
const sql = buildTableDeleteTemplate({
databaseType: context.dbType,
catalog: context.node.catalog,
database: context.node.database,
schema: context.tableSchema,
tableName: context.node.label,
columns: context.columns,
});
openSqlTemplateTab(context.node.connectionId!, context.node.database!, context.node.schema, sql);
openSqlTemplateTab(context.node.connectionId!, context.node.database!, context.node.schema, context.node.catalog, sql);
} catch (e: any) {
toast(t("connection.connectFailed", { message: translateBackendError(t, e?.message || String(e)) }), 5000);
}
@ -1299,7 +1307,7 @@ async function generateDdlTemplate() {
});
}
const formatted = await formatSqlForDisplay(ddl, sqlFormatDialectForDbType(currentDatabaseType()), settingsStore.editorSettings.sqlFormatter);
openSqlTemplateTab(node.connectionId, node.database, node.schema, formatted, `DDL - ${node.label}`);
openSqlTemplateTab(node.connectionId, node.database, node.schema, node.catalog, formatted, `DDL - ${node.label}`);
} catch (e: any) {
toast(e?.message || String(e), 5000);
}
@ -3042,7 +3050,7 @@ function createView() {
const viewName = "new_view";
const effectiveDbType = effectiveDatabaseTypeForConnection(connectionStore.getConfig(node.connectionId));
const viewSqlName = effectiveDbType === "informix" || !node.schema ? viewName : `${node.schema}.${viewName}`;
const tabId = queryStore.createTab(node.connectionId, node.database, t("contextMenu.createView"), "query", node.schema);
const tabId = queryStore.createTab(node.connectionId, node.database, t("contextMenu.createView"), "query", node.schema, undefined, node.catalog);
queryStore.updateSql(tabId, `CREATE VIEW ${viewSqlName} AS\nSELECT\n *\nFROM table_name;\n`);
queryStore.setObjectSource(tabId, {
schema: node.schema,
@ -3077,7 +3085,7 @@ function createMysqlObjectTemplate() {
const template = mysqlObjectTemplateForGroup(connectionStore.getConfig(node.connectionId), node);
if (!template) return;
connectionStore.activeConnectionId = node.connectionId;
const tabId = queryStore.createTab(node.connectionId, node.database, t(template.titleKey), "query", node.schema);
const tabId = queryStore.createTab(node.connectionId, node.database, t(template.titleKey), "query", node.schema, undefined, node.catalog);
queryStore.updateSql(tabId, template.sql);
}

57
apps/desktop/src/composables/__tests__/useDatabaseOptions.spec.ts vendored
View File

@ -1,16 +1,30 @@
import { beforeEach, describe, expect, it, vi } from "vitest";
import { databaseOptionsForConnection, fetchNamespaceOptionsForConnection, fetchSqlFileTargetOptions, namespaceOptionsAreSchemas, useDatabaseOptions } from "@/composables/useDatabaseOptions";
import {
databaseAfterCatalogChange,
databaseOptionsForConnection,
fetchNamespaceOptionsForConnection,
fetchSqlFileTargetOptions,
namespaceOptionsAreSchemas,
normalizedQueryTabCatalog,
queryCatalogSelectorVisible,
selectedQueryCatalogName,
useDatabaseOptions,
} from "@/composables/useDatabaseOptions";
const mocks = vi.hoisted(() => ({
ensureConnected: vi.fn(),
getConfig: vi.fn(),
listDatabases: vi.fn(),
listSchemas: vi.fn(),
listDorisCatalogs: vi.fn(),
listDorisCatalogDatabases: vi.fn(),
}));
vi.mock("@/lib/backend/api", () => ({
listDatabases: mocks.listDatabases,
listSchemas: mocks.listSchemas,
listDorisCatalogs: mocks.listDorisCatalogs,
listDorisCatalogDatabases: mocks.listDorisCatalogDatabases,
}));
vi.mock("@/stores/connectionStore", () => ({
@ -20,6 +34,47 @@ vi.mock("@/stores/connectionStore", () => ({
}),
}));
const internalCatalog = { name: "default_catalog", catalog_type: "internal", is_current: true };
const paimonCatalog = { name: "paimon_catalog", catalog_type: "paimon", is_current: false };
describe("query catalog options", () => {
beforeEach(() => {
vi.clearAllMocks();
});
it("shows the selector only when an external catalog exists", () => {
expect(queryCatalogSelectorVisible([internalCatalog])).toBe(false);
expect(queryCatalogSelectorVisible([internalCatalog, paimonCatalog])).toBe(true);
});
it("maps the internal catalog to an undefined tab catalog", () => {
const catalogs = [internalCatalog, paimonCatalog];
expect(selectedQueryCatalogName(catalogs)).toBe("default_catalog");
expect(normalizedQueryTabCatalog(catalogs, "default_catalog")).toBeUndefined();
expect(normalizedQueryTabCatalog(catalogs, "paimon_catalog")).toBe("paimon_catalog");
});
it("keeps a database only when it exists in the selected catalog", () => {
expect(databaseAfterCatalogChange("bi", ["bi", "default"])).toBe("bi");
expect(databaseAfterCatalogChange("bi", ["analytics"])).toBe("");
});
it("loads and caches catalog-scoped databases", async () => {
mocks.getConfig.mockReturnValue({ db_type: "starrocks" });
mocks.listDorisCatalogs.mockResolvedValue([internalCatalog, paimonCatalog]);
mocks.listDorisCatalogDatabases.mockResolvedValue([{ name: "bi" }, { name: "analytics" }]);
const options = useDatabaseOptions();
await expect(options.loadCatalogOptions("connection-1")).resolves.toEqual([internalCatalog, paimonCatalog]);
await expect(options.loadCatalogDatabaseOptions("connection-1", "paimon_catalog")).resolves.toEqual(["bi", "analytics"]);
await options.loadCatalogDatabaseOptions("connection-1", "paimon_catalog");
expect(options.catalogDatabaseOptions.value["connection-1:paimon_catalog"]).toEqual(["bi", "analytics"]);
expect(mocks.listDorisCatalogDatabases).toHaveBeenCalledTimes(1);
});
});
describe("namespace options", () => {
beforeEach(() => {
vi.clearAllMocks();

94
apps/desktop/src/composables/useDatabaseOptions.ts
View File

@ -2,10 +2,33 @@ import { ref } from "vue";
import { useConnectionStore } from "@/stores/connectionStore";
import { filterDatabaseNamesForConnection, filterSchemaNamesForConnection } from "@/lib/database/visibleDatabases";
import { usesTreeSchemaMode } from "@/lib/database/databaseCapabilities";
import type { ConnectionConfig } from "@/types/database";
import { isInternalDorisCatalog } from "@/lib/database/databaseFeatureSupport";
import type { CatalogInfo, ConnectionConfig } from "@/types/database";
import * as api from "@/lib/backend/api";
type NamespaceOptionsConnection = Pick<ConnectionConfig, "database" | "db_type" | "driver_profile" | "visible_databases" | "visible_schemas">;
export function catalogDatabaseOptionsKey(connectionId: string, catalog: string): string {
return `${connectionId}:${catalog}`;
}
export function queryCatalogSelectorVisible(catalogs: CatalogInfo[]): boolean {
return catalogs.some((catalog) => !isInternalDorisCatalog(catalog.catalog_type, catalog.name));
}
export function selectedQueryCatalogName(catalogs: CatalogInfo[], tabCatalog?: string): string {
if (tabCatalog) return tabCatalog;
return catalogs.find((catalog) => isInternalDorisCatalog(catalog.catalog_type, catalog.name))?.name ?? "";
}
export function normalizedQueryTabCatalog(catalogs: CatalogInfo[], selectedCatalog: string): string | undefined {
if (!selectedCatalog) return undefined;
const catalog = catalogs.find((candidate) => candidate.name === selectedCatalog);
return catalog && isInternalDorisCatalog(catalog.catalog_type, catalog.name) ? undefined : selectedCatalog;
}
export function databaseAfterCatalogChange(currentDatabase: string, databaseOptions: string[]): string {
return databaseOptions.includes(currentDatabase) ? currentDatabase : "";
}
export function databaseOptionsForConnection(databaseNames: string[], connection: Pick<ConnectionConfig, "db_type" | "visible_databases"> | undefined): string[] {
const names = filterDatabaseNamesForConnection(databaseNames, connection);
@ -42,6 +65,12 @@ export function useDatabaseOptions() {
const databaseOptions = ref<Record<string, string[]>>({});
const loadingDatabaseOptions = ref<Record<string, boolean>>({});
const catalogOptions = ref<Record<string, CatalogInfo[]>>({});
const loadingCatalogOptions = ref<Record<string, boolean>>({});
const catalogDatabaseOptions = ref<Record<string, string[]>>({});
const loadingCatalogDatabaseOptions = ref<Record<string, boolean>>({});
const catalogRequests = new Map<string, Promise<CatalogInfo[]>>();
const catalogDatabaseRequests = new Map<string, Promise<string[]>>();
async function loadDatabaseOptions(connectionId: string) {
const connection = connectionStore.getConfig(connectionId);
@ -70,6 +99,56 @@ export function useDatabaseOptions() {
}
}
async function loadCatalogOptions(connectionId: string): Promise<CatalogInfo[]> {
const cached = catalogOptions.value[connectionId];
if (cached) return cached;
const pending = catalogRequests.get(connectionId);
if (pending) return pending;
const request = (async () => {
loadingCatalogOptions.value[connectionId] = true;
try {
await connectionStore.ensureConnected(connectionId);
const catalogs = await api.listDorisCatalogs(connectionId);
catalogOptions.value[connectionId] = catalogs;
return catalogs;
} finally {
loadingCatalogOptions.value[connectionId] = false;
catalogRequests.delete(connectionId);
}
})();
catalogRequests.set(connectionId, request);
return request;
}
async function loadCatalogDatabaseOptions(connectionId: string, catalog: string): Promise<string[]> {
const key = catalogDatabaseOptionsKey(connectionId, catalog);
const cached = catalogDatabaseOptions.value[key];
if (cached) return cached;
const pending = catalogDatabaseRequests.get(key);
if (pending) return pending;
const request = (async () => {
loadingCatalogDatabaseOptions.value[key] = true;
try {
await connectionStore.ensureConnected(connectionId);
const connection = connectionStore.getConfig(connectionId);
const databases = await api.listDorisCatalogDatabases(connectionId, catalog);
const names = databaseOptionsForConnection(
databases.map((database) => database.name),
connection,
);
catalogDatabaseOptions.value[key] = names;
return names;
} finally {
loadingCatalogDatabaseOptions.value[key] = false;
catalogDatabaseRequests.delete(key);
}
})();
catalogDatabaseRequests.set(key, request);
return request;
}
async function getDatabaseOptions(connectionId: string): Promise<string[]> {
if (!databaseOptions.value[connectionId]) {
await loadDatabaseOptions(connectionId);
@ -77,5 +156,16 @@ export function useDatabaseOptions() {
return databaseOptions.value[connectionId] ?? [];
}
return { databaseOptions, loadingDatabaseOptions, loadDatabaseOptions, getDatabaseOptions };
return {
databaseOptions,
loadingDatabaseOptions,
loadDatabaseOptions,
getDatabaseOptions,
catalogOptions,
loadingCatalogOptions,
loadCatalogOptions,
catalogDatabaseOptions,
loadingCatalogDatabaseOptions,
loadCatalogDatabaseOptions,
};
}

2
apps/desktop/src/i18n/locales/en.ts
View File

@ -720,6 +720,8 @@ export default {
saveAllChanges: "Save all",
selectConnection: "Select connection",
searchConnection: "Search connections...",
selectCatalog: "Select catalog",
searchCatalog: "Search catalogs...",
selectDatabase: "Select database",
selectDatabaseRequired: "Select a database first",
searchDatabase: "Search databases...",

2
apps/desktop/src/i18n/locales/es.ts
View File

@ -700,6 +700,8 @@ export default withEnglishFallback({
saveAllChanges: "Guardar todo",
selectConnection: "Seleccionar conexión",
searchConnection: "Buscar conexiones...",
selectCatalog: "Seleccionar catálogo",
searchCatalog: "Buscar catálogos...",
selectDatabase: "Seleccionar base de datos",
selectDatabaseRequired: "Selecciona una base de datos primero",
searchDatabase: "Buscar bases de datos...",

2
apps/desktop/src/i18n/locales/it.ts
View File

@ -698,6 +698,8 @@ export default withEnglishFallback({
saveAllChanges: "Salva tutto",
selectConnection: "Seleziona connessione",
searchConnection: "Cerca connessioni...",
selectCatalog: "Seleziona catalogo",
searchCatalog: "Cerca cataloghi...",
selectDatabase: "Seleziona database",
selectDatabaseRequired: "Seleziona prima un database",
searchDatabase: "Cerca database...",

2
apps/desktop/src/i18n/locales/ja.ts
View File

@ -698,6 +698,8 @@ export default withEnglishFallback({
saveAllChanges: "すべて保存",
selectConnection: "接続を選択",
searchConnection: "接続を検索...",
selectCatalog: "カタログを選択",
searchCatalog: "カタログを検索...",
selectDatabase: "データベースを選択",
searchDatabase: "データベースを検索...",
selectSchema: "スキーマを選択",

2
apps/desktop/src/i18n/locales/pt-BR.ts
View File

@ -699,6 +699,8 @@ export default withEnglishFallback({
saveAllChanges: "Salvar tudo",
selectConnection: "Selecionar conexão",
searchConnection: "Pesquisar conexões...",
selectCatalog: "Selecionar catálogo",
searchCatalog: "Pesquisar catálogos...",
selectDatabase: "Selecionar banco de dados",
selectDatabaseRequired: "Selecione um banco de dados primeiro",
searchDatabase: "Pesquisar bancos de dados...",

2
apps/desktop/src/i18n/locales/zh-CN.ts
View File

@ -721,6 +721,8 @@ export default withEnglishFallback({
saveAllChanges: "全部保存",
selectConnection: "选择连接",
searchConnection: "搜索连接...",
selectCatalog: "选择 Catalog",
searchCatalog: "搜索 Catalog...",
selectDatabase: "选择数据库",
selectDatabaseRequired: "请先选择数据库",
searchDatabase: "搜索数据库...",

2
apps/desktop/src/i18n/locales/zh-TW.ts
View File

@ -698,6 +698,8 @@ export default withEnglishFallback({
saveAllChanges: "全部儲存",
selectConnection: "選擇連線",
searchConnection: "搜尋連線……",
selectCatalog: "選擇 Catalog",
searchCatalog: "搜尋 Catalog……",
selectDatabase: "選擇資料庫",
selectDatabaseRequired: "請先選擇資料庫",
searchDatabase: "搜尋資料庫……",

46
apps/desktop/src/lib/__tests__/sql/newQueryContext.spec.ts vendored
View File

@ -1,5 +1,6 @@
import { describe, expect, it } from "vitest";
import { buildSelectAllSql, isNewQueryPrefillSupported, resolveNewQueryInitialSql, resolveNewQueryTable, type ResolveNewQueryTableInput } from "@/lib/sql/newQueryContext";
import { buildSelectAllSql, isNewQueryPrefillSupported, resolveNewQueryInitialSql, resolveNewQueryTable, resolveNewQueryTarget } from "@/lib/sql/newQueryContext";
import type { ResolveNewQueryTableInput } from "@/lib/sql/newQueryContext";
import type { QueryTab, TreeNode } from "@/types/database";
function dataTab(overrides: Partial<Pick<QueryTab, "mode" | "connectionId" | "database" | "schema" | "tableMeta" | "structureTableName" | "title">> = {}): ResolveNewQueryTableInput["activeTab"] {
@ -18,6 +19,46 @@ function tableNode(overrides: Partial<Pick<TreeNode, "type" | "connectionId" | "
return { type: "table", connectionId: "conn-1", database: "app_db", schema: "public", tableName: "orders", label: "orders", ...overrides };
}
describe("resolveNewQueryTarget", () => {
it("inherits an external catalog from the active object browser", () => {
expect(
resolveNewQueryTarget({
activeTab: {
connectionId: "conn-1",
database: "bi",
objectBrowser: { catalog: "paimon_catalog" },
},
connections: [{ id: "conn-1", database: "" }],
preferredSource: "tab",
}),
).toEqual({
connectionId: "conn-1",
database: "bi",
schema: undefined,
catalog: "paimon_catalog",
shouldRefreshDefaultDatabase: false,
});
});
it("inherits an external catalog from active table metadata", () => {
expect(
resolveNewQueryTarget({
activeTab: {
connectionId: "conn-1",
database: "bi",
tableMeta: {
catalog: "paimon_catalog",
tableName: "events",
columns: [],
primaryKeys: [],
},
},
connections: [{ id: "conn-1", database: "" }],
})?.catalog,
).toBe("paimon_catalog");
});
});
describe("resolveNewQueryTable", () => {
it("resolves the table from an active data tab", () => {
const table = resolveNewQueryTable({ activeTab: dataTab(), preferredSource: "tab" });
@ -119,6 +160,9 @@ describe("buildSelectAllSql", () => {
expect(buildSelectAllSql("mysql", { tableName: "a`b" })).toBe("SELECT * FROM `a``b`");
expect(buildSelectAllSql("postgres", { tableName: 'a"b' })).toBe('SELECT * FROM "a""b"');
});
it("qualifies a StarRocks external-catalog table with catalog and database", () => {
expect(buildSelectAllSql("starrocks", { catalog: "paimon_catalog", database: "bi", tableName: "events" })).toBe("SELECT * FROM `paimon_catalog`.`bi`.`events`");
});
});
describe("isNewQueryPrefillSupported", () => {

14
apps/desktop/src/lib/backend/http.ts
View File

@ -813,6 +813,7 @@ export async function executeQuery(
executionId?: string,
options?: {
maxRows?: number;
catalog?: string;
fetchSize?: number;
pageSize?: number;
resultSessionId?: string;
@ -839,6 +840,7 @@ export async function executeMulti(
executionId?: string,
options?: {
maxRows?: number;
catalog?: string;
fetchSize?: number;
pageSize?: number;
resultSessionId?: string;
@ -874,6 +876,7 @@ export async function executeMultiWithProgress(
schema?: string,
options?: {
maxRows?: number;
catalog?: string;
fetchSize?: number;
pageSize?: number;
resultSessionId?: string;
@ -895,20 +898,22 @@ export async function executeMultiWithProgress(
return results;
}
export async function closeQuerySession(connectionId: string, database: string, sessionId: string, clientSessionId?: string): Promise<boolean> {
export async function closeQuerySession(connectionId: string, database: string, sessionId: string, clientSessionId?: string, catalog?: string): Promise<boolean> {
return post("/api/query/close-session", {
connectionId,
database,
sessionId,
clientSessionId,
catalog,
});
}
export async function closeClientConnectionSession(connectionId: string, database: string, clientSessionId: string): Promise<boolean> {
export async function closeClientConnectionSession(connectionId: string, database: string, clientSessionId: string, catalog?: string): Promise<boolean> {
return post("/api/query/close-client-session", {
connectionId,
database,
clientSessionId,
catalog,
});
}
@ -939,16 +944,17 @@ export async function executeScriptWith2pc(connectionId: string, database: strin
});
}
export async function executeInTransaction(connectionId: string, database: string, statements: string[], schema?: string): Promise<QueryResult> {
export async function executeInTransaction(connectionId: string, database: string, statements: string[], schema?: string, catalog?: string): Promise<QueryResult> {
return post("/api/query/execute-in-transaction", {
connectionId,
database,
statements,
schema,
catalog,
});
}
export async function beginManualTransaction(_connectionId: string, _database: string, _schema?: string): Promise<string> {
export async function beginManualTransaction(_connectionId: string, _database: string, _schema?: string, _catalog?: string): Promise<string> {
throw new Error("Manual transaction management is only available in the desktop app.");
}

16
apps/desktop/src/lib/backend/tauri.ts
View File

@ -1021,6 +1021,7 @@ export async function executeQuery(
executionId?: string,
options?: {
maxRows?: number;
catalog?: string;
fetchSize?: number;
pageSize?: number;
resultSessionId?: string;
@ -1047,6 +1048,7 @@ export async function executeMulti(
executionId?: string,
options?: {
maxRows?: number;
catalog?: string;
fetchSize?: number;
pageSize?: number;
resultSessionId?: string;
@ -1082,6 +1084,7 @@ export async function executeMultiWithProgress(
schema?: string,
options?: {
maxRows?: number;
catalog?: string;
fetchSize?: number;
pageSize?: number;
resultSessionId?: string;
@ -1118,20 +1121,22 @@ export async function cancelQuery(executionId: string): Promise<boolean> {
return invoke("cancel_query", { executionId });
}
export async function closeQuerySession(connectionId: string, database: string, sessionId: string, clientSessionId?: string): Promise<boolean> {
export async function closeQuerySession(connectionId: string, database: string, sessionId: string, clientSessionId?: string, catalog?: string): Promise<boolean> {
return invoke("close_query_session", {
connectionId,
database,
sessionId,
clientSessionId,
catalog,
});
}
export async function closeClientConnectionSession(connectionId: string, database: string, clientSessionId: string): Promise<boolean> {
export async function closeClientConnectionSession(connectionId: string, database: string, clientSessionId: string, catalog?: string): Promise<boolean> {
return invoke("close_client_connection_session", {
connectionId,
database,
clientSessionId,
catalog,
});
}
@ -1158,17 +1163,18 @@ export async function executeScriptWith2pc(connectionId: string, database: strin
});
}
export async function executeInTransaction(connectionId: string, database: string, statements: string[], schema?: string): Promise<QueryResult> {
export async function executeInTransaction(connectionId: string, database: string, statements: string[], schema?: string, catalog?: string): Promise<QueryResult> {
return invoke("execute_in_transaction", {
connectionId,
database,
statements,
schema,
catalog,
});
}
export async function beginManualTransaction(connectionId: string, database: string, schema?: string): Promise<string> {
return invoke("begin_manual_transaction", { connectionId, database, schema });
export async function beginManualTransaction(connectionId: string, database: string, schema?: string, catalog?: string): Promise<string> {
return invoke("begin_manual_transaction", { connectionId, database, schema, catalog });
}
export async function executeInManualTransaction(txnSessionId: string, sql: string, database: string, schema?: string, maxRows?: number): Promise<QueryResult[]> {

18
apps/desktop/src/lib/sql/newQueryContext.ts
View File

@ -13,7 +13,7 @@ export interface NewQueryTarget {
export type NewQueryContextSource = "tab" | "sidebar";
interface ResolveNewQueryTargetInput {
activeTab?: Pick<QueryTab, "connectionId" | "database" | "schema" | "catalog">;
activeTab?: Pick<QueryTab, "connectionId" | "database" | "schema" | "catalog" | "objectBrowser" | "tableMeta">;
selectedTreeNode?: Pick<TreeNode, "connectionId" | "database" | "schema" | "catalog"> | null;
activeConnectionId?: string | null;
connections: Pick<ConnectionConfig, "id" | "database">[];
@ -49,16 +49,18 @@ export function resolveNewQueryTarget(input: ResolveNewQueryTargetInput): NewQue
: null;
}
function targetFromContext(context: Pick<QueryTab | TreeNode, "connectionId" | "database" | "schema" | "catalog"> | undefined, connections: Pick<ConnectionConfig, "id" | "database">[]): NewQueryTarget | null {
function targetFromContext(context: Pick<QueryTab, "connectionId" | "database" | "schema" | "catalog" | "objectBrowser" | "tableMeta"> | Pick<TreeNode, "connectionId" | "database" | "schema" | "catalog"> | undefined, connections: Pick<ConnectionConfig, "id" | "database">[]): NewQueryTarget | null {
if (!context?.connectionId) return null;
const connection = connections.find((item) => item.id === context.connectionId);
if (!connection) return null;
const database = context.database || resolveDefaultDatabase(connection, []);
const objectBrowser = "objectBrowser" in context ? context.objectBrowser : undefined;
const tableMeta = "tableMeta" in context ? context.tableMeta : undefined;
return {
connectionId: context.connectionId,
database,
schema: "schema" in context ? (context as { schema?: string }).schema : undefined,
catalog: "catalog" in context ? (context as { catalog?: string }).catalog : undefined,
schema: context.schema ?? objectBrowser?.schema ?? tableMeta?.schema,
catalog: context.catalog ?? objectBrowser?.catalog ?? tableMeta?.catalog,
shouldRefreshDefaultDatabase: !context.database,
};
}
@ -72,7 +74,7 @@ export interface NewQueryTable {
}
export interface ResolveNewQueryTableInput {
activeTab?: Pick<QueryTab, "mode" | "connectionId" | "database" | "schema" | "tableMeta" | "structureTableName" | "title"> | null;
activeTab?: Pick<QueryTab, "mode" | "connectionId" | "database" | "schema" | "catalog" | "tableMeta" | "structureTableName" | "title"> | null;
selectedTreeNode?: Pick<TreeNode, "type" | "connectionId" | "database" | "schema" | "catalog" | "tableName" | "label"> | null;
preferredSource?: NewQueryContextSource;
}
@ -106,7 +108,7 @@ function tableFromTab(tab: ResolveNewQueryTableInput["activeTab"]): NewQueryTabl
if (tab.mode === "structure") {
const tableName = (tab.structureTableName || "").trim();
if (!tableName) return null;
return { connectionId: tab.connectionId, database: tab.database, schema: tab.schema, tableName };
return { connectionId: tab.connectionId, database: tab.database, schema: tab.schema, catalog: tab.catalog, tableName };
}
return null;
}
@ -137,8 +139,8 @@ export function resolveNewQueryTable(input: ResolveNewQueryTableInput): NewQuery
* the same per-dialect identifier quoting and schema/catalog qualification used
* by the table-data view.
*/
export function buildSelectAllSql(databaseType: DatabaseType | undefined, table: Pick<NewQueryTable, "schema" | "catalog" | "tableName">): string {
const ref = qualifiedTableName({ databaseType, schema: table.schema, catalog: table.catalog, tableName: table.tableName });
export function buildSelectAllSql(databaseType: DatabaseType | undefined, table: Pick<NewQueryTable, "schema" | "catalog" | "tableName"> & Partial<Pick<NewQueryTable, "database">>): string {
const ref = qualifiedTableName({ databaseType, database: table.database, schema: table.schema, catalog: table.catalog, tableName: table.tableName });
return `SELECT * FROM ${ref}`;
}

4
apps/desktop/src/lib/table/tableSqlTemplates.ts
View File

@ -4,6 +4,8 @@ import { qualifiedTableName, quoteTableIdentifier } from "@/lib/table/tableSelec
export interface TableSqlTemplateOptions {
databaseType?: DatabaseType;
schema?: string;
catalog?: string;
database?: string;
tableName: string;
columns?: ColumnInfo[];
tableType?: string;
@ -96,6 +98,8 @@ export function buildTableDeleteTemplate(options: TableSqlTemplateOptions): stri
function templateTableName(options: TableSqlTemplateOptions): string {
return qualifiedTableName({
databaseType: options.databaseType,
catalog: options.catalog,
database: options.database,
schema: options.schema,
tableName: options.tableName,
});

10
apps/desktop/src/stores/__tests__/queryStore.multiStatementError.spec.ts vendored
View File

@ -338,6 +338,16 @@ describe("queryStore multi-statement errors", () => {
expect(mocks.executeMulti).toHaveBeenCalledWith("mysql-1", "app", "SELECT 1", undefined, expect.any(String), expect.objectContaining({ continueOnError: false }));
});
it("passes the selected external catalog to query execution", async () => {
mocks.executeMulti.mockResolvedValue([{ columns: ["value"], rows: [[1]], affected_rows: 0, execution_time_ms: 1 }]);
const { useQueryStore } = await import("@/stores/queryStore");
const store = useQueryStore();
const tabId = store.createTab("mysql-1", "bi", "Query", "query", undefined, undefined, "paimon_catalog");
await store.executeTabSql(tabId, "SELECT * FROM events");
expect(mocks.executeMulti).toHaveBeenCalledWith("mysql-1", "bi", "SELECT * FROM events", undefined, expect.any(String), expect.objectContaining({ catalog: "paimon_catalog" }));
});
it("keeps old and new executions as result runs, then lets normal execution replace the active run", async () => {
mocks.executeMulti

39
apps/desktop/src/stores/__tests__/queryStore.switchTab.spec.ts vendored
View File

@ -1,5 +1,6 @@
import { createPinia, setActivePinia } from "pinia";
import { beforeEach, describe, expect, it, vi } from "vitest";
import { useQueryStore } from "@/stores/queryStore";
describe("queryStore switchTab", () => {
beforeEach(() => {
@ -74,6 +75,44 @@ describe("queryStore switchTab", () => {
expect(settingsStore.settingsPageActive).toBe(false);
});
it("keeps same-named tabs from different catalogs distinct", async () => {
const queryStore = useQueryStore();
const paimonTabId = queryStore.createTab("sr-1", "bi", "events", "query", undefined, undefined, "paimon_catalog");
const internalTabId = queryStore.createTab("sr-1", "bi", "events", "query", undefined, undefined, "internal");
expect(internalTabId).not.toBe(paimonTabId);
expect(queryStore.tabs).toHaveLength(2);
});
it("preserves catalog context when duplicating a query tab", async () => {
const queryStore = useQueryStore();
const tabId = queryStore.createTab("sr-1", "bi", undefined, "query", undefined, "SELECT 1", "paimon_catalog");
queryStore.duplicateTab(tabId);
expect(queryStore.tabs).toHaveLength(2);
expect(queryStore.tabs[1].catalog).toBe("paimon_catalog");
});
it("switches catalog and database as one query context", () => {
const queryStore = useQueryStore();
const tabId = queryStore.createTab("sr-1", "internal_db", undefined, "query");
const tab = queryStore.tabs.find((candidate) => candidate.id === tabId)!;
tab.result = {
columns: ["id"],
rows: [[1]],
affected_rows: 0,
execution_time_ms: 1,
};
queryStore.updateCatalog(tabId, "paimon_catalog", "bi");
expect(tab.catalog).toBe("paimon_catalog");
expect(tab.database).toBe("bi");
expect(tab.result).toBeUndefined();
});
it("stores local data-grid column filters on the tab result", async () => {
const { useQueryStore } = await import("@/stores/queryStore");
const queryStore = useQueryStore();

70
apps/desktop/src/stores/queryStore.ts
View File

@ -538,10 +538,11 @@ export const useQueryStore = defineStore("query", () => {
const sessionId = tab?.resultSessionId ?? tab?.result?.session_id;
if (!tab || !sessionId || sessionId === preserveSessionId) return;
try {
const catalog = tab.mode === "data" ? tab.tableMeta?.catalog : undefined;
const catalog = tab.mode === "data" ? tab.tableMeta?.catalog : tab.catalog;
const connection = catalog ? useConnectionStore().getConfig(tab.connectionId) : undefined;
const executionDatabase = dataTabExecutionDatabase(connection, tab.database, catalog);
await api.closeQuerySession(tab.connectionId, executionDatabase, sessionId, tab.id);
if (catalog) await api.closeQuerySession(tab.connectionId, executionDatabase, sessionId, tab.id, catalog);
else await api.closeQuerySession(tab.connectionId, executionDatabase, sessionId, tab.id);
} catch (error) {
console.warn("[DBX][query-session:close:error]", { tabId: tab.id, sessionId, error });
if (throwOnError) throw error;
@ -555,9 +556,10 @@ export const useQueryStore = defineStore("query", () => {
}
}
async function closeClientSessionId(connectionId: string, database: string, clientSessionId: string, logContext: Record<string, unknown> = {}, throwOnError = false) {
async function closeClientSessionId(connectionId: string, database: string, clientSessionId: string, catalog: string | undefined, logContext: Record<string, unknown> = {}, throwOnError = false) {
try {
await api.closeClientConnectionSession(connectionId, database, clientSessionId);
if (catalog) await api.closeClientConnectionSession(connectionId, database, clientSessionId, catalog);
else await api.closeClientConnectionSession(connectionId, database, clientSessionId);
} catch (error) {
console.warn("[DBX][client-session:close:error]", { ...logContext, clientSessionId, error });
if (throwOnError) throw error;
@ -566,12 +568,12 @@ export const useQueryStore = defineStore("query", () => {
async function closeClientConnectionSession(tab: QueryTab | undefined, throwOnError = false) {
if (!tab?.connectionId) return;
const catalog = tab.mode === "data" ? tab.tableMeta?.catalog : undefined;
const catalog = tab.mode === "data" ? tab.tableMeta?.catalog : tab.catalog;
const connection = catalog ? useConnectionStore().getConfig(tab.connectionId) : undefined;
const executionDatabase = dataTabExecutionDatabase(connection, tab.database, catalog);
const clientSessionIds = [...new Set([tabClientSessionId(tab), ...BACKGROUND_CLIENT_SESSION_SUFFIXES.map((suffix) => tabClientSessionId(tab, suffix)), tab.explainClientSessionId].filter((sessionId): sessionId is string => !!sessionId))];
for (const clientSessionId of clientSessionIds) {
await closeClientSessionId(tab.connectionId, executionDatabase, clientSessionId, { tabId: tab.id }, throwOnError);
await closeClientSessionId(tab.connectionId, executionDatabase, clientSessionId, catalog, { tabId: tab.id }, throwOnError);
}
}
@ -1153,13 +1155,13 @@ export const useQueryStore = defineStore("query", () => {
return saveTabs(tabs.value, activeTabId.value);
}
function findTabByIdentity(connectionId: string, database: string, title: string, mode: QueryTab["mode"], schema?: string) {
return tabs.value.find((tab) => tab.connectionId === connectionId && tab.database === database && tab.title === title && tab.mode === mode && (tab.schema || "") === (schema || ""));
function findTabByIdentity(connectionId: string, database: string, title: string, mode: QueryTab["mode"], schema?: string, catalog?: string) {
return tabs.value.find((tab) => tab.connectionId === connectionId && tab.database === database && tab.title === title && tab.mode === mode && (tab.schema || "") === (schema || "") && (tab.catalog || "") === (catalog || ""));
}
function createTab(connectionId: string, database: string, title?: string, mode: QueryTab["mode"] = "query", schema?: string, initialSql?: string, catalog?: string, options: { forceNew?: boolean } = {}) {
if (title && !options.forceNew) {
const existing = findTabByIdentity(connectionId, database, title, mode, schema);
const existing = findTabByIdentity(connectionId, database, title, mode, schema, catalog);
if (existing) {
switchTab(existing.id);
return existing.id;
@ -1937,6 +1939,7 @@ export const useQueryStore = defineStore("query", () => {
connectionId: original.connectionId,
database: original.database,
schema: original.schema,
catalog: original.catalog,
sql: original.sql,
originalSql: "",
savedSqlId: undefined,
@ -2409,6 +2412,24 @@ export const useQueryStore = defineStore("query", () => {
tab.tableMeta = undefined;
}
function updateCatalog(id: string, catalog: string | undefined, database: string) {
const tab = tabs.value.find((candidate) => candidate.id === id);
if (!tab || (tab.catalog === catalog && tab.database === database)) return;
rollbackTabTransaction(tab);
void closeResultSession(tab);
void closeClientConnectionSession(tab);
tab.catalog = catalog;
tab.database = database;
tab.schema = undefined;
tab.objectBrowser = undefined;
clearResultPayload(tab);
tab.lastExecutedSql = undefined;
tab.resultBaseSql = undefined;
tab.resultSortedSql = undefined;
clearExplain(tab);
tab.tableMeta = undefined;
}
function updateSchema(id: string, schema: string | undefined) {
const tab = tabs.value.find((t) => t.id === id);
if (!tab || tab.schema === schema) return;
@ -2434,6 +2455,8 @@ export const useQueryStore = defineStore("query", () => {
void closeClientConnectionSession(tab);
tab.connectionId = connectionId;
tab.database = database;
tab.catalog = undefined;
tab.objectBrowser = undefined;
tab.schema = undefined;
clearResultPayload(tab);
tab.lastExecutedSql = undefined;
@ -2978,6 +3001,7 @@ export const useQueryStore = defineStore("query", () => {
connectionId: string;
database: string;
schema?: string;
catalog?: string;
countSql?: string;
countSqlTarget?: () => Promise<TotalRowCountSqlTarget | undefined>;
result: QueryResult;
@ -3022,6 +3046,7 @@ export const useQueryStore = defineStore("query", () => {
queryExecutionLog("info", "count:start", { traceId: options.traceId, elapsed: options.elapsed() });
const countResult = await api.executeQuery(options.connectionId, options.database, countTarget.sql, countTarget.schema, countExecutionId, {
clientSessionId,
catalog: options.catalog,
timeoutSecs: options.timeoutSecs,
});
const total = Number(countResult.rows?.[0]?.[0] ?? 0);
@ -3043,7 +3068,7 @@ export const useQueryStore = defineStore("query", () => {
error,
});
} finally {
void closeClientSessionId(options.connectionId, options.database, clientSessionId, { tabId: options.tabId });
void closeClientSessionId(options.connectionId, options.database, clientSessionId, options.catalog, { tabId: options.tabId });
}
})();
}
@ -3678,7 +3703,7 @@ export const useQueryStore = defineStore("query", () => {
if (tab.autoCommit === false) {
if (!tab.txnSessionId) {
queryExecutionLog("info", "begin-manual-txn:start", { traceId, elapsed: elapsed() });
tab.txnSessionId = await api.beginManualTransaction(tab.connectionId, executionDatabase, executionSchema);
tab.txnSessionId = await api.beginManualTransaction(tab.connectionId, executionDatabase, executionSchema, tab.catalog);
queryExecutionLog("info", "begin-manual-txn:done", { traceId, txnSessionId: tab.txnSessionId, elapsed: elapsed() });
}
queryExecutionLog("info", "execute-in-txn:invoke", { traceId, txnSessionId: tab.txnSessionId, elapsed: elapsed() });
@ -3704,6 +3729,7 @@ export const useQueryStore = defineStore("query", () => {
: {}),
...(clientSessionId ? { clientSessionId } : {}),
timeoutSecs: queryTimeoutSecs,
catalog: tab.catalog,
continueOnError: settingsStore.editorSettings.continueOnErrorOnBatch,
};
queryExecutionLog("info", "execute-multi:invoke", {
@ -3839,6 +3865,7 @@ export const useQueryStore = defineStore("query", () => {
connectionId: current.connectionId,
database: executionDatabase,
schema: current.schema,
catalog: current.catalog,
countSql,
countSqlTarget: dataCountTarget
? async () => ({
@ -4084,6 +4111,7 @@ export const useQueryStore = defineStore("query", () => {
try {
tableResult = await api.executeQuery(tab.connectionId, tab.database, tableSql, tab.schema, executionId, {
clientSessionId,
catalog: tab.catalog,
timeoutSecs: queryTimeoutSecs,
});
} catch (error: unknown) {
@ -4095,6 +4123,7 @@ export const useQueryStore = defineStore("query", () => {
try {
tableResult = await api.executeQuery(tab.connectionId, tab.database, tableSql, tab.schema, executionId, {
clientSessionId,
catalog: tab.catalog,
timeoutSecs: queryTimeoutSecs,
});
} catch (fallbackError: unknown) {
@ -4135,6 +4164,7 @@ export const useQueryStore = defineStore("query", () => {
if (latest?.explainExecutionId === executionId) latest.explainSql = jsonBuilt.sql;
const jsonResult = await api.executeQuery(tab.connectionId, tab.database, jsonBuilt.sql, tab.schema, executionId, {
clientSessionId,
catalog: tab.catalog,
timeoutSecs: queryTimeoutSecs,
});
const current = tabs.value.find((t) => t.id === id);
@ -4158,7 +4188,7 @@ export const useQueryStore = defineStore("query", () => {
current.explainExecutionId = undefined;
}
if (current?.explainClientSessionId === clientSessionId) current.explainClientSessionId = undefined;
void closeClientSessionId(tab.connectionId, tab.database, clientSessionId, { tabId: tab.id, explainExecutionId: executionId });
void closeClientSessionId(tab.connectionId, tab.database, clientSessionId, tab.catalog, { tabId: tab.id, explainExecutionId: executionId });
}
return { ok: true as const, sql: tab.explainSql ?? tableSql };
}
@ -4238,7 +4268,7 @@ export const useQueryStore = defineStore("query", () => {
current.explainExecutionId = undefined;
}
if (current?.explainClientSessionId === clientSessionId) current.explainClientSessionId = undefined;
await closeClientSessionId(tab.connectionId, tab.database, clientSessionId, { tabId: tab.id, explainExecutionId: executionId });
await closeClientSessionId(tab.connectionId, tab.database, clientSessionId, tab.catalog, { tabId: tab.id, explainExecutionId: executionId });
}
return { ok: true as const, sql: built.sql };
}
@ -4257,6 +4287,7 @@ export const useQueryStore = defineStore("query", () => {
try {
const result = await api.executeQuery(tab.connectionId, tab.database, built.sql, tab.schema, executionId, {
clientSessionId,
catalog: tab.catalog,
timeoutSecs: queryTimeoutSecs,
});
const current = tabs.value.find((t) => t.id === id);
@ -4276,7 +4307,7 @@ export const useQueryStore = defineStore("query", () => {
current.isExplaining = false;
current.explainExecutionId = undefined;
}
void closeClientSessionId(tab.connectionId, tab.database, clientSessionId, { tabId: tab.id });
void closeClientSessionId(tab.connectionId, tab.database, clientSessionId, tab.catalog, { tabId: tab.id });
}
return { ok: true as const, sql: built.sql };
}
@ -4650,6 +4681,7 @@ export const useQueryStore = defineStore("query", () => {
maxRows: pageLimit,
fetchSize: pageLimit,
clientSessionId,
catalog: tableMeta.catalog,
timeoutSecs: queryTimeoutSecs,
});
const result = results[0];
@ -4662,7 +4694,7 @@ export const useQueryStore = defineStore("query", () => {
offset += result.rows.length;
}
} finally {
void closeClientSessionId(tab.connectionId, executionDatabase, clientSessionId, { tabId: tab.id });
void closeClientSessionId(tab.connectionId, executionDatabase, clientSessionId, tableMeta.catalog, { tabId: tab.id });
}
return {
@ -4722,9 +4754,10 @@ export const useQueryStore = defineStore("query", () => {
pageSize: plan.pageLimit,
resultSessionId: sessionId,
clientSessionId,
catalog: tab.catalog,
timeoutSecs: queryTimeoutSecs,
}
: { maxRows: plan.pageLimit, fetchSize: plan.pageLimit, clientSessionId, timeoutSecs: queryTimeoutSecs };
: { maxRows: plan.pageLimit, fetchSize: plan.pageLimit, clientSessionId, catalog: tab.catalog, timeoutSecs: queryTimeoutSecs };
const results = await api.executeMulti(tab.connectionId, tab.database, plan.sqlToExecute, tab.schema, exportExecutionId, executionOptions);
const result = results[0];
if (!result) break;
@ -4738,8 +4771,8 @@ export const useQueryStore = defineStore("query", () => {
offset += result.rows.length;
}
} finally {
if (sessionId) void api.closeQuerySession(tab.connectionId, tab.database, sessionId, clientSessionId);
void closeClientSessionId(tab.connectionId, tab.database, clientSessionId, { tabId: tab.id });
if (sessionId) void api.closeQuerySession(tab.connectionId, tab.database, sessionId, clientSessionId, tab.catalog);
void closeClientSessionId(tab.connectionId, tab.database, clientSessionId, tab.catalog, { tabId: tab.id });
}
return {
@ -4929,6 +4962,7 @@ export const useQueryStore = defineStore("query", () => {
togglePinnedTab,
reorderTab,
updateDatabase,
updateCatalog,
updateSchema,
updateConnection,
setTableMeta,

50
crates/dbx-core/src/connection.rs
View File

@ -82,6 +82,16 @@ fn oceanbase_mysql_setup_queries(config: &ConnectionConfig) -> Vec<String> {
oceanbase_mysql_query_timeout_sql(config, config.query_timeout_secs).into_iter().collect()
}
fn mysql_pool_setup_queries(config: &ConnectionConfig, url: &str) -> Vec<String> {
let mut queries = oceanbase_mysql_setup_queries(config);
if let Some(dialect) = db::mysql::mysql_catalog_dialect(config.db_type, config.driver_profile.as_deref()) {
if let Some(query) = db::mysql::catalog_setup_query_for_url(dialect, url) {
queries.push(query);
}
}
queries
}
pub enum PoolKind {
Mysql(db::mysql::MySqlPool, MysqlMode),
Postgres(deadpool_postgres::Pool),
@ -363,7 +373,7 @@ pub async fn connect_mysql_metadata_pool(
) -> Result<(db::mysql::MySqlPool, MysqlMode), String> {
let url = connection_url_for_endpoint(db_config, host, port);
let idle_timeout_secs = Some(db_config.idle_timeout_secs);
let extra_setup_queries = oceanbase_mysql_setup_queries(db_config);
let extra_setup_queries = mysql_pool_setup_queries(db_config, &url);
if db_config.needs_bare_mysql() {
return match connect_bare_mysql_pool_with_setup(
db_config,
@ -475,7 +485,7 @@ pub async fn connect_bare_metadata_pool(
max_connections: usize,
) -> Result<db::mysql::MySqlPool, String> {
let url = connection_url_for_endpoint(db_config, host, port);
let extra_setup_queries = oceanbase_mysql_setup_queries(db_config);
let extra_setup_queries = mysql_pool_setup_queries(db_config, &url);
if db_config.effective_database().is_none() {
return connect_bare_mysql_pool_with_setup(
db_config,
@ -1220,7 +1230,7 @@ impl AppState {
&url,
connect_timeout,
mysql_pool_max_connections,
&oceanbase_mysql_setup_queries(&db_config),
&mysql_pool_setup_queries(&db_config, &url),
)
.await?
};
@ -3872,12 +3882,12 @@ async fn detect_ob_oracle_mode(config: &ConnectionConfig, pool: &db::mysql::MySq
mod tests {
use super::{
agent_connect_timeout, connection_remote_endpoint, connection_url_for_endpoint, database_connection_config,
metadata_connection_config, mysql_metadata_fallback_url, oceanbase_mysql_query_timeout_sql,
oceanbase_mysql_setup_queries, prestosql_jdbc_config_for_endpoint, redacted_connection_url_for_endpoint,
redis_sentinel_transport_id, redis_sentinel_transport_prefix, sqlserver_legacy_agent_config,
sqlserver_legacy_driver_error, sqlserver_uses_legacy_driver, task_client_session_id, uses_bare_mysql_pool,
uses_tcp_probe, validate_connection_url_params, validate_h2_database_path, AppState, MysqlMode, PoolKind,
PRESTOSQL_JDBC_DRIVER_CLASS,
metadata_connection_config, mysql_metadata_fallback_url, mysql_pool_setup_queries,
oceanbase_mysql_query_timeout_sql, oceanbase_mysql_setup_queries, prestosql_jdbc_config_for_endpoint,
redacted_connection_url_for_endpoint, redis_sentinel_transport_id, redis_sentinel_transport_prefix,
sqlserver_legacy_agent_config, sqlserver_legacy_driver_error, sqlserver_uses_legacy_driver,
task_client_session_id, uses_bare_mysql_pool, uses_tcp_probe, validate_connection_url_params,
validate_h2_database_path, AppState, MysqlMode, PoolKind, PRESTOSQL_JDBC_DRIVER_CLASS,
};
use crate::agent_connection::{
agent_connect_params, mongo_legacy_error_with_auth_hint, mongo_uses_legacy_driver,
@ -4182,6 +4192,28 @@ mod tests {
assert!(oceanbase_mysql_setup_queries(&config).is_empty());
}
#[test]
fn doris_pool_setup_uses_switch_for_configured_catalog() {
let mut config = mysql_config(Some("bi"));
config.db_type = DatabaseType::Doris;
assert_eq!(
mysql_pool_setup_queries(&config, "mysql://root:secret@localhost:9030/bi?catalog=paimon%5Fcatalog"),
vec!["SWITCH `paimon_catalog`"]
);
}
#[test]
fn starrocks_pool_setup_uses_set_catalog_for_configured_catalog() {
let mut config = mysql_config(Some("bi"));
config.db_type = DatabaseType::StarRocks;
assert_eq!(
mysql_pool_setup_queries(&config, "mysql://root:secret@localhost:9030/bi?catalog=paimon%5Fcatalog"),
vec!["SET CATALOG `paimon_catalog`"]
);
}
#[test]
fn mongo_legacy_retry_covers_old_server_handshake_eof() {
let err = r#"MongoDB connection failed: Kind: Server selection timeout: No available servers. Topology: { Type: Unknown, Servers: [ { Address: db.example.com:27017, Type: Unknown, Error: Kind: I/O error: unexpected end of file } ] }"#;

166
crates/dbx-core/src/db/mysql.rs
View File

@ -26,6 +26,27 @@ use super::file_validator::validate_file_path;
pub type MySqlPool = mysql_async::Pool;
const MYSQL_TCP_KEEPALIVE_MS: u32 = 30_000;
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum MySqlCatalogDialect {
Doris,
StarRocks,
}
pub(crate) fn mysql_catalog_dialect(
db_type: DatabaseType,
driver_profile: Option<&str>,
) -> Option<MySqlCatalogDialect> {
match db_type {
DatabaseType::Doris => Some(MySqlCatalogDialect::Doris),
DatabaseType::StarRocks => Some(MySqlCatalogDialect::StarRocks),
_ => match driver_profile.map(str::to_ascii_lowercase).as_deref() {
Some("doris" | "selectdb") => Some(MySqlCatalogDialect::Doris),
Some("starrocks") => Some(MySqlCatalogDialect::StarRocks),
_ => None,
},
}
}
#[derive(Clone, Copy, Debug, Default)]
pub struct MySqlQueryDialect {
supports_admin_show_results: bool,
@ -1012,7 +1033,6 @@ fn mysql_setup_queries_for_database_with_mode(
setup_mode: MySqlSetupMode,
) -> Vec<String> {
let charset = mysql_connection_charset(url).unwrap_or("utf8mb4");
let catalog = mysql_connection_catalog(url);
let database = setup_database.map(ToOwned::to_owned).or_else(|| mysql_connection_database(url));
let mut queries = Vec::new();
if let Some(database) = database.as_deref() {
@ -1031,22 +1051,51 @@ fn mysql_setup_queries_for_database_with_mode(
if let Some(query) = setup_mode.group_concat_max_len_query() {
queries.push(query);
}
// StarRocks/Doris expose external storage (Paimon, Hive, ...) through a
// catalog. `SET catalog` must run *before* `USE <database>` (the database
// lives in the external catalog and is unknown to the default one).
// mysql_async drains the setup list back-to-front (Vec::pop), so push it
// last to make it execute first. The handshake does not send the database
// as schema (see `mysql_async_url`, which strips the path when a catalog is
// configured), so the connection establishes in the default catalog and
// this setup query is what switches it. The pool re-runs these queries
// after every connection reset, so the catalog stays current.
if let Some(catalog) = catalog.as_deref() {
queries.push(format!("SET catalog = {}", quote_identifier(catalog)));
}
queries.extend(extra_setup_queries.iter().cloned());
queries
}
fn catalog_switch_query(dialect: MySqlCatalogDialect, catalog: &str) -> String {
let catalog = quote_identifier(catalog);
match dialect {
MySqlCatalogDialect::Doris => format!("SWITCH {catalog}"),
MySqlCatalogDialect::StarRocks => format!("SET CATALOG {catalog}"),
}
}
pub(crate) fn catalog_setup_query_for_url(dialect: MySqlCatalogDialect, url: &str) -> Option<String> {
mysql_connection_catalog(url).map(|catalog| catalog_switch_query(dialect, &catalog))
}
pub(crate) fn catalog_database_context_queries(
dialect: Option<MySqlCatalogDialect>,
catalog: Option<&str>,
database: &str,
) -> Result<Vec<String>, String> {
let Some(catalog) = catalog.filter(|value| !value.trim().is_empty()) else {
return Ok(Vec::new());
};
let dialect = dialect.ok_or("Catalog selection is only supported for Doris and StarRocks")?;
let mut queries = Vec::with_capacity(2);
queries.push(catalog_switch_query(dialect, catalog));
if !database.trim().is_empty() {
queries.push(format!("USE {}", quote_identifier(database)));
}
Ok(queries)
}
pub(crate) async fn apply_catalog_database_context(
conn: &mut mysql_async::Conn,
dialect: Option<MySqlCatalogDialect>,
catalog: Option<&str>,
database: &str,
) -> Result<(), String> {
for query in catalog_database_context_queries(dialect, catalog, database)? {
conn.query_drop(&query).await.map_err(|error| format!("Failed to select query catalog/database: {error}"))?;
}
Ok(())
}
fn should_enable_explicit_timestamp_defaults(sql: &str) -> bool {
if !starts_with_executable_sql_keyword(sql, &["CREATE", "ALTER"]) {
return false;
@ -1125,8 +1174,8 @@ fn mysql_connection_database(url: &str) -> Option<String> {
/// Extracts an opt-in `catalog=<name>` URL parameter. dbx strips it from the
/// URL before handing it to mysql_async (see `is_dbx_handled_mysql_url_param`)
/// and instead emits `SET catalog = <name>` during connection setup. This is
/// how StarRocks/Doris connections reach an external catalog such as Paimon.
/// and emits the database-specific catalog switch during connection setup.
/// This is how StarRocks/Doris connections reach an external catalog such as Paimon.
fn mysql_connection_catalog(url: &str) -> Option<String> {
let (_, query) = url.split_once('?')?;
let query = query.split('#').next().unwrap_or(query);
@ -1594,7 +1643,7 @@ fn mysql_url_param_value_is_true(value: &str) -> bool {
/// Strips the database path from a `mysql://[user[:pass]@]host[:port][/path]`
/// URL, returning only the scheme and authority. Used so mysql_async does not
/// send the database as the schema during the MySQL handshake (StarRocks would
/// reject an external-catalog database before `SET catalog` runs in setup).
/// reject an external-catalog database before the catalog switch runs in setup).
fn strip_mysql_url_path(base: &str) -> &str {
let Some(rest) = base.strip_prefix("mysql://") else {
return base;
@ -4111,6 +4160,32 @@ mod tests {
use super::*;
use crate::db::connection_timeout;
use mysql_async::consts::ColumnFlags;
#[test]
fn catalog_database_context_uses_database_specific_syntax_before_database() {
assert_eq!(
catalog_database_context_queries(Some(MySqlCatalogDialect::Doris), Some("paimon`catalog"), "bi").unwrap(),
vec!["SWITCH `paimon``catalog`", "USE `bi`"]
);
assert_eq!(
catalog_database_context_queries(Some(MySqlCatalogDialect::StarRocks), Some("paimon`catalog"), "bi")
.unwrap(),
vec!["SET CATALOG `paimon``catalog`", "USE `bi`"]
);
assert_eq!(
catalog_database_context_queries(Some(MySqlCatalogDialect::Doris), None, "").unwrap(),
Vec::<String>::new()
);
assert!(catalog_database_context_queries(None, Some("paimon_catalog"), "bi").is_err());
}
#[test]
fn catalog_dialect_supports_native_and_profile_connections() {
assert_eq!(mysql_catalog_dialect(DatabaseType::Doris, None), Some(MySqlCatalogDialect::Doris));
assert_eq!(mysql_catalog_dialect(DatabaseType::StarRocks, None), Some(MySqlCatalogDialect::StarRocks));
assert_eq!(mysql_catalog_dialect(DatabaseType::Mysql, Some("selectdb")), Some(MySqlCatalogDialect::Doris));
assert_eq!(mysql_catalog_dialect(DatabaseType::Mysql, Some("STARROCKS")), Some(MySqlCatalogDialect::StarRocks));
assert_eq!(mysql_catalog_dialect(DatabaseType::Mysql, None), None);
}
fn mysql_test_object(name: &str, object_type: &str) -> ObjectInfo {
ObjectInfo {
@ -5316,7 +5391,7 @@ mod tests {
}
#[test]
fn mysql_compatible_setup_queries_keep_catalog_and_extra_queries() {
fn mysql_compatible_setup_queries_leave_catalog_to_database_specific_setup() {
let extra = vec!["SET ob_query_timeout = 30000000".to_string()];
let queries = mysql_setup_queries_with_mode(
"mysql://root:secret@localhost:9030/clip?catalog=paimon_catalog",
@ -5324,15 +5399,19 @@ mod tests {
MySqlSetupMode::Compatible,
);
assert_eq!(
queries,
vec![
"USE `clip`",
"SET NAMES utf8mb4",
"SET catalog = `paimon_catalog`",
"SET ob_query_timeout = 30000000"
]
assert_eq!(queries, vec!["USE `clip`", "SET NAMES utf8mb4", "SET ob_query_timeout = 30000000"]);
}
#[test]
fn mysql_setup_appends_database_specific_catalog_for_reverse_execution() {
let extra = vec!["SWITCH `paimon_catalog`".to_string()];
let queries = mysql_setup_queries_with_mode(
"mysql://root:secret@localhost:9030/clip?catalog=paimon_catalog",
&extra,
MySqlSetupMode::Compatible,
);
assert_eq!(queries, vec!["USE `clip`", "SET NAMES utf8mb4", "SWITCH `paimon_catalog`"]);
}
#[test]
@ -5703,38 +5782,21 @@ mod tests {
}
#[test]
fn mysql_setup_queries_switch_catalog_when_present() {
// `SET catalog` is pushed last so mysql_async's back-to-front setup
// execution (Vec::pop) runs it before `USE <database>`.
fn catalog_setup_query_for_url_uses_database_specific_syntax() {
assert_eq!(
mysql_setup_queries("mysql://host:3306/clip?catalog=paimon_catalog", &[]),
vec![
"USE `clip`",
"SET NAMES utf8mb4",
"SET SESSION group_concat_max_len = 1048576",
"SET catalog = `paimon_catalog`"
]
catalog_setup_query_for_url(MySqlCatalogDialect::Doris, "mysql://host:3306/clip?catalog=paimon_catalog"),
Some("SWITCH `paimon_catalog`".to_string())
);
}
#[test]
fn mysql_setup_queries_switch_catalog_without_database() {
assert_eq!(
mysql_setup_queries("mysql://host:3306/?catalog=paimon_catalog", &[]),
vec!["SET NAMES utf8mb4", "SET SESSION group_concat_max_len = 1048576", "SET catalog = `paimon_catalog`"]
catalog_setup_query_for_url(
MySqlCatalogDialect::StarRocks,
"mysql://host:3306/clip?catalog=paimon_catalog"
),
Some("SET CATALOG `paimon_catalog`".to_string())
);
}
#[test]
fn mysql_setup_queries_decodes_catalog_parameter() {
assert_eq!(
mysql_setup_queries("mysql://host:3306/db?catalog=my%5Fcatalog", &[]),
vec![
"USE `db`",
"SET NAMES utf8mb4",
"SET SESSION group_concat_max_len = 1048576",
"SET catalog = `my_catalog`"
]
catalog_setup_query_for_url(MySqlCatalogDialect::Doris, "mysql://host:3306/db?catalog=my%5Fcatalog"),
Some("SWITCH `my_catalog`".to_string())
);
}

199
crates/dbx-core/src/query.rs
View File

@ -221,6 +221,25 @@ async fn connection_mysql_query_dialect(state: &AppState, connection_id: &str) -
.unwrap_or_default()
}
async fn connection_mysql_catalog_dialect(
state: &AppState,
connection_id: &str,
) -> Option<db::mysql::MySqlCatalogDialect> {
let configs = state.configs.read().await;
configs
.get(connection_id)
.and_then(|config| db::mysql::mysql_catalog_dialect(config.db_type, config.driver_profile.as_deref()))
}
async fn connection_mysql_catalog_dialect_for_pool_key(
state: &AppState,
pool_key: &str,
) -> Option<db::mysql::MySqlCatalogDialect> {
let configs = state.configs.read().await;
crate::connection::config_for_pool_key(pool_key, &configs)
.and_then(|config| db::mysql::mysql_catalog_dialect(config.db_type, config.driver_profile.as_deref()))
}
async fn connection_database_type_for_pool_key(state: &AppState, pool_key: &str) -> Option<DatabaseType> {
let configs = state.configs.read().await;
configs
@ -401,6 +420,8 @@ pub struct QueryExecutionOptions {
pub max_rows: Option<usize>,
pub fetch_size: Option<usize>,
pub page_size: Option<usize>,
/// Doris / StarRocks catalog selected for this query tab.
pub catalog: Option<String>,
pub result_session_id: Option<String>,
pub client_session_id: Option<String>,
/// Query timeout in seconds. `None` uses the default (30s).
@ -1245,6 +1266,14 @@ fn resolve_query_timeout(timeout_secs: Option<u64>) -> Option<Duration> {
}
}
fn query_pool_database<'a>(database: &'a str, catalog: Option<&str>) -> Option<&'a str> {
if database.is_empty() || catalog.is_some() {
None
} else {
Some(database)
}
}
pub async fn operation_budget_for_pool_key(
state: &AppState,
pool_key: &str,
@ -1317,6 +1346,7 @@ pub async fn do_execute(
crate::query_execution_sql::check_read_only(sql, &name, database_type)?;
}
let pool_db_type = connection_database_type_for_pool_key(state, pool_key).await;
let mysql_catalog_dialect = connection_mysql_catalog_dialect_for_pool_key(state, pool_key).await;
let connections = state.connections.read().await;
let pool = connections.get(pool_key).ok_or("Connection not found")?;
@ -1424,6 +1454,17 @@ pub async fn do_execute(
});
}
apply_oceanbase_mysql_session_timeout(state, pool_key, &mut conn, options.timeout_secs).await?;
wait_for_result_opt(
cancel_token.clone(),
query_timeout,
db::mysql::apply_catalog_database_context(
&mut conn,
mysql_catalog_dialect,
options.catalog.as_deref(),
database.unwrap_or_default(),
),
)
.await?;
wait_for_query_opt(
cancel_token,
query_timeout,
@ -1826,17 +1867,11 @@ pub async fn execute_sql_statement_with_options(
// When a query tab has a client session, keep even database-less execution
// on that tab-scoped pool so connection-level state (for example MySQL @vars)
// survives across runs.
let pool_key = if database.is_empty() {
state
.get_or_create_pool_for_session(connection_id, None, options.client_session_id.as_deref())
.await
.map_err(|e| query_error_with_omitted_sql_context(&e, sql))?
} else {
state
.get_or_create_pool_for_session(connection_id, Some(database), options.client_session_id.as_deref())
.await
.map_err(|e| query_error_with_omitted_sql_context(&e, sql))?
};
let pool_database = query_pool_database(database, options.catalog.as_deref());
let pool_key = state
.get_or_create_pool_for_session(connection_id, pool_database, options.client_session_id.as_deref())
.await
.map_err(|e| query_error_with_omitted_sql_context(&e, sql))?;
if is_canceled(&cancel_token) {
return Err(canceled_error());
@ -1853,9 +1888,9 @@ pub async fn execute_sql_statement_with_options(
let action = result.as_ref().err().map(|e| query_pool_error_action(db_type, sql, e));
match action {
Some(PoolErrorAction::ReconnectAndRetry) if !is_canceled(&cancel_token) => {
let db_opt = if database.is_empty() { None } else { Some(database) };
let pool_database = query_pool_database(database, options.catalog.as_deref());
let new_key = state
.reconnect_pool_for_session(connection_id, db_opt, options.client_session_id.as_deref())
.reconnect_pool_for_session(connection_id, pool_database, options.client_session_id.as_deref())
.await
.map_err(|e| query_error_with_omitted_sql_context(&e, sql))?;
with_sql_context(
@ -1952,12 +1987,10 @@ pub async fn close_query_session(
database: &str,
session_id: &str,
client_session_id: Option<&str>,
catalog: Option<&str>,
) -> Result<bool, String> {
let pool_key = if database.is_empty() {
state.get_or_create_pool_for_session(connection_id, None, client_session_id).await?
} else {
state.get_or_create_pool_for_session(connection_id, Some(database), client_session_id).await?
};
let pool_database = query_pool_database(database, catalog);
let pool_key = state.get_or_create_pool_for_session(connection_id, pool_database, client_session_id).await?;
let connections = state.connections.read().await;
let pool = connections.get(&pool_key).ok_or("Connection not found")?;
@ -2050,22 +2083,16 @@ pub async fn execute_multi_core_with_options_for_client_and_progress(
options: QueryExecutionOptions,
progress: Option<ExecuteMultiProgressCallback>,
) -> Result<Vec<ExecuteMultiResult>, String> {
let pool_database = query_pool_database(database, options.catalog.as_deref());
// Reject MongoDB queries that fall through to the generic executor.
if connection_is_mongodb(state, connection_id).await {
return Err(MONGO_SHELL_COMMAND_HINT.to_string());
}
let pool_key = if database.is_empty() {
state
.get_or_create_pool_for_session(connection_id, None, options.client_session_id.as_deref())
.await
.map_err(|e| query_error_with_omitted_sql_context(&e, sql))?
} else {
state
.get_or_create_pool_for_session(connection_id, Some(database), options.client_session_id.as_deref())
.await
.map_err(|e| query_error_with_omitted_sql_context(&e, sql))?
};
let pool_key = state
.get_or_create_pool_for_session(connection_id, pool_database, options.client_session_id.as_deref())
.await
.map_err(|e| query_error_with_omitted_sql_context(&e, sql))?;
if let Some(execution_id) = options.execution_id.as_deref() {
state.running_queries.set_pool_key(execution_id, pool_key.clone());
}
@ -2111,7 +2138,15 @@ pub async fn execute_multi_core_with_options_for_client_and_progress(
// When use_transaction is explicitly true and we have multiple statements,
// route through the transaction wrapper instead of the sequential auto-commit loop.
if options.use_transaction == Some(true) && statements.len() > 1 {
let result = execute_statements_in_transaction(state, connection_id, database, &statements, schema).await?;
let result = execute_statements_in_transaction(
state,
connection_id,
database,
&statements,
schema,
options.catalog.as_deref(),
)
.await?;
return Ok(vec![result.into()]);
}
@ -2142,6 +2177,7 @@ pub async fn execute_multi_core_with_options_for_client_and_progress(
// Read-only check for MySQL batch path
check_read_only_for_connection_multi(state, &pool_key, &statements).await?;
let mysql_dialect = connection_mysql_query_dialect(state, connection_id).await;
let mysql_catalog_dialect = connection_mysql_catalog_dialect(state, connection_id).await;
return execute_multi_mysql(
state,
&pool_key,
@ -2149,6 +2185,8 @@ pub async fn execute_multi_core_with_options_for_client_and_progress(
&pool,
mode,
mysql_dialect,
mysql_catalog_dialect,
database,
&statements,
cancel_token,
options,
@ -2269,6 +2307,7 @@ where
(results, None)
}
#[allow(clippy::too_many_arguments)]
async fn execute_multi_mysql(
state: &AppState,
pool_key: &str,
@ -2276,6 +2315,8 @@ async fn execute_multi_mysql(
pool: &db::mysql::MySqlPool,
mode: crate::connection::MysqlMode,
dialect: db::mysql::MySqlQueryDialect,
catalog_dialect: Option<db::mysql::MySqlCatalogDialect>,
database: &str,
statements: &[String],
cancel_token: Option<CancellationToken>,
options: QueryExecutionOptions,
@ -2304,6 +2345,12 @@ async fn execute_multi_mysql(
}
};
apply_oceanbase_mysql_session_timeout(state, pool_key, &mut conn, options.timeout_secs).await?;
wait_for_result_opt(
cancel_token.clone(),
query_timeout,
db::mysql::apply_catalog_database_context(&mut conn, catalog_dialect, options.catalog.as_deref(), database),
)
.await?;
let mut executor = MysqlBatchConnection {
conn: &mut conn,
@ -2796,7 +2843,9 @@ pub async fn execute_schema_diff_deploy(
};
let atomicity = classify_schema_diff_atomicity(db_type, &parsed, has_transactional_path);
match execute_statements_in_transaction_on_pool(state, &pool_key, connection_id, database, &parsed, schema).await {
match execute_statements_in_transaction_on_pool(state, &pool_key, connection_id, database, &parsed, schema, None)
.await
{
Ok(result) => SchemaDiffDeployResult {
transaction_id: tx_id,
status: crate::two_phase_commit::TransactionStatus::Committed.as_str().to_string(),
@ -2852,18 +2901,17 @@ pub async fn execute_statements_in_transaction(
database: &str,
statements: &[String],
schema: Option<&str>,
catalog: Option<&str>,
) -> Result<db::QueryResult, String> {
let sql_ctx = statements.first().map(|s| s.as_str()).unwrap_or("");
let pool_key = if database.is_empty() {
connection_id.to_string()
} else {
state
.get_or_create_pool(connection_id, Some(database))
.await
.map_err(|e| query_error_with_omitted_sql_context(&e, sql_ctx))?
};
let pool_database = query_pool_database(database, catalog);
let pool_key = state
.get_or_create_pool(connection_id, pool_database)
.await
.map_err(|e| query_error_with_omitted_sql_context(&e, sql_ctx))?;
execute_statements_in_transaction_on_pool(state, &pool_key, connection_id, database, statements, schema).await
execute_statements_in_transaction_on_pool(state, &pool_key, connection_id, database, statements, schema, catalog)
.await
}
/// Execute multiple SQL statements transactionally on an already-resolved pool.
@ -2875,12 +2923,14 @@ pub async fn execute_statements_in_transaction_on_pool(
database: &str,
statements: &[String],
schema: Option<&str>,
catalog: Option<&str>,
) -> Result<db::QueryResult, String> {
// Read-only check: intercept all transaction paths before dispatching
check_read_only_for_connection_multi(state, pool_key, statements).await?;
let start = std::time::Instant::now();
let db_type = connection_database_type(state, connection_id).await;
let mysql_catalog_dialect = connection_mysql_catalog_dialect(state, connection_id).await;
let operation_budget = configured_operation_budget_for_pool_key(state, pool_key).await;
// Clone the pool handle within the lock, then drop it before any async work.
@ -2926,7 +2976,18 @@ pub async fn execute_statements_in_transaction_on_pool(
exec_tx_pg_inner(pool, statements, schema, start, operation_budget.clone(), cancel_context).await
}
Some(TxPath::Mysql(pool, _bare)) => {
exec_tx_mysql_inner(state, pool_key, pool, statements, start, operation_budget.clone()).await
exec_tx_mysql_inner(
state,
pool_key,
pool,
statements,
start,
operation_budget.clone(),
mysql_catalog_dialect,
catalog,
database,
)
.await
}
Some(TxPath::Sqlite(pool)) => exec_tx_sqlite_inner(pool, statements, start).await,
Some(TxPath::CloudflareD1(client)) => {
@ -3068,9 +3129,13 @@ async fn exec_tx_mysql_inner(
statements: &[String],
start: std::time::Instant,
budget: DbOperationBudget,
catalog_dialect: Option<db::mysql::MySqlCatalogDialect>,
catalog: Option<&str>,
database: &str,
) -> Result<db::QueryResult, String> {
let mut conn = db::mysql::get_conn_with_health_check_with_timeout(&pool, budget.checkout_timeout).await?;
apply_oceanbase_mysql_session_timeout(state, pool_key, &mut conn, None).await?;
db::mysql::apply_catalog_database_context(&mut conn, catalog_dialect, catalog, database).await?;
mysql_query_drop_with_timeout(
&mut conn,
"START TRANSACTION",
@ -3308,8 +3373,9 @@ pub async fn begin_manual_transaction(
connection_id: &str,
database: &str,
schema: Option<&str>,
catalog: Option<&str>,
) -> Result<String, String> {
begin_transaction_session(state, connection_id, database, schema, false).await
begin_transaction_session(state, connection_id, database, schema, catalog, false).await
}
/// Start a read-only, repeatable snapshot for a database backup.
@ -3318,7 +3384,7 @@ pub async fn begin_database_backup_snapshot(
connection_id: &str,
database: &str,
) -> Result<String, String> {
begin_transaction_session(state, connection_id, database, None, true).await
begin_transaction_session(state, connection_id, database, None, None, true).await
}
fn postgres_transaction_begin_sql(consistent_snapshot: bool) -> &'static str {
@ -3354,13 +3420,12 @@ async fn begin_transaction_session(
connection_id: &str,
database: &str,
schema: Option<&str>,
catalog: Option<&str>,
consistent_snapshot: bool,
) -> Result<String, String> {
let pool_key = if database.is_empty() {
connection_id.to_string()
} else {
state.get_or_create_pool(connection_id, Some(database)).await?
};
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?;
// 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.
@ -3397,6 +3462,7 @@ async fn begin_transaction_session(
}
TxnPoolHandle::Mysql(mysql_pool) => {
let mut conn = mysql_pool.get_conn().await.map_err(|e| format!("Failed to get MySQL connection: {e}"))?;
db::mysql::apply_catalog_database_context(&mut conn, mysql_catalog_dialect, catalog, database).await?;
if let Some(isolation_sql) = mysql_transaction_isolation_sql(consistent_snapshot) {
conn.query_drop(isolation_sql).await.map_err(|e| format!("SET TRANSACTION failed: {e}"))?;
}
@ -3889,6 +3955,43 @@ mod tests {
};
use crate::storage::Storage;
#[test]
fn external_catalog_queries_do_not_bind_database_during_pool_creation() {
assert_eq!(query_pool_database("bi", Some("paimon_catalog")), None);
assert_eq!(query_pool_database("bi", None), Some("bi"));
assert_eq!(query_pool_database("", None), None);
}
#[tokio::test]
async fn query_and_transaction_paths_resolve_catalog_dialect_from_connection() {
let dir = std::env::temp_dir().join(format!("dbx-catalog-dialect-{}", uuid::Uuid::new_v4()));
std::fs::create_dir_all(&dir).unwrap();
let storage = Storage::open(&dir.join("storage.db")).await.unwrap();
let state = AppState::new(storage);
let mut doris = test_connection_config(DatabaseType::Doris);
doris.id = "doris".to_string();
let mut starrocks = test_connection_config(DatabaseType::StarRocks);
starrocks.id = "starrocks".to_string();
{
let mut configs = state.configs.write().await;
configs.insert(doris.id.clone(), doris);
configs.insert(starrocks.id.clone(), starrocks);
}
assert_eq!(
connection_mysql_catalog_dialect_for_pool_key(&state, "doris:bi").await,
Some(db::mysql::MySqlCatalogDialect::Doris)
);
assert_eq!(
connection_mysql_catalog_dialect(&state, "starrocks").await,
Some(db::mysql::MySqlCatalogDialect::StarRocks)
);
drop(state);
let _ = std::fs::remove_dir_all(dir);
}
#[test]
fn schema_diff_atomicity_marks_mysql_ddl_as_partial() {
let atomicity = classify_schema_diff_atomicity(

1
crates/dbx-core/src/query_result_export.rs
View File

@ -588,6 +588,7 @@ pub async fn export_query_result_core(
&request.database,
&session_id,
request.client_session_id.as_deref(),
None,
)
.await;
}

1
crates/dbx-core/src/table_export.rs
View File

@ -582,6 +582,7 @@ async fn close_table_export_cursor_if_open(
&request.database,
&session_id,
Some(&client_session_id),
None,
)
.await;
}

1
crates/dbx-core/src/table_import.rs
View File

@ -3744,6 +3744,7 @@ async fn execute_import_transaction(
database,
statements,
(!schema.trim().is_empty()).then_some(schema),
None,
)
.await;
*db_write_ms += started_at.elapsed().as_millis();

4
crates/dbx-core/tests/live_manual_transaction.rs
View File

@ -47,7 +47,7 @@ async fn live_manual_transaction_postgres_preserves_typed_selects_and_empty_meta
let database = config.database.clone().expect("database");
let (state, db_path) = app_state_with_config(config.clone()).await;
let txn = begin_manual_transaction(&state, &config.id, &database, None).await.expect("begin");
let txn = begin_manual_transaction(&state, &config.id, &database, None, None).await.expect("begin");
let typed = execute_in_manual_transaction(
&state,
&txn,
@ -85,7 +85,7 @@ async fn live_manual_transaction_mysql_streams_with_row_limit() {
let database = config.database.clone().expect("database");
let (state, db_path) = app_state_with_config(config.clone()).await;
let txn = begin_manual_transaction(&state, &config.id, &database, None).await.expect("begin");
let txn = begin_manual_transaction(&state, &config.id, &database, None, None).await.expect("begin");
let limited = execute_in_manual_transaction(
&state,
&txn,

20
crates/dbx-web/src/routes/query.rs
View File

@ -16,6 +16,7 @@ pub struct ExecuteQueryRequest {
pub database: String,
pub sql: String,
pub schema: Option<String>,
pub catalog: Option<String>,
pub execution_id: Option<String>,
pub max_rows: Option<usize>,
pub fetch_size: Option<usize>,
@ -40,6 +41,7 @@ pub struct CloseSessionRequest {
pub connection_id: String,
pub database: String,
pub session_id: String,
pub catalog: Option<String>,
pub client_session_id: Option<String>,
}
@ -48,6 +50,7 @@ pub struct CloseSessionRequest {
pub struct CloseClientConnectionSessionRequest {
pub connection_id: String,
pub database: String,
pub catalog: Option<String>,
pub client_session_id: String,
}
@ -58,6 +61,7 @@ pub struct ExecuteBatchRequest {
pub database: String,
pub statements: Vec<String>,
pub schema: Option<String>,
pub catalog: Option<String>,
pub timeout_secs: Option<u64>,
}
@ -349,6 +353,7 @@ pub async fn execute_query(
max_rows: req.max_rows,
fetch_size: req.fetch_size,
page_size: req.page_size,
catalog: req.catalog,
result_session_id: req.result_session_id,
client_session_id: req.client_session_id,
timeout_secs: req.timeout_secs,
@ -392,6 +397,7 @@ pub async fn execute_multi(
max_rows: req.max_rows,
fetch_size: req.fetch_size,
page_size: req.page_size,
catalog: req.catalog,
result_session_id: req.result_session_id,
client_session_id: req.client_session_id,
timeout_secs: req.timeout_secs,
@ -449,6 +455,7 @@ pub async fn close_query_session(
&req.database,
&req.session_id,
req.client_session_id.as_deref(),
req.catalog.as_deref(),
)
.await
.map_err(AppError::from)?;
@ -460,7 +467,7 @@ pub async fn close_client_connection_session(
State(state): State<Arc<WebState>>,
Json(req): Json<CloseClientConnectionSessionRequest>,
) -> Result<Json<serde_json::Value>, AppError> {
let database = if req.database.trim().is_empty() { None } else { Some(req.database.as_str()) };
let database = query_session_database(&req.database, req.catalog.as_deref());
let closed = state
.app
.close_client_session_pool(&req.connection_id, database, &req.client_session_id)
@ -469,6 +476,13 @@ pub async fn close_client_connection_session(
Ok(Json(serde_json::json!(closed)))
}
fn query_session_database<'a>(database: &'a str, catalog: Option<&str>) -> Option<&'a str> {
if database.trim().is_empty() || catalog.is_some() {
None
} else {
Some(database)
}
}
pub async fn execute_script(
State(state): State<Arc<WebState>>,
@ -507,6 +521,7 @@ pub async fn execute_in_transaction(
&req.database,
&req.statements,
req.schema.as_deref(),
req.catalog.as_deref(),
)
.await
.map_err(AppError::from)?;
@ -926,6 +941,7 @@ mod tests {
database: "testdb".to_string(),
statements: vec!["SELECT 1".to_string()],
schema: None,
catalog: None,
timeout_secs: None,
};
@ -949,6 +965,7 @@ mod tests {
database: "testdb".to_string(),
statements: vec![],
schema: None,
catalog: None,
timeout_secs: None,
};
@ -969,6 +986,7 @@ mod tests {
database: "testdb".to_string(),
statements: vec!["CREATE TABLE t1 (id INT)".to_string(), "CREATE TABLE t2 (id INT)".to_string()],
schema: None,
catalog: None,
timeout_secs: None,
};

33
src-tauri/src/commands/query.rs
View File

@ -25,6 +25,7 @@ pub async fn execute_query(
database: String,
sql: String,
schema: Option<String>,
catalog: Option<String>,
execution_id: Option<String>,
max_rows: Option<usize>,
fetch_size: Option<usize>,
@ -54,6 +55,7 @@ pub async fn execute_query(
max_rows,
fetch_size,
page_size,
catalog,
result_session_id,
client_session_id,
timeout_secs,
@ -74,6 +76,7 @@ pub async fn execute_multi(
database: String,
sql: String,
schema: Option<String>,
catalog: Option<String>,
execution_id: Option<String>,
max_rows: Option<usize>,
fetch_size: Option<usize>,
@ -125,6 +128,7 @@ pub async fn execute_multi(
max_rows,
fetch_size,
page_size,
catalog,
result_session_id,
client_session_id,
timeout_secs,
@ -167,9 +171,17 @@ pub async fn close_query_session(
database: String,
session_id: String,
client_session_id: Option<String>,
catalog: Option<String>,
) -> Result<bool, String> {
dbx_core::query::close_query_session(&state, &connection_id, &database, &session_id, client_session_id.as_deref())
.await
dbx_core::query::close_query_session(
&state,
&connection_id,
&database,
&session_id,
client_session_id.as_deref(),
catalog.as_deref(),
)
.await
}
#[tauri::command]
@ -178,11 +190,20 @@ pub async fn close_client_connection_session(
connection_id: String,
database: String,
client_session_id: String,
catalog: Option<String>,
) -> Result<bool, String> {
let database = if database.trim().is_empty() { None } else { Some(database.as_str()) };
let database = query_session_database(&database, catalog.as_deref());
state.close_client_session_pool(&connection_id, database, &client_session_id).await
}
fn query_session_database<'a>(database: &'a str, catalog: Option<&str>) -> Option<&'a str> {
if database.trim().is_empty() || catalog.is_some() {
None
} else {
Some(database)
}
}
#[tauri::command]
pub async fn execute_batch(
state: State<'_, Arc<AppState>>,
@ -230,6 +251,7 @@ pub async fn execute_in_transaction(
database: String,
statements: Vec<String>,
schema: Option<String>,
catalog: Option<String>,
) -> Result<db::QueryResult, String> {
dbx_core::query::execute_statements_in_transaction(
&state,
@ -237,6 +259,7 @@ pub async fn execute_in_transaction(
&database,
&statements,
schema.as_deref(),
catalog.as_deref(),
)
.await
}
@ -274,8 +297,10 @@ pub async fn begin_manual_transaction(
connection_id: String,
database: String,
schema: Option<String>,
catalog: Option<String>,
) -> Result<String, String> {
dbx_core::query::begin_manual_transaction(&state, &connection_id, &database, schema.as_deref()).await
dbx_core::query::begin_manual_transaction(&state, &connection_id, &database, schema.as_deref(), catalog.as_deref())
.await
}
#[tauri::command]