From adb6d86b536079a3ba0320df142807efedf755ea Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=A6=82=E6=A2=A6=E6=8A=80=E6=9C=AF?= <596392912@qq.com> Date: Thu, 30 Jul 2026 16:03:32 +0800 Subject: [PATCH] feat(database): support Doris and StarRocks catalogs --- apps/desktop/src/App.vue | 1 + .../src/components/layout/AppDialogs.vue | 1 + .../transfer/DataTransferDialog.vue | 198 +++++- .../__tests__/DataTransferDialog.spec.ts | 9 +- .../__tests__/useDatabaseOptions.spec.ts | 14 + .../__tests__/useExportTracker.spec.ts | 20 + .../src/composables/useDatabaseOptions.ts | 17 +- .../src/composables/useDialogSources.ts | 4 + .../src/composables/useExportTracker.ts | 3 + .../composables/useSidebarTreeToolRuntime.ts | 1 + apps/desktop/src/i18n/locales/en.ts | 4 + apps/desktop/src/i18n/locales/es.ts | 4 + apps/desktop/src/i18n/locales/it.ts | 4 + apps/desktop/src/i18n/locales/ja.ts | 4 + apps/desktop/src/i18n/locales/pt-BR.ts | 4 + apps/desktop/src/i18n/locales/zh-CN.ts | 4 + apps/desktop/src/i18n/locales/zh-TW.ts | 4 + .../database/dataTransferSelection.spec.ts | 38 ++ apps/desktop/src/lib/backend/tauri.ts | 2 + .../src/lib/database/dataTransferSelection.ts | 20 + .../src/lib/sidebar/treeRefreshTarget.ts | 7 +- .../connectionStore.dorisCatalog.spec.ts | 126 ++++ .../queryStore.database-open.spec.ts | 15 + apps/desktop/src/stores/connectionStore.ts | 32 +- apps/desktop/src/stores/queryStore.ts | 1 + .../assets/database-drivers.manifest.json | 6 +- crates/dbx-core/src/connection.rs | 142 ++++- crates/dbx-core/src/data_compare.rs | 1 + crates/dbx-core/src/database_capabilities.rs | 9 +- crates/dbx-core/src/database_export.rs | 13 +- .../dbx-core/src/db/cloudflare_d1/import.rs | 2 +- .../src/db/cloudflare_d1/transfer_tests.rs | 1 + crates/dbx-core/src/schema.rs | 5 +- .../dbx-core/src/sql_dialect/identifiers.rs | 23 +- crates/dbx-core/src/sql_dialect/tests.rs | 4 +- crates/dbx-core/src/table_export.rs | 3 +- crates/dbx-core/src/table_import.rs | 15 +- crates/dbx-core/src/transfer.rs | 589 ++++++++++++++---- .../dbx-core/tests/database_capabilities.rs | 5 +- .../dbx-core/tests/live_postgres_transfer.rs | 4 + .../tests/live_sqlserver_completion.rs | 2 + crates/dbx-web/src/routes/transfer.rs | 87 ++- packages/app-tests/treeRefreshTarget.test.ts | 39 ++ src-tauri/src/commands/transfer.rs | 85 ++- 44 files changed, 1330 insertions(+), 242 deletions(-) create mode 100644 apps/desktop/src/lib/__tests__/database/dataTransferSelection.spec.ts create mode 100644 apps/desktop/src/lib/database/dataTransferSelection.ts create mode 100644 apps/desktop/src/stores/__tests__/connectionStore.dorisCatalog.spec.ts diff --git a/apps/desktop/src/App.vue b/apps/desktop/src/App.vue index 35a876a360..29ca624f85 100644 --- a/apps/desktop/src/App.vue +++ b/apps/desktop/src/App.vue @@ -1553,6 +1553,7 @@ function changeActiveSchema(schema: string | undefined) { const tab = activeTab.value; if (tab) queryStore.updateSchema(tab.id, schema); } + function openGitHub() { openUrl("https://github.com/t8y2/dbx"); } diff --git a/apps/desktop/src/components/layout/AppDialogs.vue b/apps/desktop/src/components/layout/AppDialogs.vue index 717d9ff8fe..a42616efad 100644 --- a/apps/desktop/src/components/layout/AppDialogs.vue +++ b/apps/desktop/src/components/layout/AppDialogs.vue @@ -178,6 +178,7 @@ watch( v-model:open="dialogs.showTransferDialog.value" :prefill-connection-id="dialogs.transferPrefillConnectionId.value" :prefill-database="dialogs.transferPrefillDatabase.value" + :prefill-catalog="dialogs.transferPrefillCatalog.value" :prefill-schema="dialogs.transferPrefillSchema.value" :prefill-tables="dialogs.transferPrefillTables.value" :prefill-target-connection-id="dialogs.transferPrefillTargetConnectionId.value" diff --git a/apps/desktop/src/components/transfer/DataTransferDialog.vue b/apps/desktop/src/components/transfer/DataTransferDialog.vue index 90ad1b1b96..09c86f1d1b 100644 --- a/apps/desktop/src/components/transfer/DataTransferDialog.vue +++ b/apps/desktop/src/components/transfer/DataTransferDialog.vue @@ -15,8 +15,11 @@ import * as api from "@/lib/backend/api"; import type { TransferMode, TransferTableNameCase } from "@/lib/backend/api"; import type { DatabaseType } from "@/types/database"; import { isSchemaAware, supportsTransfer } from "@/lib/database/databaseCapabilities"; -import { databaseOptionsForConnection, fetchNamespaceOptionsForConnection, namespaceOptionsAreSchemas } from "@/composables/useDatabaseOptions"; +import { isDorisFamilyCatalogCapable } from "@/lib/database/databaseFeatureSupport"; +import { isSameTransferDatabase, normalizeTransferCatalog } from "@/lib/database/dataTransferSelection"; +import { databaseOptionsForConnection, fetchCatalogNamespaceOptions, fetchNamespaceOptionsForConnection, namespaceOptionsAreSchemas } from "@/composables/useDatabaseOptions"; import { useExportTracker } from "@/composables/useExportTracker"; +import type { CatalogInfo } from "@/types/database"; import { ArrowRightLeft, ArrowLeftRight, Loader2, Square, CheckSquare } from "@lucide/vue"; const { t } = useI18n(); @@ -26,6 +29,7 @@ const open = defineModel("open", { default: false }); const props = defineProps<{ prefillConnectionId?: string; prefillDatabase?: string; + prefillCatalog?: string; prefillSchema?: string; prefillTables?: string[]; prefillTargetConnectionId?: string; @@ -39,6 +43,8 @@ const sqlConnections = computed(() => store.connections.filter((c) => supportsTr // Source state const sourceConnectionId = ref(""); +const sourceCatalog = ref(""); +const sourceCatalogs = ref([]); const sourceDatabase = ref(""); const sourceDatabases = ref([]); const sourceSchemas = ref([]); @@ -52,6 +58,8 @@ const pendingSelectedTablesPrefill = ref(null); // Target state const targetConnectionId = ref(""); +const targetCatalog = ref(""); +const targetCatalogs = ref([]); const targetDatabase = ref(""); const targetDatabases = ref([]); const targetSchemas = ref([]); @@ -68,7 +76,7 @@ const ownershipDialogOpen = ref(false); const ownershipMissingOwners = ref([]); const ownershipTargetOwner = ref(""); const pendingOwnershipRequest = ref(null); -const pendingOwnershipRefresh = ref<{ targetConnection: string; targetDatabase: string; targetSchema: string; shouldRefreshTargetTree: boolean } | null>(null); +const pendingOwnershipRefresh = ref<{ shouldRefreshTargetTree: boolean } | null>(null); const filteredTables = computed(() => { const q = tableSearch.value.toLowerCase(); @@ -85,11 +93,22 @@ function isMongoConnection(id: string): boolean { return connectionType(id) === "mongodb"; } +function isCatalogCapable(id: string): boolean { + const config = store.getConfig(id); + return isDorisFamilyCatalogCapable(config?.db_type, config?.driver_profile); +} + const canStart = computed(() => { const effectiveSourceSchema = sourceSchema.value || sourceDatabase.value; const effectiveTargetSchema = targetSchema.value || targetDatabase.value; - const sameSourceAndTarget = sourceConnectionId.value === targetConnectionId.value && sourceDatabase.value === targetDatabase.value && effectiveSourceSchema === effectiveTargetSchema; - return sourceConnectionId.value && sourceDatabase.value && targetConnectionId.value && targetDatabase.value && selectedTables.value.size > 0 && !sameSourceAndTarget; + const sameCatalogAndDatabase = isSameTransferDatabase( + { connectionId: sourceConnectionId.value, catalog: sourceCatalog.value, catalogs: sourceCatalogs.value, database: sourceDatabase.value }, + { connectionId: targetConnectionId.value, catalog: targetCatalog.value, catalogs: targetCatalogs.value, database: targetDatabase.value }, + ); + const sameSourceAndTarget = sameCatalogAndDatabase && effectiveSourceSchema === effectiveTargetSchema; + return ( + !!sourceConnectionId.value && !!sourceDatabase.value && !!targetConnectionId.value && !!targetDatabase.value && (sourceCatalogs.value.length <= 1 || !!sourceCatalog.value) && (targetCatalogs.value.length <= 1 || !!targetCatalog.value) && selectedTables.value.size > 0 && !sameSourceAndTarget + ); }); function toggleSelectAll() { @@ -108,6 +127,37 @@ function toggleTable(table: string) { } } +async function loadCatalogs(connectionId: string, side: "source" | "target") { + if (!connectionId || !isCatalogCapable(connectionId)) { + if (side === "source") { + sourceCatalogs.value = []; + sourceCatalog.value = ""; + } else { + targetCatalogs.value = []; + targetCatalog.value = ""; + } + return; + } + try { + const catalogs = await api.listDorisCatalogs(connectionId); + if (side === "source") { + sourceCatalogs.value = catalogs; + sourceCatalog.value = catalogs.length === 1 ? catalogs[0].name : ""; + } else { + targetCatalogs.value = catalogs; + targetCatalog.value = catalogs.length === 1 ? catalogs[0].name : ""; + } + } catch { + if (side === "source") { + sourceCatalogs.value = []; + sourceCatalog.value = ""; + } else { + targetCatalogs.value = []; + targetCatalog.value = ""; + } + } +} + async function loadDatabases(connectionId: string, target: "source" | "target") { if (!connectionId) return; try { @@ -128,6 +178,26 @@ async function loadDatabases(connectionId: string, target: "source" | "target") } } +async function loadDatabasesForCatalog(connectionId: string, catalog: string, target: "source" | "target") { + if (!connectionId || !catalog) return; + try { + await store.ensureConnected(connectionId); + const config = store.getConfig(connectionId); + if (!config) return; + const names = await fetchCatalogNamespaceOptions(connectionId, catalog, config); + if (target === "source") { + sourceDatabases.value = names; + sourceDatabase.value = names.length === 1 ? names[0] : ""; + } else { + targetDatabases.value = names; + targetDatabase.value = names.length === 1 ? names[0] : ""; + } + } catch { + if (target === "source") sourceDatabases.value = []; + else targetDatabases.value = []; + } +} + async function loadSchemas(connectionId: string, database: string, side: "source" | "target", preferredSchema = "") { if (!connectionId || !database) return; if (isMongoConnection(connectionId)) { @@ -182,7 +252,8 @@ async function loadTables() { const config = store.getConfig(sourceConnectionId.value); const needsSchema = isSchemaAware(config?.db_type); const schema = needsSchema && sourceSchema.value ? sourceSchema.value : sourceDatabase.value; - const tables = await api.listTables(sourceConnectionId.value, sourceDatabase.value, schema); + const catalog = sourceCatalog.value || undefined; + const tables = await api.listTables(sourceConnectionId.value, sourceDatabase.value, schema, undefined, undefined, undefined, undefined, catalog); sourceTables.value = tables.filter((t) => t.table_type === "TABLE" || t.table_type === "BASE TABLE").map((t) => t.name); applyPendingTableSelection(); } catch { @@ -195,17 +266,36 @@ async function loadTables() { const skipSourceWatch = ref(false); const skipTargetWatch = ref(false); -watch(sourceConnectionId, (id) => { +watch(sourceConnectionId, async (id) => { if (skipSourceWatch.value) { skipSourceWatch.value = false; return; } + sourceCatalog.value = ""; + sourceCatalogs.value = []; sourceDatabase.value = ""; sourceTables.value = []; selectedTables.value.clear(); pendingSourceSchemaPrefill.value = ""; pendingSelectedTablesPrefill.value = null; - loadDatabases(id, "source"); + if (isCatalogCapable(id)) { + await loadCatalogs(id, "source"); + if (sourceCatalog.value) { + await loadDatabasesForCatalog(id, sourceCatalog.value, "source"); + } + } else { + await loadDatabases(id, "source"); + } +}); + +watch(sourceCatalog, async (catalog) => { + if (!sourceConnectionId.value) return; + sourceDatabase.value = ""; + sourceTables.value = []; + selectedTables.value.clear(); + if (catalog) { + await loadDatabasesForCatalog(sourceConnectionId.value, catalog, "source"); + } }); watch(sourceDatabase, async (db) => { @@ -227,15 +317,35 @@ watch(sourceDatabase, async (db) => { watch(sourceSchema, () => loadTables()); -watch(targetConnectionId, (id) => { +watch(targetConnectionId, async (id) => { if (skipTargetWatch.value) { skipTargetWatch.value = false; return; } + targetCatalog.value = ""; + targetCatalogs.value = []; targetDatabase.value = ""; targetSchemas.value = []; targetSchema.value = ""; - loadDatabases(id, "target"); + pendingTargetSchemaPrefill.value = ""; + if (isCatalogCapable(id)) { + await loadCatalogs(id, "target"); + if (targetCatalog.value) { + await loadDatabasesForCatalog(id, targetCatalog.value, "target"); + } + } else { + await loadDatabases(id, "target"); + } +}); + +watch(targetCatalog, async (catalog) => { + if (!targetConnectionId.value) return; + targetDatabase.value = ""; + targetSchemas.value = []; + targetSchema.value = ""; + if (catalog) { + await loadDatabasesForCatalog(targetConnectionId.value, catalog, "target"); + } }); watch(targetDatabase, async (db) => { @@ -264,13 +374,30 @@ watch( if (props.prefillConnectionId) { skipSourceWatch.value = true; sourceConnectionId.value = props.prefillConnectionId; - await loadDatabases(props.prefillConnectionId, "source"); + if (isCatalogCapable(props.prefillConnectionId)) { + await loadCatalogs(props.prefillConnectionId, "source"); + if (props.prefillCatalog) { + sourceCatalog.value = props.prefillCatalog; + } + if (sourceCatalog.value) { + await loadDatabasesForCatalog(props.prefillConnectionId, sourceCatalog.value, "source"); + } + } else { + await loadDatabases(props.prefillConnectionId, "source"); + } if (props.prefillDatabase) sourceDatabase.value = props.prefillDatabase; } if (props.prefillTargetConnectionId) { skipTargetWatch.value = true; targetConnectionId.value = props.prefillTargetConnectionId; - await loadDatabases(props.prefillTargetConnectionId, "target"); + if (isCatalogCapable(props.prefillTargetConnectionId)) { + await loadCatalogs(props.prefillTargetConnectionId, "target"); + if (targetCatalog.value) { + await loadDatabasesForCatalog(props.prefillTargetConnectionId, targetCatalog.value, "target"); + } + } else { + await loadDatabases(props.prefillTargetConnectionId, "target"); + } if (props.prefillTargetDatabase) targetDatabase.value = props.prefillTargetDatabase; } } @@ -280,6 +407,8 @@ watch( function resetState() { sourceConnectionId.value = ""; + sourceCatalog.value = ""; + sourceCatalogs.value = []; sourceDatabase.value = ""; sourceDatabases.value = []; sourceSchemas.value = []; @@ -290,6 +419,8 @@ function resetState() { pendingSelectedTablesPrefill.value = null; tableSearch.value = ""; targetConnectionId.value = ""; + targetCatalog.value = ""; + targetCatalogs.value = []; targetDatabase.value = ""; targetDatabases.value = []; targetSchemas.value = []; @@ -323,9 +454,11 @@ async function startTransfer() { sourceConnectionId: sourceConnectionId.value, sourceDatabase: sourceDatabaseName, sourceSchema: effectiveSourceSchema, + sourceCatalog: normalizeTransferCatalog(sourceCatalog.value, sourceCatalogs.value) || undefined, targetConnectionId: targetConnection, targetDatabase: targetDatabaseName, targetSchema: effectiveTargetSchema, + targetCatalog: normalizeTransferCatalog(targetCatalog.value, targetCatalogs.value) || undefined, tables: [...selectedTables.value], createTable: createTable.value, mode: transferMode.value, @@ -342,9 +475,6 @@ async function startTransfer() { ownershipTargetOwner.value = preview.targetOwner; pendingOwnershipRequest.value = request; pendingOwnershipRefresh.value = { - targetConnection, - targetDatabase: targetDatabaseName, - targetSchema: effectiveTargetSchema, shouldRefreshTargetTree, }; ownershipDialogOpen.value = true; @@ -357,16 +487,16 @@ async function startTransfer() { } } - runTransfer(request, targetConnection, targetDatabaseName, effectiveTargetSchema, shouldRefreshTargetTree); + runTransfer(request, shouldRefreshTargetTree); } -function runTransfer(request: api.TransferRequest, targetConnection: string, targetDatabaseName: string, effectiveTargetSchema: string, shouldRefreshTargetTree: boolean) { +function runTransfer(request: api.TransferRequest, shouldRefreshTargetTree: boolean) { isSubmitting.value = true; - startDataTransferTask(request, `${request.sourceDatabase} → ${targetDatabaseName}`, { + startDataTransferTask(request, `${request.sourceDatabase} → ${request.targetDatabase}`, { formatOverlapError: (tables) => t("transfer.targetTableBusy", { tables: tables.join(", ") }), onDone: async () => { if (shouldRefreshTargetTree) { - await store.refreshObjectListTreeNode(targetConnection, targetDatabaseName, effectiveTargetSchema); + await store.refreshObjectListTreeNode(request.targetConnectionId, request.targetDatabase, request.targetSchema, request.targetCatalog); } }, }); @@ -386,7 +516,7 @@ function resolveOwnershipDecision(policy: api.TransferOwnershipPolicy | null) { isSubmitting.value = false; return; } - runTransfer({ ...request, ownershipPolicy: policy }, refresh.targetConnection, refresh.targetDatabase, refresh.targetSchema, refresh.shouldRefreshTargetTree); + runTransfer({ ...request, ownershipPolicy: policy }, refresh.shouldRefreshTargetTree); } function getConnectionName(id: string) { @@ -437,6 +567,21 @@ function getConnectionName(id: string) { + +
+ + +
+
+ +
+ + +
+
{ }); it("allows a transfer between different schemas in the same database", () => { - expect(dialogSource).toContain("const sameSourceAndTarget = sourceConnectionId.value === targetConnectionId.value"); - expect(dialogSource).toContain("effectiveSourceSchema === effectiveTargetSchema"); + expect(dialogSource).toContain("isSameTransferDatabase"); + expect(dialogSource).toContain("const sameSourceAndTarget = sameCatalogAndDatabase && effectiveSourceSchema === effectiveTargetSchema"); + }); + + it("keeps catalog filtering and completion refresh catalog-aware", () => { + expect(dialogSource).toContain("fetchCatalogNamespaceOptions(connectionId, catalog, config)"); + expect(dialogSource).toContain("store.refreshObjectListTreeNode(request.targetConnectionId, request.targetDatabase, request.targetSchema, request.targetCatalog)"); }); }); diff --git a/apps/desktop/src/composables/__tests__/useDatabaseOptions.spec.ts b/apps/desktop/src/composables/__tests__/useDatabaseOptions.spec.ts index e4f21985ae..b73721fadc 100644 --- a/apps/desktop/src/composables/__tests__/useDatabaseOptions.spec.ts +++ b/apps/desktop/src/composables/__tests__/useDatabaseOptions.spec.ts @@ -2,6 +2,7 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; import { databaseAfterCatalogChange, databaseOptionsForConnection, + fetchCatalogNamespaceOptions, fetchNamespaceOptionsForConnection, fetchSqlFileTargetOptions, namespaceOptionsAreSchemas, @@ -130,6 +131,19 @@ describe("namespace options", () => { ).toEqual(["analytics"]); }); + it("preserves visible database filtering for catalog-scoped transfer options", async () => { + mocks.listDorisCatalogDatabases.mockResolvedValue([{ name: "app" }, { name: "analytics" }]); + + await expect( + fetchCatalogNamespaceOptions("connection-1", "hive", { + db_type: "starrocks", + visible_databases: ["analytics"], + }), + ).resolves.toEqual(["analytics"]); + + expect(mocks.listDorisCatalogDatabases).toHaveBeenCalledWith("connection-1", "hive"); + }); + it("propagates metadata loading errors", async () => { const error = new Error("schema metadata failed"); mocks.listSchemas.mockRejectedValue(error); diff --git a/apps/desktop/src/composables/__tests__/useExportTracker.spec.ts b/apps/desktop/src/composables/__tests__/useExportTracker.spec.ts index 874f7972b3..fd0acc39da 100644 --- a/apps/desktop/src/composables/__tests__/useExportTracker.spec.ts +++ b/apps/desktop/src/composables/__tests__/useExportTracker.spec.ts @@ -139,6 +139,26 @@ describe("data transfer task duration", () => { await Promise.resolve(); }); + it("allows the same target table in different catalogs", async () => { + const resolvers: Array<() => void> = []; + vi.mocked(api.startTransfer).mockImplementation( + () => + new Promise((resolve) => { + resolvers.push(resolve); + }), + ); + const tracker = useExportTracker(); + + tracker.startDataTransferTask({ ...transferRequest("hive"), targetCatalog: "hive" }, "hive"); + const iceberg = tracker.startDataTransferTask({ ...transferRequest("iceberg"), targetCatalog: "iceberg" }, "iceberg"); + + expect(iceberg.status).toBe("Running"); + expect(api.startTransfer).toHaveBeenCalledTimes(2); + + for (const resolve of resolvers) resolve(); + await Promise.resolve(); + }); + it("keeps SQL-file backend elapsed time unchanged", () => { const tracker = useExportTracker(); const task = tracker.addSqlFileTask("sql", "script.sql", "/tmp/script.sql"); diff --git a/apps/desktop/src/composables/useDatabaseOptions.ts b/apps/desktop/src/composables/useDatabaseOptions.ts index 00f47ffb7c..4937b4aa6a 100644 --- a/apps/desktop/src/composables/useDatabaseOptions.ts +++ b/apps/desktop/src/composables/useDatabaseOptions.ts @@ -1,6 +1,7 @@ import { ref } from "vue"; import { useConnectionStore } from "@/stores/connectionStore"; import { filterDatabaseNamesForConnection, filterSchemaNamesForConnection } from "@/lib/database/visibleDatabases"; +import { isDorisFamilyCatalogCapable } from "@/lib/database/databaseFeatureSupport"; import { usesTreeSchemaMode } from "@/lib/database/databaseCapabilities"; import { isInternalDorisCatalog } from "@/lib/database/databaseFeatureSupport"; import type { CatalogInfo, ConnectionConfig } from "@/types/database"; @@ -56,6 +57,14 @@ export async function fetchNamespaceOptionsForConnection(connectionId: string, c ); } +export async function fetchCatalogNamespaceOptions(connectionId: string, catalog: string, connection: NamespaceOptionsConnection): Promise { + const dbs = await api.listDorisCatalogDatabases(connectionId, catalog); + return databaseOptionsForConnection( + dbs.map((db) => db.name), + connection, + ); +} + export async function fetchSqlFileTargetOptions(connectionId: string, connection: NamespaceOptionsConnection): Promise { return fetchNamespaceOptionsForConnection(connectionId, connection); } @@ -72,7 +81,7 @@ export function useDatabaseOptions() { const catalogRequests = new Map>(); const catalogDatabaseRequests = new Map>(); - async function loadDatabaseOptions(connectionId: string) { + async function loadDatabaseOptions(connectionId: string, catalog?: string) { const connection = connectionStore.getConfig(connectionId); if (!connection || loadingDatabaseOptions.value[connectionId]) return; @@ -87,6 +96,12 @@ export function useDatabaseOptions() { ); } else if (connection.db_type === "mongodb") { databaseOptions.value[connectionId] = filterDatabaseNamesForConnection(await api.mongoListDatabases(connectionId), connection); + } else if (catalog && isDorisFamilyCatalogCapable(connection?.db_type, connection?.driver_profile)) { + const dbs = await api.listDorisCatalogDatabases(connectionId, catalog); + databaseOptions.value[connectionId] = databaseOptionsForConnection( + dbs.map((db) => db.name), + connection, + ); } else { const dbs = await api.listDatabases(connectionId); databaseOptions.value[connectionId] = databaseOptionsForConnection( diff --git a/apps/desktop/src/composables/useDialogSources.ts b/apps/desktop/src/composables/useDialogSources.ts index b9eaeb7524..b185b71d24 100644 --- a/apps/desktop/src/composables/useDialogSources.ts +++ b/apps/desktop/src/composables/useDialogSources.ts @@ -23,6 +23,7 @@ const pendingImportContent = ref(""); const transferPrefillConnectionId = ref(""); const transferPrefillDatabase = ref(""); +const transferPrefillCatalog = ref(""); const transferPrefillSchema = ref(""); const transferPrefillTables = ref([]); const transferPrefillTargetConnectionId = ref(""); @@ -70,6 +71,7 @@ let watchersRegistered = false; function clearTransferPrefill() { transferPrefillConnectionId.value = ""; transferPrefillDatabase.value = ""; + transferPrefillCatalog.value = ""; transferPrefillSchema.value = ""; transferPrefillTables.value = []; transferPrefillTargetConnectionId.value = ""; @@ -92,6 +94,7 @@ export function useDialogSources() { if (v) { transferPrefillConnectionId.value = v.connectionId; transferPrefillDatabase.value = v.database; + transferPrefillCatalog.value = v.catalog ?? ""; transferPrefillSchema.value = v.schema ?? ""; transferPrefillTables.value = v.tables ?? []; transferPrefillTargetConnectionId.value = v.targetConnectionId ?? ""; @@ -331,6 +334,7 @@ export function useDialogSources() { pendingImportContent, transferPrefillConnectionId, transferPrefillDatabase, + transferPrefillCatalog, transferPrefillSchema, transferPrefillTables, transferPrefillTargetConnectionId, diff --git a/apps/desktop/src/composables/useExportTracker.ts b/apps/desktop/src/composables/useExportTracker.ts index fb4ba0f6c4..afb06307d8 100644 --- a/apps/desktop/src/composables/useExportTracker.ts +++ b/apps/desktop/src/composables/useExportTracker.ts @@ -35,6 +35,7 @@ export interface ExportTask { totalTables?: number; currentTable?: string; targetConnectionId?: string; + targetCatalog?: string; targetDatabase?: string; targetSchema?: string; targetTables?: string[]; @@ -201,6 +202,7 @@ function findActiveOverlappingTransfer(request: api.TransferRequest): string[] { if (task.kind !== "data-transfer") continue; if (task.status !== "Running" && task.status !== "Writing") continue; if (task.targetConnectionId !== request.targetConnectionId) continue; + if ((task.targetCatalog ?? "") !== (request.targetCatalog ?? "")) continue; if (task.targetDatabase !== request.targetDatabase) continue; if (task.targetSchema !== request.targetSchema) continue; @@ -321,6 +323,7 @@ export function useExportTracker() { const task = existingTask ?? addDataTransferTask(request.transferId, label, request.tables.length); task.startedAt ??= Date.now(); task.targetConnectionId = request.targetConnectionId; + task.targetCatalog = request.targetCatalog; task.targetDatabase = request.targetDatabase; task.targetSchema = request.targetSchema; task.targetTables = request.tables.map((table) => targetTableName(table, request.targetTableNameCase)); diff --git a/apps/desktop/src/composables/useSidebarTreeToolRuntime.ts b/apps/desktop/src/composables/useSidebarTreeToolRuntime.ts index 12699f98af..4d52c1af8b 100644 --- a/apps/desktop/src/composables/useSidebarTreeToolRuntime.ts +++ b/apps/desktop/src/composables/useSidebarTreeToolRuntime.ts @@ -21,6 +21,7 @@ export function useSidebarTreeToolRuntime(options: SidebarTreeToolRuntimeOptions connectionStore.transferSource = { connectionId: activeNode.value.connectionId, database: activeNode.value.database ?? "", + catalog: activeNode.value.catalog, }; } diff --git a/apps/desktop/src/i18n/locales/en.ts b/apps/desktop/src/i18n/locales/en.ts index 5a3ded4cbe..a98bd8dedc 100644 --- a/apps/desktop/src/i18n/locales/en.ts +++ b/apps/desktop/src/i18n/locales/en.ts @@ -3475,6 +3475,10 @@ export default { source: "Source", target: "Target", sourceConnection: "Source Connection", + sourceCatalog: "Source Catalog", + targetCatalog: "Target Catalog", + selectCatalog: "Select catalog...", + searchCatalog: "Search catalog...", sourceDatabase: "Source Database", sourceSchema: "Source Schema", targetConnection: "Target Connection", diff --git a/apps/desktop/src/i18n/locales/es.ts b/apps/desktop/src/i18n/locales/es.ts index 2dc93b8a0e..60b70eb1d5 100644 --- a/apps/desktop/src/i18n/locales/es.ts +++ b/apps/desktop/src/i18n/locales/es.ts @@ -3343,6 +3343,10 @@ export default withEnglishFallback({ searchConnection: "Buscar conexiones...", searchDatabase: "Buscar bases de datos...", searchSchema: "Buscar esquemas...", + sourceCatalog: "Catalog de origen", + targetCatalog: "Catalog de destino", + selectCatalog: "Seleccione Catalog...", + searchCatalog: "Buscar Catalog...", }, tableImport: { title: "Importar datos a tabla", diff --git a/apps/desktop/src/i18n/locales/it.ts b/apps/desktop/src/i18n/locales/it.ts index 000f31ee93..f3f78a67a2 100644 --- a/apps/desktop/src/i18n/locales/it.ts +++ b/apps/desktop/src/i18n/locales/it.ts @@ -3341,6 +3341,10 @@ export default withEnglishFallback({ searchConnection: "Cerca connessioni...", searchDatabase: "Cerca database...", searchSchema: "Cerca schemi...", + sourceCatalog: "Catalogo sorgente", + targetCatalog: "Catalogo destinazione", + selectCatalog: "Seleziona Catalogo...", + searchCatalog: "Cerca Catalogo...", }, tableImport: { title: "Importa Dati Tabella", diff --git a/apps/desktop/src/i18n/locales/ja.ts b/apps/desktop/src/i18n/locales/ja.ts index 8afe4464c6..022630a479 100644 --- a/apps/desktop/src/i18n/locales/ja.ts +++ b/apps/desktop/src/i18n/locales/ja.ts @@ -3342,6 +3342,10 @@ export default withEnglishFallback({ searchConnection: "接続を検索...", searchDatabase: "データベースを検索...", searchSchema: "スキーマを検索...", + sourceCatalog: "ソースCatalog", + targetCatalog: "ターゲットCatalog", + selectCatalog: "Catalogを選択してください...", + searchCatalog: "Catalogを検索...", }, tableImport: { title: "テーブルデータをインポート", diff --git a/apps/desktop/src/i18n/locales/pt-BR.ts b/apps/desktop/src/i18n/locales/pt-BR.ts index eab0472f66..4868532fd3 100644 --- a/apps/desktop/src/i18n/locales/pt-BR.ts +++ b/apps/desktop/src/i18n/locales/pt-BR.ts @@ -3343,6 +3343,10 @@ export default withEnglishFallback({ searchConnection: "Pesquisar conexões...", searchDatabase: "Pesquisar bancos de dados...", searchSchema: "Pesquisar schemas...", + sourceCatalog: "Catalog de origem", + targetCatalog: "Catalog de destino", + selectCatalog: "Por favor, selecione Catalog...", + searchCatalog: "Buscar Catalog...", }, tableImport: { title: "Importar Dados da Tabela", diff --git a/apps/desktop/src/i18n/locales/zh-CN.ts b/apps/desktop/src/i18n/locales/zh-CN.ts index a97c750f6f..558e0052f8 100644 --- a/apps/desktop/src/i18n/locales/zh-CN.ts +++ b/apps/desktop/src/i18n/locales/zh-CN.ts @@ -3475,6 +3475,10 @@ export default withEnglishFallback({ source: "源", target: "目标", sourceConnection: "源连接", + sourceCatalog: "源 Catalog", + targetCatalog: "目标 Catalog", + selectCatalog: "请选择 Catalog...", + searchCatalog: "搜索 Catalog...", sourceDatabase: "源数据库", sourceSchema: "源 Schema", targetConnection: "目标连接", diff --git a/apps/desktop/src/i18n/locales/zh-TW.ts b/apps/desktop/src/i18n/locales/zh-TW.ts index 0bc359e3cc..bf3dc2011b 100644 --- a/apps/desktop/src/i18n/locales/zh-TW.ts +++ b/apps/desktop/src/i18n/locales/zh-TW.ts @@ -3014,6 +3014,10 @@ export default withEnglishFallback({ searchConnection: "搜尋連線...", searchDatabase: "搜尋資料庫...", searchSchema: "搜尋結構描述...", + sourceCatalog: "源 Catalog", + targetCatalog: "目標 Catalog", + selectCatalog: "請選擇 Catalog...", + searchCatalog: "搜尋 Catalog...", }, tableImport: { title: "匯入資料表資料", diff --git a/apps/desktop/src/lib/__tests__/database/dataTransferSelection.spec.ts b/apps/desktop/src/lib/__tests__/database/dataTransferSelection.spec.ts new file mode 100644 index 0000000000..3855622b65 --- /dev/null +++ b/apps/desktop/src/lib/__tests__/database/dataTransferSelection.spec.ts @@ -0,0 +1,38 @@ +import { describe, expect, it } from "vitest"; +import { isSameTransferDatabase, normalizeTransferCatalog, type TransferDatabaseSelection } from "@/lib/database/dataTransferSelection"; +import type { CatalogInfo } from "@/types/database"; + +function catalog(name: string, catalog_type: string): CatalogInfo { + return { name, catalog_type, is_current: false }; +} + +function selection(overrides: Partial = {}): TransferDatabaseSelection { + return { + connectionId: "connection-1", + catalog: "", + catalogs: [], + database: "sales", + ...overrides, + }; +} + +describe("data transfer database selection", () => { + it("normalizes empty, Doris internal, and StarRocks default catalogs", () => { + expect(normalizeTransferCatalog("", [])).toBe(""); + expect(normalizeTransferCatalog("internal", [])).toBe(""); + expect(normalizeTransferCatalog("default_catalog", [catalog("default_catalog", "Internal")])).toBe(""); + }); + + it("treats internal catalog aliases as the same transfer database", () => { + expect(isSameTransferDatabase(selection(), selection({ catalog: "internal", catalogs: [catalog("internal", "internal")] }))).toBe(true); + expect(isSameTransferDatabase(selection(), selection({ catalog: "default_catalog", catalogs: [catalog("default_catalog", "Internal")] }))).toBe(true); + }); + + it("keeps different external catalogs distinct", () => { + expect(isSameTransferDatabase(selection({ catalog: "iceberg", catalogs: [catalog("iceberg", "Iceberg")] }), selection({ catalog: "hive", catalogs: [catalog("hive", "Hive")] }))).toBe(false); + }); + + it("compares connection and database fields without concatenation collisions", () => { + expect(isSameTransferDatabase(selection({ connectionId: "ab", database: "c" }), selection({ connectionId: "a", database: "bc" }))).toBe(false); + }); +}); diff --git a/apps/desktop/src/lib/backend/tauri.ts b/apps/desktop/src/lib/backend/tauri.ts index f1b52c00c0..d18c092d7c 100644 --- a/apps/desktop/src/lib/backend/tauri.ts +++ b/apps/desktop/src/lib/backend/tauri.ts @@ -3170,9 +3170,11 @@ export interface TransferRequest { sourceConnectionId: string; sourceDatabase: string; sourceSchema: string; + sourceCatalog?: string; targetConnectionId: string; targetDatabase: string; targetSchema: string; + targetCatalog?: string; tables: string[]; createTable: boolean; mode: TransferMode; diff --git a/apps/desktop/src/lib/database/dataTransferSelection.ts b/apps/desktop/src/lib/database/dataTransferSelection.ts new file mode 100644 index 0000000000..857c762a51 --- /dev/null +++ b/apps/desktop/src/lib/database/dataTransferSelection.ts @@ -0,0 +1,20 @@ +import { isInternalDorisCatalog } from "@/lib/database/databaseFeatureSupport"; +import type { CatalogInfo } from "@/types/database"; + +export interface TransferDatabaseSelection { + connectionId: string; + catalog: string; + catalogs: readonly CatalogInfo[]; + database: string; +} + +export function normalizeTransferCatalog(catalog: string, catalogs: readonly CatalogInfo[]): string { + const normalizedCatalog = catalog.trim(); + if (!normalizedCatalog) return ""; + const catalogInfo = catalogs.find((item) => item.name.trim() === normalizedCatalog); + return isInternalDorisCatalog(catalogInfo?.catalog_type, normalizedCatalog) ? "" : normalizedCatalog; +} + +export function isSameTransferDatabase(source: TransferDatabaseSelection, target: TransferDatabaseSelection): boolean { + return source.connectionId === target.connectionId && source.database === target.database && normalizeTransferCatalog(source.catalog, source.catalogs) === normalizeTransferCatalog(target.catalog, target.catalogs); +} diff --git a/apps/desktop/src/lib/sidebar/treeRefreshTarget.ts b/apps/desktop/src/lib/sidebar/treeRefreshTarget.ts index 62ebb56769..09cfd34e53 100644 --- a/apps/desktop/src/lib/sidebar/treeRefreshTarget.ts +++ b/apps/desktop/src/lib/sidebar/treeRefreshTarget.ts @@ -1,12 +1,13 @@ import type { TreeNode } from "@/types/database"; -export function findDatabaseTreeNode(nodes: TreeNode[], connectionId: string, database: string): TreeNode | null { +export function findDatabaseTreeNode(nodes: TreeNode[], connectionId: string, database: string, catalog?: string): TreeNode | null { for (const node of nodes) { - if (node.type === "database" && node.connectionId === connectionId && node.database === database) { + const catalogMatches = catalog ? node.catalog === catalog : !node.catalog; + if (node.type === "database" && node.connectionId === connectionId && node.database === database && catalogMatches) { return node; } if (node.children) { - const found = findDatabaseTreeNode(node.children, connectionId, database); + const found = findDatabaseTreeNode(node.children, connectionId, database, catalog); if (found) return found; } } diff --git a/apps/desktop/src/stores/__tests__/connectionStore.dorisCatalog.spec.ts b/apps/desktop/src/stores/__tests__/connectionStore.dorisCatalog.spec.ts new file mode 100644 index 0000000000..9bcfba8e5e --- /dev/null +++ b/apps/desktop/src/stores/__tests__/connectionStore.dorisCatalog.spec.ts @@ -0,0 +1,126 @@ +import { createPinia, setActivePinia } from "pinia"; +import { beforeEach, describe, expect, it, vi } from "vitest"; +import type { ConnectionConfig, TreeNode } from "@/types/database"; + +function installLocalStorage() { + const data = new Map(); + vi.stubGlobal("localStorage", { + getItem: vi.fn((key: string) => data.get(key) ?? null), + setItem: vi.fn((key: string, value: string) => data.set(key, value)), + removeItem: vi.fn((key: string) => data.delete(key)), + }); +} + +function starRocksConnection(): ConnectionConfig { + return { + id: "starrocks-1", + name: "StarRocks", + db_type: "starrocks", + host: "127.0.0.1", + port: 9030, + username: "root", + password: "", + database: "sales", + visible_databases: ["analytics"], + } as ConnectionConfig; +} + +function catalogNode(connectionId: string, catalog: string, catalogType: string, children: TreeNode[] = []): TreeNode { + return { + id: `${connectionId}:doris-catalog:${catalog}`, + label: catalog, + type: "doris-catalog", + connectionId, + catalog, + catalogType, + isExpanded: true, + children, + }; +} + +function databaseNode(connectionId: string, catalog: string, database: string): TreeNode { + return { + id: `${connectionId}:doris-catalog:${catalog}:${database}`, + label: database, + type: "database", + connectionId, + catalog, + database, + isExpanded: true, + children: [], + }; +} + +describe("connectionStore Doris catalog tree", () => { + beforeEach(() => { + vi.resetModules(); + vi.unstubAllGlobals(); + installLocalStorage(); + setActivePinia(createPinia()); + }); + + it("applies visible database filtering inside external catalogs", async () => { + const listDorisCatalogDatabases = vi.fn().mockResolvedValue([{ name: "app" }, { name: "analytics" }]); + vi.doMock("@/lib/backend/tauriRuntime", () => ({ isTauriRuntime: () => false })); + vi.doMock("@/lib/backend/api", () => ({ + checkConnectionHealth: vi.fn().mockResolvedValue(undefined), + listDorisCatalogDatabases, + loadSchemaCache: vi.fn().mockResolvedValue(null), + saveConnections: vi.fn().mockResolvedValue(undefined), + saveSchemaCache: vi.fn().mockResolvedValue(undefined), + saveSidebarLayout: vi.fn().mockResolvedValue(undefined), + })); + + const { useConnectionStore } = await import("@/stores/connectionStore"); + const store = useConnectionStore(); + const connection = starRocksConnection(); + const catalog = catalogNode(connection.id, "hive", "Hive"); + store.connections = [connection]; + store.connectedIds.add(connection.id); + store.treeNodes = [{ id: connection.id, label: connection.name, type: "connection", connectionId: connection.id, children: [catalog] }]; + + await store.loadTreeNodeChildren(catalog, { force: true }); + + expect(catalog.children?.map((node) => [node.catalog, node.database])).toEqual([["hive", "analytics"]]); + }); + + it("refreshes the selected catalog when database names collide", async () => { + const listTables = vi.fn().mockResolvedValue([{ name: "orders", table_type: "BASE TABLE", comment: null }]); + vi.doMock("@/lib/backend/tauriRuntime", () => ({ isTauriRuntime: () => false })); + vi.doMock("@/lib/backend/api", () => ({ + checkConnectionHealth: vi.fn().mockResolvedValue(undefined), + deleteSchemaCachePrefix: vi.fn().mockResolvedValue(undefined), + listTables, + loadSchemaCache: vi.fn().mockResolvedValue(null), + saveConnections: vi.fn().mockResolvedValue(undefined), + saveSchemaCache: vi.fn().mockResolvedValue(undefined), + saveSidebarLayout: vi.fn().mockResolvedValue(undefined), + })); + + const { useConnectionStore } = await import("@/stores/connectionStore"); + const { useSettingsStore } = await import("@/stores/settingsStore"); + const store = useConnectionStore(); + useSettingsStore().editorSettings.sidebarObjectDisplay = "grouped"; + const connection = starRocksConnection(); + const hiveDatabase = databaseNode(connection.id, "hive", "sales"); + const icebergDatabase = databaseNode(connection.id, "iceberg", "sales"); + store.connections = [connection]; + store.connectedIds.add(connection.id); + store.treeNodes = [ + { + id: connection.id, + label: connection.name, + type: "connection", + connectionId: connection.id, + children: [catalogNode(connection.id, "hive", "Hive", [hiveDatabase]), catalogNode(connection.id, "iceberg", "Iceberg", [icebergDatabase])], + }, + ]; + + await store.refreshObjectListTreeNode(connection.id, "sales", "sales", "iceberg"); + + expect(listTables).toHaveBeenCalledTimes(1); + expect(listTables.mock.calls[0]?.[7]).toBe("iceberg"); + expect(hiveDatabase.children).toEqual([]); + expect(icebergDatabase.children?.map((node) => [node.catalog, node.database, node.label])).toEqual([["iceberg", "sales", "orders"]]); + }); +}); diff --git a/apps/desktop/src/stores/__tests__/queryStore.database-open.spec.ts b/apps/desktop/src/stores/__tests__/queryStore.database-open.spec.ts index 85909cc8fa..e419570487 100644 --- a/apps/desktop/src/stores/__tests__/queryStore.database-open.spec.ts +++ b/apps/desktop/src/stores/__tests__/queryStore.database-open.spec.ts @@ -82,6 +82,21 @@ describe("queryStore database open state", () => { expect(store.openTableStructure("doris-1", "sales", undefined, "orders", undefined, undefined, "iceberg_catalog")).toBe(icebergTabId); }); + it("clears the catalog when changing a query tab connection", async () => { + const { useQueryStore } = await import("@/stores/queryStore"); + const store = useQueryStore(); + + const tabId = store.createTab("doris-1", "sales", "Query", "query", undefined, undefined, "hive_catalog"); + const tab = store.tabs.find((item) => item.id === tabId)!; + + store.updateConnection(tabId, "doris-2", "analytics"); + + expect(tab.connectionId).toBe("doris-2"); + expect(tab.catalog).toBeUndefined(); + expect(tab.database).toBe("analytics"); + expect(tab.schema).toBeUndefined(); + }); + it("closes data and structure tabs for a dropped table object", async () => { const { useQueryStore } = await import("@/stores/queryStore"); const store = useQueryStore(); diff --git a/apps/desktop/src/stores/connectionStore.ts b/apps/desktop/src/stores/connectionStore.ts index 1fc0681ea7..970059abaf 100644 --- a/apps/desktop/src/stores/connectionStore.ts +++ b/apps/desktop/src/stores/connectionStore.ts @@ -358,6 +358,7 @@ export const useConnectionStore = defineStore("connection", () => { const transferSource = ref<{ connectionId: string; database: string; + catalog?: string; schema?: string; tables?: string[]; targetConnectionId?: string; @@ -3849,21 +3850,21 @@ export const useConnectionStore = defineStore("connection", () => { if (useCachedChildren(node, options, load)) return; const config = getConfig(connectionId); const databases = await withMetadataLoadTimeout(connectionId, api.listDorisCatalogDatabases(connectionId, catalog), "databases"); + const visibleNames = filterDatabaseNamesForConnection( + databases.map((database) => database.name), + config, + ); + const visibleNameSet = new Set(visibleNames); + const visibleDatabases = databases.filter((database) => visibleNameSet.has(database.name)); let databaseNodes: TreeNode[]; if (isInternalDorisCatalog(node.catalogType, catalog)) { // The internal catalog's databases are rendered as standard database // nodes so they reuse the existing table-loading / table-open paths. // Detection is type-based (catalogType=`internal`), so StarRocks // `default_catalog` routes here too — its tables carry no catalog. - const visibleNames = filterDatabaseNamesForConnection( - databases.map((database) => database.name), - config, - ); - const visibleNameSet = new Set(visibleNames); - const visibleDatabases = databases.filter((database) => visibleNameSet.has(database.name)); databaseNodes = buildDatabaseTreeNodes(connectionId, visibleDatabases, { includeDefaultWhenEmpty: false }); } else { - databaseNodes = sortSidebarDatabases(databases).flatMap((database) => { + databaseNodes = sortSidebarDatabases(visibleDatabases).flatMap((database) => { const name = database.name.trim(); if (!name) return []; return [ @@ -5128,24 +5129,31 @@ export const useConnectionStore = defineStore("connection", () => { await refreshTreeNode(node); } - async function refreshDatabaseTreeNode(connectionId: string, database: string) { - const node = findDatabaseTreeNode(treeNodes.value, connectionId, database); + async function refreshDatabaseTreeNode(connectionId: string, database: string, catalog?: string) { + const node = findDatabaseTreeNode(treeNodes.value, connectionId, database, catalog); if (node) { await refreshTreeNode(node); return; } + if (catalog) { + const catalogNode = findNode(treeNodes.value, dorisCatalogId(connectionId, catalog)); + if (catalogNode) { + await refreshTreeNode(catalogNode); + return; + } + } await loadDatabases(connectionId, { force: true }); } - async function refreshObjectListTreeNode(connectionId: string, database: string, schema?: string) { + async function refreshObjectListTreeNode(connectionId: string, database: string, schema?: string, catalog?: string) { invalidateMetadataCaches({ connectionId, database, schema }); - const shouldRefreshSchemaNode = !!schema; + const shouldRefreshSchemaNode = !!schema && !catalog; const node = shouldRefreshSchemaNode ? findNode(treeNodes.value, `${connectionId}:${database}:${schema}`) : null; if (node) { await refreshTreeNode(node); return; } - await refreshDatabaseTreeNode(connectionId, database); + await refreshDatabaseTreeNode(connectionId, database, catalog); } function isSchemaAwareDatabase(connectionId: string): boolean { diff --git a/apps/desktop/src/stores/queryStore.ts b/apps/desktop/src/stores/queryStore.ts index e138d4c4c8..8498f9a4e9 100644 --- a/apps/desktop/src/stores/queryStore.ts +++ b/apps/desktop/src/stores/queryStore.ts @@ -2579,6 +2579,7 @@ export const useQueryStore = defineStore("query", () => { void closeResultSession(tab); void closeClientConnectionSession(tab); tab.connectionId = connectionId; + tab.catalog = undefined; tab.database = database; tab.catalog = undefined; tab.objectBrowser = undefined; diff --git a/crates/dbx-core/assets/database-drivers.manifest.json b/crates/dbx-core/assets/database-drivers.manifest.json index 948fc14a87..19ace4bdd7 100644 --- a/crates/dbx-core/assets/database-drivers.manifest.json +++ b/crates/dbx-core/assets/database-drivers.manifest.json @@ -506,7 +506,7 @@ "runtimeMode": "native", "mcpMode": "direct", "singleConnectionPool": false, - "metadataConnectionScoped": false, + "metadataConnectionScoped": true, "skipTcpProbe": false, "defaultPort": 9030, "supportLevel": "operate", @@ -535,7 +535,7 @@ "runtimeMode": "native", "mcpMode": "direct", "singleConnectionPool": false, - "metadataConnectionScoped": false, + "metadataConnectionScoped": true, "skipTcpProbe": false, "defaultPort": 9030, "supportLevel": "operate", @@ -564,7 +564,7 @@ "runtimeMode": "native", "mcpMode": "direct", "singleConnectionPool": false, - "metadataConnectionScoped": false, + "metadataConnectionScoped": true, "skipTcpProbe": false, "defaultPort": 9306, "supportLevel": "operate", diff --git a/crates/dbx-core/src/connection.rs b/crates/dbx-core/src/connection.rs index 379273be40..06d10a0869 100644 --- a/crates/dbx-core/src/connection.rs +++ b/crates/dbx-core/src/connection.rs @@ -304,6 +304,17 @@ pub fn metadata_connection_config(config: &ConnectionConfig) -> ConnectionConfig } pub fn database_connection_config(config: &ConnectionConfig, database: Option<&str>) -> ConnectionConfig { + database_connection_config_with_catalog(config, database, None) +} + +/// Like [`database_connection_config`], but optionally injects a Doris/StarRocks +/// `catalog=` URL parameter so mysql_async emits `SET catalog` during +/// connection setup (before any `USE `). +pub fn database_connection_config_with_catalog( + config: &ConnectionConfig, + database: Option<&str>, + catalog: Option<&str>, +) -> ConnectionConfig { let mut db_config = if database.is_some() { config.clone() } else { metadata_connection_config(config) }; if let Some(db) = database { if !matches!( @@ -317,9 +328,35 @@ pub fn database_connection_config(config: &ConnectionConfig, database: Option<&s db_config.database = Some(db.to_string()); } } + if let Some(catalog) = catalog.map(str::trim).filter(|catalog| !catalog.is_empty()) { + db_config.url_params = Some(upsert_connection_url_param(db_config.url_params.as_deref(), "catalog", catalog)); + } db_config } +/// Insert or replace a single `key=value` entry in a connection URL-params string. +pub fn upsert_connection_url_param(params: Option<&str>, key: &str, value: &str) -> String { + let key = key.trim(); + let value = value.trim(); + let key_lower = key.to_ascii_lowercase(); + let encoded_value = percent_encoding::utf8_percent_encode(value, percent_encoding::NON_ALPHANUMERIC).to_string(); + let mut parts: Vec = params + .unwrap_or("") + .trim() + .trim_start_matches('?') + .split('&') + .filter(|part| !part.trim().is_empty()) + .filter(|part| { + part.split_once('=') + .map(|(existing_key, _)| existing_key.trim().to_ascii_lowercase() != key_lower) + .unwrap_or(true) + }) + .map(str::to_string) + .collect(); + parts.push(format!("{key}={encoded_value}")); + parts.join("&") +} + pub fn prestosql_jdbc_config_for_endpoint(config: &ConnectionConfig, host: &str, port: u16) -> ConnectionConfig { let mut jdbc_config = config.clone(); jdbc_config.connection_string = @@ -1138,13 +1175,22 @@ impl AppState { self.get_or_create_pool_for_session(connection_id, database, None).await } + pub async fn get_or_create_pool_with_catalog( + &self, + connection_id: &str, + database: Option<&str>, + catalog: Option<&str>, + ) -> Result { + self.get_or_create_pool_for_session_with_catalog(connection_id, database, catalog, None).await + } + pub async fn get_or_create_pool_for_connection_attempt( &self, connection_id: &str, database: Option<&str>, attempt: u64, ) -> Result { - self.get_or_create_pool_for_session_inner(connection_id, database, None, Some(attempt)).await + self.get_or_create_pool_for_session_inner(connection_id, database, None, None, Some(attempt)).await } pub async fn get_or_create_pool_for_session( @@ -1153,13 +1199,24 @@ impl AppState { database: Option<&str>, client_session_id: Option<&str>, ) -> Result { - self.get_or_create_pool_for_session_inner(connection_id, database, client_session_id, None).await + self.get_or_create_pool_for_session_with_catalog(connection_id, database, None, client_session_id).await + } + + pub async fn get_or_create_pool_for_session_with_catalog( + &self, + connection_id: &str, + database: Option<&str>, + catalog: Option<&str>, + client_session_id: Option<&str>, + ) -> Result { + self.get_or_create_pool_for_session_inner(connection_id, database, catalog, client_session_id, None).await } async fn get_or_create_pool_for_session_inner( &self, connection_id: &str, database: Option<&str>, + catalog: Option<&str>, client_session_id: Option<&str>, connection_attempt: Option, ) -> Result { @@ -1170,8 +1227,9 @@ impl AppState { validate_connection_url_params(&config)?; let db_type = Some(config.db_type); let validate_existing_pool = should_validate_existing_pool_before_reuse(config.db_type); + let catalog = catalog.map(str::trim).filter(|value| !value.is_empty()); - let base_pool_key = base_pool_key_for(db_type, connection_id, database, false); + let base_pool_key = base_pool_key_for_with_catalog(db_type, connection_id, database, catalog, false); let pool_key = session_scoped_pool_key_for(Some(&config), base_pool_key.clone(), client_session_id); loop { @@ -1198,7 +1256,7 @@ impl AppState { break; } - let db_config = database_connection_config(&config, database); + let db_config = database_connection_config_with_catalog(&config, database, catalog); validate_h2_file_connection(&db_config)?; self.ensure_current_connection_attempt(connection_id, connection_attempt).await?; @@ -2383,13 +2441,24 @@ impl AppState { connection_id: &str, database: Option<&str>, client_session_id: Option<&str>, + ) -> Result { + self.reconnect_pool_for_session_with_catalog(connection_id, database, None, client_session_id).await + } + + pub async fn reconnect_pool_for_session_with_catalog( + &self, + connection_id: &str, + database: Option<&str>, + catalog: Option<&str>, + client_session_id: Option<&str>, ) -> Result { let config = { let configs = self.configs.read().await; configs.get(connection_id).cloned() }; let db_type = config.as_ref().map(|config| config.db_type); - let base_pool_key = base_pool_key_for(db_type, connection_id, database, true); + let catalog = catalog.map(str::trim).filter(|value| !value.is_empty()); + let base_pool_key = base_pool_key_for_with_catalog(db_type, connection_id, database, catalog, true); let pool_key = session_scoped_pool_key_for(config.as_ref(), base_pool_key, client_session_id); if self.uses_forwarded_transport(connection_id).await { self.remove_connection_pools(connection_id).await; @@ -2403,7 +2472,7 @@ impl AppState { close_pool_kind_with_timeout(pool_key.clone(), pool).await; } } - self.get_or_create_pool_for_session(connection_id, database, client_session_id).await + self.get_or_create_pool_for_session_with_catalog(connection_id, database, catalog, client_session_id).await } pub async fn close_client_session_pool( @@ -3558,6 +3627,16 @@ fn base_pool_key_for( connection_id: &str, database: Option<&str>, include_elasticsearch_single_pool: bool, +) -> String { + base_pool_key_for_with_catalog(db_type, connection_id, database, None, include_elasticsearch_single_pool) +} + +fn base_pool_key_for_with_catalog( + db_type: Option, + connection_id: &str, + database: Option<&str>, + catalog: Option<&str>, + include_elasticsearch_single_pool: bool, ) -> String { let is_single_connection_pool = db_type.as_ref().is_some_and(|db_type| { let is_single = database_capabilities::is_single_connection_pool(db_type) @@ -3573,13 +3652,17 @@ fn base_pool_key_for( is_single && (!database_capabilities::is_agent_type(db_type) || shares_database_pool_with_connection(db_type)) }); - if is_single_connection_pool { + let key = if is_single_connection_pool { connection_id.to_string() } else { match database.filter(|db| !db.trim().is_empty()) { Some(db) => format!("{connection_id}:{db}"), None => connection_id.to_string(), } + }; + match catalog.map(str::trim).filter(|value| !value.is_empty()) { + Some(catalog) => format!("{key}:catalog:{catalog}"), + None => key, } } @@ -3784,12 +3867,13 @@ 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, 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, + database_connection_config_with_catalog, 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, upsert_connection_url_param, 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, @@ -3864,6 +3948,38 @@ mod tests { } } + #[test] + fn upsert_connection_url_param_inserts_and_replaces_catalog() { + assert_eq!(upsert_connection_url_param(None, "catalog", "paimon"), "catalog=paimon"); + assert_eq!( + upsert_connection_url_param(Some("charset=utf8mb4"), "catalog", "hive_catalog"), + "charset=utf8mb4&catalog=hive%5Fcatalog" + ); + assert_eq!( + upsert_connection_url_param(Some("catalog=old&charset=utf8mb4"), "catalog", "new_cat"), + "charset=utf8mb4&catalog=new%5Fcat" + ); + } + + #[test] + fn database_connection_config_with_catalog_keeps_database_for_use_after_set_catalog() { + let mut config = mysql_config(None); + config.db_type = DatabaseType::StarRocks; + let db_config = database_connection_config_with_catalog(&config, Some("ads"), Some("paimon_catalog")); + assert_eq!(db_config.database.as_deref(), Some("ads")); + assert_eq!(db_config.url_params.as_deref(), Some("catalog=paimon%5Fcatalog")); + let url = db_config.connection_url(); + assert!(url.contains("/ads"), "url should include database for USE setup: {url}"); + assert!(url.contains("catalog=paimon%5Fcatalog"), "url should include catalog param: {url}"); + } + + #[test] + fn metadata_connection_config_clears_starrocks_default_database() { + let mut config = mysql_config(Some("ads")); + config.db_type = DatabaseType::StarRocks; + assert_eq!(metadata_connection_config(&config).database, None); + } + #[test] fn task_client_session_ids_are_stable_and_isolated() { assert_eq!(task_client_session_id("table-export", "job-1"), "table-export:job-1"); diff --git a/crates/dbx-core/src/data_compare.rs b/crates/dbx-core/src/data_compare.rs index 5b627e9dd8..fe5dac23d4 100644 --- a/crates/dbx-core/src/data_compare.rs +++ b/crates/dbx-core/src/data_compare.rs @@ -526,6 +526,7 @@ pub async fn prepare_data_compare_missing_target( &target_database_type, &source_database_type, None, + None, ) )); pre_sync_statements.extend( diff --git a/crates/dbx-core/src/database_capabilities.rs b/crates/dbx-core/src/database_capabilities.rs index e5d2a5b187..0391fd8e82 100644 --- a/crates/dbx-core/src/database_capabilities.rs +++ b/crates/dbx-core/src/database_capabilities.rs @@ -37,7 +37,14 @@ pub fn is_single_connection_pool(db_type: &DatabaseType) -> bool { } pub fn is_metadata_connection_scoped(db_type: &DatabaseType) -> bool { - matches!(db_type, DatabaseType::Mysql) + // MySQL-protocol engines (incl. Doris/StarRocks) must not keep a configured + // default database when the caller asks for an unscoped metadata pool. + // External-catalog databases do not exist in the default catalog, so leaving + // `config.database` in the URL makes handshake/`USE` fail with Unknown database. + matches!( + db_type, + DatabaseType::Mysql | DatabaseType::Doris | DatabaseType::StarRocks | DatabaseType::ManticoreSearch + ) } pub fn skips_tcp_probe(db_type: &DatabaseType) -> bool { diff --git a/crates/dbx-core/src/database_export.rs b/crates/dbx-core/src/database_export.rs index dadd48feec..f4c25e2027 100644 --- a/crates/dbx-core/src/database_export.rs +++ b/crates/dbx-core/src/database_export.rs @@ -1012,7 +1012,7 @@ fn generate_postgres_sequence_owner_ddl(sequence: &PostgresExportSequence, schem Some(format!( "ALTER SEQUENCE {} OWNED BY {}.{}", postgres_sequence_qualified_name(schema, &sequence.name), - crate::transfer::qualified_table(owner_table, schema, &DatabaseType::Postgres), + crate::transfer::qualified_table(owner_table, schema, &DatabaseType::Postgres, None), quote_identifier(owner_column, &DatabaseType::Postgres) )) } @@ -1026,7 +1026,7 @@ fn generate_postgres_sequence_setval_sql(sequence: &PostgresExportSequence, sche let sequence_literal = quote_postgres_string_literal(&postgres_sequence_qualified_name(schema, &sequence.name)); match (sequence.owner_table.as_deref(), sequence.owner_column.as_deref()) { (Some(owner_table), Some(owner_column)) => { - let owner_table = crate::transfer::qualified_table(owner_table, schema, &DatabaseType::Postgres); + let owner_table = crate::transfer::qualified_table(owner_table, schema, &DatabaseType::Postgres, None); let owner_column = quote_identifier(owner_column, &DatabaseType::Postgres); Some(format!( "SELECT setval({sequence_literal}, GREATEST(COALESCE(MAX({owner_column}), {last_value}), {last_value}), true) FROM {owner_table}" @@ -1269,7 +1269,7 @@ fn record_export_error(file: &mut std::fs::File, fail_on_error: bool, message: S fn database_export_select_sql(columns: &[String], table: &str, schema: &str, db_type: &DatabaseType) -> String { let columns = columns.iter().map(|column| quote_identifier(column, db_type)).collect::>().join(", "); - let table = crate::transfer::qualified_table(table, schema, db_type); + let table = crate::transfer::qualified_table(table, schema, db_type, None); format!("SELECT {columns} FROM {table}") } @@ -1283,7 +1283,8 @@ fn write_database_export_rows( schema: &str, db_type: &DatabaseType, ) -> Result<(), String> { - let mut insert_sql = crate::transfer::generate_insert_typed(columns, column_types, rows, table, schema, db_type); + let mut insert_sql = + crate::transfer::generate_insert_typed(columns, column_types, rows, table, schema, db_type, None); if *db_type == DatabaseType::Dameng && selected_columns_include_identity_extras(columns, column_extras) { insert_sql = wrap_dameng_identity_insert_sql(&insert_sql, table, schema); } @@ -1829,7 +1830,7 @@ pub async fn export_database_sql_core( ) .await?; } else { - let count_query = crate::transfer::count_sql(table_name, &request.schema, &db_type); + let count_query = crate::transfer::count_sql(table_name, &request.schema, &db_type, None); let total_rows = match crate::transfer::execute_read_on_pool(state, &pool_key, &count_query).await { Ok(result) => { let count = result.rows.first().and_then(|row| row.first()).and_then(|value| match value { @@ -2137,7 +2138,7 @@ fn filter_export_table_infos( } fn drop_table_if_exists_sql(table_name: &str, schema: &str, db_type: &DatabaseType) -> String { - format!("DROP TABLE IF EXISTS {};", crate::transfer::qualified_table(table_name, schema, db_type)) + format!("DROP TABLE IF EXISTS {};", crate::transfer::qualified_table(table_name, schema, db_type, None)) } fn build_database_export_object_source_sql( diff --git a/crates/dbx-core/src/db/cloudflare_d1/import.rs b/crates/dbx-core/src/db/cloudflare_d1/import.rs index 69fd5436e7..3220b908df 100644 --- a/crates/dbx-core/src/db/cloudflare_d1/import.rs +++ b/crates/dbx-core/src/db/cloudflare_d1/import.rs @@ -35,7 +35,7 @@ pub(crate) fn build_import_insert_batches( .collect::>(); build_sql_batches(rows.len(), max_rows, "import row", |range| { - generate_insert_typed(&columns, &column_types, &rows[range], table, schema, &DatabaseType::CloudflareD1) + generate_insert_typed(&columns, &column_types, &rows[range], table, schema, &DatabaseType::CloudflareD1, None) }) .map(|batches| { batches.into_iter().map(|batch| ImportSqlBatch { sql: batch.sql, row_count: batch.item_count }).collect() diff --git a/crates/dbx-core/src/db/cloudflare_d1/transfer_tests.rs b/crates/dbx-core/src/db/cloudflare_d1/transfer_tests.rs index a1d56c1988..9070f7fc22 100644 --- a/crates/dbx-core/src/db/cloudflare_d1/transfer_tests.rs +++ b/crates/dbx-core/src/db/cloudflare_d1/transfer_tests.rs @@ -13,6 +13,7 @@ fn uses_sqlite_upsert_syntax() { "main", &DatabaseType::CloudflareD1, &[String::from("id")], + None, ); assert!(sql.contains("ON CONFLICT (\"id\") DO UPDATE SET")); diff --git a/crates/dbx-core/src/schema.rs b/crates/dbx-core/src/schema.rs index 03dac4d3cd..4af48cce1f 100644 --- a/crates/dbx-core/src/schema.rs +++ b/crates/dbx-core/src/schema.rs @@ -316,7 +316,7 @@ pub async fn list_doris_catalog_databases_core( return Ok(vec![]); } }; - if catalog == "internal" { + if catalog == "internal" || catalog.eq_ignore_ascii_case("default_catalog") { return Ok(filter_mysql_system_databases_for_config(databases, db_config.as_ref())); } Ok(databases) @@ -460,7 +460,8 @@ pub async fn resolve_external_doris_catalog( catalog: Option<&str>, ) -> Option { let catalog = catalog?.trim(); - if catalog.is_empty() || catalog == "internal" { + if catalog.is_empty() || catalog.eq_ignore_ascii_case("internal") || catalog.eq_ignore_ascii_case("default_catalog") + { return None; } let config = connection_config(state, connection_id).await?; diff --git a/crates/dbx-core/src/sql_dialect/identifiers.rs b/crates/dbx-core/src/sql_dialect/identifiers.rs index f919b115da..ef16aad748 100644 --- a/crates/dbx-core/src/sql_dialect/identifiers.rs +++ b/crates/dbx-core/src/sql_dialect/identifiers.rs @@ -386,9 +386,28 @@ pub(crate) fn quote_transfer_identifier(name: &str, database_type: &DatabaseType } } -pub(crate) fn qualified_transfer_table(table_name: &str, schema: &str, database_type: &DatabaseType) -> String { +/// Qualified table name for transfer SQL. +/// +/// * Without catalog: produces `schema.table` (or just `table` for MySQL family). +/// * With catalog AND the database type supports external catalogs (Doris/StarRocks): +/// produces `catalog.schema.table` — the 3-part form those engines require to +/// address objects in an external (non-internal) catalog. +pub(crate) fn qualified_transfer_table( + table_name: &str, + schema: &str, + database_type: &DatabaseType, + catalog: Option<&str>, +) -> String { let table = quote_transfer_identifier(table_name, database_type); - if schema.is_empty() || matches!(database_type, DatabaseType::Mysql | DatabaseType::MongoDb | DatabaseType::Questdb) + if let Some(catalog) = catalog { + format!( + "{}.{}.{}", + quote_transfer_identifier(catalog, database_type), + quote_transfer_identifier(schema, database_type), + table + ) + } else if schema.is_empty() + || matches!(database_type, DatabaseType::Mysql | DatabaseType::MongoDb | DatabaseType::Questdb) { table } else { diff --git a/crates/dbx-core/src/sql_dialect/tests.rs b/crates/dbx-core/src/sql_dialect/tests.rs index b82548a39c..4a67c84d2a 100644 --- a/crates/dbx-core/src/sql_dialect/tests.rs +++ b/crates/dbx-core/src/sql_dialect/tests.rs @@ -8,8 +8,8 @@ fn transfer_identifier_policy_preserves_legacy_output() { assert_eq!(quote_transfer_identifier("user`events", &DatabaseType::Doris), "`user``events`"); assert_eq!(quote_transfer_identifier("user]events", &DatabaseType::SqlServer), "[user]]events]"); assert_eq!(quote_transfer_identifier("user\"events", &DatabaseType::Postgres), "\"user\"\"events\""); - assert_eq!(qualified_transfer_table("events", "warehouse", &DatabaseType::Hive), "`warehouse`.`events`"); - assert_eq!(qualified_transfer_table("events", "warehouse", &DatabaseType::Mysql), "`events`"); + assert_eq!(qualified_transfer_table("events", "warehouse", &DatabaseType::Hive, None), "`warehouse`.`events`"); + assert_eq!(qualified_transfer_table("events", "warehouse", &DatabaseType::Mysql, None), "`events`"); } #[test] diff --git a/crates/dbx-core/src/table_export.rs b/crates/dbx-core/src/table_export.rs index ebfd0cebe9..72ad5a43a0 100644 --- a/crates/dbx-core/src/table_export.rs +++ b/crates/dbx-core/src/table_export.rs @@ -229,7 +229,7 @@ fn table_cursor_sql( col_names: &[String], primary_keys: &[String], ) -> String { - let full_table = qualified_table(&request.table_name, request.schema.as_deref().unwrap_or(""), db_type); + let full_table = qualified_table(&request.table_name, request.schema.as_deref().unwrap_or(""), db_type, None); let col_list = col_names.iter().map(|column| quote_identifier(column, db_type)).collect::>().join(", "); let predicate = crate::sql_dialect::normalize_where_input(request.where_input.as_deref()); let where_clause = if predicate.is_empty() { String::new() } else { format!(" WHERE ({predicate})") }; @@ -1155,6 +1155,7 @@ async fn export_table_data_core_inner( request.schema.as_deref().unwrap_or(""), &db_type, request.where_input.as_deref(), + None, ); match execute_table_export_count(state, &pool_key, request, &count_query, cancel_token.clone()).await { Ok(result) => result diff --git a/crates/dbx-core/src/table_import.rs b/crates/dbx-core/src/table_import.rs index 447a3a8b2b..ae8a0b1904 100644 --- a/crates/dbx-core/src/table_import.rs +++ b/crates/dbx-core/src/table_import.rs @@ -3068,7 +3068,8 @@ fn build_import_insert_batch_with_plan( return Ok(None); } let mapped_rows = map_import_rows_with_plan(rows, plan, db_type, kingbase_oracle_mode, date_time_format); - let sql = generate_insert_typed(&plan.target_columns, &plan.column_types, &mapped_rows, table, schema, db_type); + let sql = + generate_insert_typed(&plan.target_columns, &plan.column_types, &mapped_rows, table, schema, db_type, None); Ok((!sql.trim().is_empty()).then_some(ImportSqlBatch { sql, row_count: rows.len() })) } @@ -3093,6 +3094,7 @@ fn build_import_insert_batches_with_plan( table, schema, db_type, + None, SqlBatchLimits::for_database(db_type, rows.len()).with_hard_sql_bytes(hard_sql_bytes), )?; Ok(batches.into_iter().map(|(sql, row_count)| ImportSqlBatch { sql, row_count }).collect()) @@ -3359,7 +3361,7 @@ fn build_import_insert_batches_with_format( } pub fn truncate_sql(table: &str, schema: &str, db_type: &DatabaseType) -> String { - let full_table = qualified_table(table, schema, db_type); + let full_table = qualified_table(table, schema, db_type, None); match db_type { DatabaseType::Sqlite | DatabaseType::CloudflareD1 => format!("DELETE FROM {full_table}"), _ => format!("TRUNCATE TABLE {full_table}"), @@ -3643,7 +3645,7 @@ pub fn build_import_create_table_plan( return Err("No columns mapped for import".to_string()); } - let full_table = qualified_table(table.trim(), schema, db_type); + let full_table = qualified_table(table.trim(), schema, db_type, None); let column_sql = columns .iter() .map(|column| format!("{} {}", quote_identifier(&column.name, db_type), column.data_type)) @@ -3875,7 +3877,7 @@ fn build_postgres_copy_text_row( } fn postgres_copy_sql(plan: &CompiledImportPlan, table: &str, schema: &str) -> String { - let table = qualified_table(table, schema, &DatabaseType::Postgres); + let table = qualified_table(table, schema, &DatabaseType::Postgres, None); let columns = plan .target_columns .iter() @@ -4630,7 +4632,7 @@ fn compile_sqlserver_bulk_import_plan( target_types.push(column.column.data_type.clone()); } Ok(SqlServerBulkImportPlan { - target_table: qualified_table(table, schema, &DatabaseType::SqlServer), + target_table: qualified_table(table, schema, &DatabaseType::SqlServer, None), target_columns: import_plan.target_columns.clone(), target_types, requires_identity_insert, @@ -5185,6 +5187,7 @@ where &request.database, &request.schema, &request.table, + None, ) .await .unwrap_or_default() @@ -5535,6 +5538,7 @@ where &request.database, &request.schema, &request.table, + None, ) .await .unwrap_or_default() @@ -6012,6 +6016,7 @@ where &request.database, &request.schema, &request.table, + None, ) .await .unwrap_or_default() diff --git a/crates/dbx-core/src/transfer.rs b/crates/dbx-core/src/transfer.rs index aeb0a22eec..1cfe91c747 100644 --- a/crates/dbx-core/src/transfer.rs +++ b/crates/dbx-core/src/transfer.rs @@ -82,9 +82,11 @@ pub struct TransferRequest { pub source_connection_id: String, pub source_database: String, pub source_schema: String, + pub source_catalog: Option, pub target_connection_id: String, pub target_database: String, pub target_schema: String, + pub target_catalog: Option, pub tables: Vec, pub create_table: bool, #[serde(default)] @@ -141,8 +143,78 @@ pub fn quote_identifier(name: &str, db_type: &DatabaseType) -> String { quote_transfer_identifier(name, db_type) } -pub fn qualified_table(table: &str, schema: &str, db_type: &DatabaseType) -> String { - qualified_transfer_table(table, schema, db_type) +/// Resolve an optional catalog for external-catalog routing in the transfer +/// pipeline. Returns `Some(catalog)` only when: +/// - the catalog is non-empty and not a built-in catalog name, AND +/// - the database type is Doris/StarRocks, or MySQL (StarRocks/Doris are often +/// saved as `db_type=mysql` with a matching `driver_profile`; the transfer UI +/// only sends `sourceCatalog`/`targetCatalog` for catalog-capable connections). +/// +/// Built-in catalogs: Doris `internal`, StarRocks `default_catalog`. Prefer +/// [`resolve_external_transfer_catalog_for_config`] when a full connection +/// config is available (also matches `driver_profile=starrocks|doris`). +pub fn resolve_external_transfer_catalog<'a>(catalog: Option<&'a str>, db_type: &DatabaseType) -> Option<&'a str> { + let catalog = normalize_external_catalog_name(catalog)?; + match db_type { + DatabaseType::Doris | DatabaseType::StarRocks | DatabaseType::Mysql => Some(catalog), + _ => None, + } +} + +/// Like [`resolve_external_transfer_catalog`], but also treats MySQL connections +/// with a Doris/StarRocks `driver_profile` as catalog-capable. +pub fn resolve_external_transfer_catalog_for_config<'a>( + catalog: Option<&'a str>, + config: &crate::models::connection::ConnectionConfig, +) -> Option<&'a str> { + let catalog = normalize_external_catalog_name(catalog)?; + if crate::schema::is_doris_family_catalog_capable_config(config) { + Some(catalog) + } else { + None + } +} + +fn normalize_external_catalog_name(catalog: Option<&str>) -> Option<&str> { + let catalog = catalog?.trim(); + if catalog.is_empty() || catalog.eq_ignore_ascii_case("internal") || catalog.eq_ignore_ascii_case("default_catalog") + { + return None; + } + Some(catalog) +} + +/// Create (or reuse) a transfer pool for the given connection/database/catalog. +/// +/// For Doris/StarRocks external catalogs the pool is created with +/// `catalog=` in the connection URL params so mysql_async runs +/// `SET catalog` **before** `USE ` during setup. The handshake does +/// not send the external database name (mysql_async strips the path when a +/// catalog is present), which is what previously caused `Unknown database`. +pub async fn ensure_transfer_pool( + state: &AppState, + connection_id: &str, + database: &str, + catalog: Option<&str>, +) -> Result { + let config = { + let configs = state.configs.read().await; + configs.get(connection_id).cloned().ok_or_else(|| format!("Connection config not found: {connection_id}"))? + }; + if let Some(catalog) = resolve_external_transfer_catalog_for_config(catalog, &config) { + // SET catalog first, then USE so session has both catalog and + // database selected — StarRocks rejects unqualified analysis with + // "No database selected" when only SET catalog ran. + state.get_or_create_pool_with_catalog(connection_id, Some(database), Some(catalog)).await + } else { + state.get_or_create_pool(connection_id, Some(database)).await + } +} + +pub fn qualified_table(table: &str, schema: &str, db_type: &DatabaseType, catalog: Option<&str>) -> String { + // Only use 3-part catalog-qualified names for Doris/StarRocks external catalogs. + let effective_catalog = resolve_external_transfer_catalog(catalog, db_type); + qualified_transfer_table(table, schema, db_type, effective_catalog) } pub fn validate_transfer_target_table_names(request: &TransferRequest) -> Result<(), String> { @@ -221,6 +293,8 @@ async fn resolve_transfer_target_table_name( source_table: &str, target_pool_key: &str, target_db_type: &DatabaseType, + _source_catalog: Option<&str>, + target_catalog: Option<&str>, ) -> ResolvedTransferTargetTable { let requested_name = request.target_table_name(source_table); if is_mongodb_transfer_type(target_db_type) { @@ -229,22 +303,45 @@ async fn resolve_transfer_target_table_name( let allow_case_insensitive_match = target_table_lookup_is_case_insensitive(state, target_pool_key, target_db_type).await; - let tables = crate::schema::list_tables_core( - state, - &request.target_connection_id, - &request.target_database, - &request.target_schema, - Some(&requested_name), - Some(TRANSFER_TARGET_TABLE_LOOKUP_LIMIT), - None, - None, - None, - ) - .await - .unwrap_or_else(|error| { - log::debug!("[transfer] failed to resolve target table metadata for {requested_name}: {error}"); - Vec::new() - }); + + // Route through the catalog-aware path when targeting an external + // Doris/StarRocks catalog — otherwise the lookup runs against the + // default / internal catalog and can miss or misidentify the table. + let tables = if let Some(catalog) = resolve_external_transfer_catalog(target_catalog, target_db_type) { + crate::schema::list_doris_catalog_tables_core( + state, + &request.target_connection_id, + catalog, + &request.target_database, + Some(&requested_name), + Some(TRANSFER_TARGET_TABLE_LOOKUP_LIMIT), + None, + None, + None, + ) + .await + .unwrap_or_else(|error| { + log::debug!("[transfer] failed to resolve target table metadata for {requested_name} in catalog '{catalog}': {error}"); + Vec::new() + }) + } else { + crate::schema::list_tables_core( + state, + &request.target_connection_id, + &request.target_database, + &request.target_schema, + Some(&requested_name), + Some(TRANSFER_TARGET_TABLE_LOOKUP_LIMIT), + None, + None, + None, + ) + .await + .unwrap_or_else(|error| { + log::debug!("[transfer] failed to resolve target table metadata for {requested_name}: {error}"); + Vec::new() + }) + }; if let Some(existing_name) = existing_transfer_target_table_name(&requested_name, &tables, allow_case_insensitive_match) @@ -439,12 +536,12 @@ fn writable_transfer_columns( } fn dameng_identity_insert_statement(table: &str, schema: &str, enabled: bool) -> String { - let full_table = qualified_table(table, schema, &DatabaseType::Dameng); + let full_table = qualified_table(table, schema, &DatabaseType::Dameng, None); format!("SET IDENTITY_INSERT {full_table} {}", if enabled { "ON" } else { "OFF" }) } pub(crate) fn wrap_dameng_identity_insert_sql(insert_sql: &str, table: &str, schema: &str) -> String { - let full_table = qualified_table(table, schema, &DatabaseType::Dameng); + let full_table = qualified_table(table, schema, &DatabaseType::Dameng, None); wrap_dameng_identity_insert_sql_for_table(insert_sql, &full_table) } @@ -758,7 +855,7 @@ fn postgres_index_column_sql(column: &str) -> String { } fn generate_postgres_index_ddl(indexes: &[db::IndexInfo], table: &str, schema: &str) -> Vec { - let full_table = qualified_table(table, schema, &DatabaseType::Postgres); + let full_table = qualified_table(table, schema, &DatabaseType::Postgres, None); let mut statements = Vec::new(); for index in indexes.iter().filter(|index| !index.is_primary) { if index.name.trim().is_empty() || index.columns.is_empty() { @@ -822,7 +919,7 @@ fn generate_postgres_foreign_key_ddl( source_schema: &str, target_schema: &str, ) -> Vec { - let full_table = qualified_table(table, target_schema, &DatabaseType::Postgres); + let full_table = qualified_table(table, target_schema, &DatabaseType::Postgres, None); let mut grouped: HashMap<&str, Vec<&db::ForeignKeyInfo>> = HashMap::new(); let mut order = Vec::new(); @@ -853,7 +950,7 @@ fn generate_postgres_foreign_key_ddl( Some(ref_schema) => ref_schema, None => target_schema, }; - let referenced_table = qualified_table(&group[0].ref_table, referenced_schema, &DatabaseType::Postgres); + let referenced_table = qualified_table(&group[0].ref_table, referenced_schema, &DatabaseType::Postgres, None); statements.push(format!( "ALTER TABLE {full_table} ADD CONSTRAINT {} FOREIGN KEY ({columns}) REFERENCES {referenced_table} ({ref_columns})", quote_identifier(name, &DatabaseType::Postgres) @@ -864,7 +961,7 @@ fn generate_postgres_foreign_key_ddl( } fn generate_postgres_sequence_sync_sql(columns: &[db::ColumnInfo], table: &str, schema: &str) -> Vec { - let full_table = qualified_table(table, schema, &DatabaseType::Postgres); + let full_table = qualified_table(table, schema, &DatabaseType::Postgres, None); columns .iter() .filter(|column| { @@ -1055,7 +1152,7 @@ fn generate_postgres_domain_ddl(domain: &PostgresDomainSource, target_schema: &s } fn generate_postgres_materialized_view_ddls(view: &PostgresMaterializedViewSource, target_schema: &str) -> Vec { - let qualified_name = qualified_table(&view.view_name, target_schema, &DatabaseType::Postgres); + let qualified_name = qualified_table(&view.view_name, target_schema, &DatabaseType::Postgres, None); vec![ format!("DROP MATERIALIZED VIEW IF EXISTS {qualified_name}"), format!("CREATE MATERIALIZED VIEW {qualified_name} AS\n{}", ensure_sql_statement_terminated(&view.source)), @@ -1092,7 +1189,7 @@ fn rewrite_postgres_trigger_table_schema( table_name: &str, target_schema: &str, ) -> String { - let qualified_target_table = qualified_table(table_name, target_schema, &DatabaseType::Postgres); + let qualified_target_table = qualified_table(table_name, target_schema, &DatabaseType::Postgres, None); let candidate_patterns = [ format!( " ON {}.{} ", @@ -1804,8 +1901,9 @@ pub fn generate_create_table_ddl( target_db: &DatabaseType, source_db: &DatabaseType, table_comment: Option<&str>, + catalog: Option<&str>, ) -> String { - let full_table = qualified_table(table, schema, target_db); + let full_table = qualified_table(table, schema, target_db, catalog); let is_mysql_family = matches!( target_db, @@ -1921,7 +2019,7 @@ pub fn generate_comment_ddl( return Vec::new(); } - let full_table = qualified_table(table, schema, target_db); + let full_table = qualified_table(table, schema, target_db, None); let mut statements = Vec::new(); // Table-level comment first (PostgreSQL/Oracle only; ClickHouse doesn't support COMMENT ON TABLE) @@ -1966,7 +2064,7 @@ pub fn generate_insert( schema: &str, db_type: &DatabaseType, ) -> String { - generate_insert_typed(columns, &vec![None; columns.len()], rows, table, schema, db_type) + generate_insert_typed(columns, &vec![None; columns.len()], rows, table, schema, db_type, None) } pub fn generate_insert_typed( @@ -1976,12 +2074,13 @@ pub fn generate_insert_typed( table: &str, schema: &str, db_type: &DatabaseType, + catalog: Option<&str>, ) -> String { if rows.is_empty() { return String::new(); } - let template = InsertSqlTemplate::new(columns, table, schema, db_type); + let template = InsertSqlTemplate::new(columns, table, schema, db_type, catalog); let value_rows = value_rows_sql(rows, column_types, db_type); template.build(&value_rows) } @@ -1993,8 +2092,8 @@ struct InsertSqlTemplate { } impl InsertSqlTemplate { - fn new(columns: &[String], table: &str, schema: &str, db_type: &DatabaseType) -> Self { - let full_table = qualified_table(table, schema, db_type); + fn new(columns: &[String], table: &str, schema: &str, db_type: &DatabaseType, catalog: Option<&str>) -> Self { + let full_table = qualified_table(table, schema, db_type, catalog); let col_list = columns.iter().map(|column| quote_identifier(column, db_type)).collect::>().join(", "); Self { standard_prefix: format!("INSERT INTO {full_table} ({col_list}) VALUES\n"), @@ -2072,7 +2171,7 @@ pub fn generate_upsert( db_type: &DatabaseType, pk_columns: &[String], ) -> String { - generate_upsert_typed(columns, &vec![None; columns.len()], rows, table, schema, db_type, pk_columns) + generate_upsert_typed(columns, &vec![None; columns.len()], rows, table, schema, db_type, pk_columns, None) } pub fn generate_upsert_typed( @@ -2083,12 +2182,13 @@ pub fn generate_upsert_typed( schema: &str, db_type: &DatabaseType, pk_columns: &[String], + catalog: Option<&str>, ) -> String { if rows.is_empty() || pk_columns.is_empty() { return String::new(); } - let full_table = qualified_table(table, schema, db_type); + let full_table = qualified_table(table, schema, db_type, catalog); let col_list = columns.iter().map(|c| quote_identifier(c, db_type)).collect::>().join(", "); let value_rows = value_rows_sql(rows, column_types, db_type); @@ -2219,7 +2319,7 @@ pub fn generate_upsert_typed( sql.push_str(&format!("\nWHEN NOT MATCHED THEN INSERT ({insert_cols}) VALUES ({insert_vals})")); sql } - _ => generate_insert_typed(columns, column_types, rows, table, schema, db_type), + _ => generate_insert_typed(columns, column_types, rows, table, schema, db_type, catalog), } } @@ -2269,10 +2369,13 @@ fn generate_transfer_write_sql( schema: &str, db_type: &DatabaseType, pk_columns: &[String], + catalog: Option<&str>, ) -> String { match mode { - TransferMode::Upsert => generate_upsert_typed(columns, column_types, rows, table, schema, db_type, pk_columns), - _ => generate_insert_typed(columns, column_types, rows, table, schema, db_type), + TransferMode::Upsert => { + generate_upsert_typed(columns, column_types, rows, table, schema, db_type, pk_columns, catalog) + } + _ => generate_insert_typed(columns, column_types, rows, table, schema, db_type, catalog), } } @@ -2284,6 +2387,7 @@ pub(crate) fn generate_insert_typed_sql_batches( table: &str, schema: &str, db_type: &DatabaseType, + catalog: Option<&str>, limits: SqlBatchLimits, ) -> Result, String> { if rows.is_empty() { @@ -2297,7 +2401,7 @@ pub(crate) fn generate_insert_typed_sql_batches( }); let target_sql_bytes = limits.target_sql_bytes.max(1); let batch_sql_bytes = limits.hard_sql_bytes.map_or(target_sql_bytes, |hard| target_sql_bytes.min(hard)); - let template = InsertSqlTemplate::new(columns, table, schema, db_type); + let template = InsertSqlTemplate::new(columns, table, schema, db_type, catalog); let value_rows = value_rows_sql(rows, column_types, db_type); let value_row_bytes = value_rows.iter().map(|row| sql_text_bytes(row, db_type)).collect::>(); let mut statements = Vec::new(); @@ -2345,6 +2449,7 @@ fn generate_transfer_write_sql_batches( schema: &str, db_type: &DatabaseType, pk_columns: &[String], + catalog: Option<&str>, ) -> Result, String> { if rows.is_empty() { return Ok(Vec::new()); @@ -2358,6 +2463,7 @@ fn generate_transfer_write_sql_batches( table, schema, db_type, + catalog, SqlBatchLimits::for_database(db_type, max_transfer_write_rows(db_type, mode)), )? .into_iter() @@ -2384,6 +2490,7 @@ fn generate_transfer_write_sql_batches( schema, db_type, pk_columns, + catalog, ); while end < rows.len() && end - start < max_rows { @@ -2396,6 +2503,7 @@ fn generate_transfer_write_sql_batches( schema, db_type, pk_columns, + catalog, ); if candidate.len() > max_sql_bytes && !accepted.is_empty() { break; @@ -2421,7 +2529,7 @@ pub fn pagination_sql( offset: u64, limit: usize, ) -> String { - let full_table = qualified_table(table, schema, db_type); + let full_table = qualified_table(table, schema, db_type, None); let col_list = columns.iter().map(|c| quote_identifier(c, db_type)).collect::>().join(", "); match db_type { @@ -2452,8 +2560,9 @@ pub fn pagination_sql_with_order( offset: u64, limit: usize, order_by_columns: &[String], + catalog: Option<&str>, ) -> String { - let full_table = qualified_table(table, schema, db_type); + let full_table = qualified_table(table, schema, db_type, catalog); let col_list = columns.iter().map(|c| quote_identifier(c, db_type)).collect::>().join(", "); let order_expression = postgres_order_by_expression(order_by_columns, db_type); @@ -2493,7 +2602,7 @@ pub fn pagination_sql_with_filter_order( order_by: Option<&str>, default_order_columns: &[String], ) -> String { - let full_table = qualified_table(table, schema, db_type); + let full_table = qualified_table(table, schema, db_type, None); let col_list = columns.iter().map(|c| quote_identifier(c, db_type)).collect::>().join(", "); let predicate = crate::sql_dialect::normalize_where_input(where_input); let where_clause = if predicate.is_empty() { String::new() } else { format!(" WHERE ({predicate})") }; @@ -2527,12 +2636,18 @@ pub fn pagination_sql_with_filter_order( } } -pub fn count_sql(table: &str, schema: &str, db_type: &DatabaseType) -> String { - count_sql_with_where(table, schema, db_type, None) +pub fn count_sql(table: &str, schema: &str, db_type: &DatabaseType, catalog: Option<&str>) -> String { + count_sql_with_where(table, schema, db_type, None, catalog) } -pub fn count_sql_with_where(table: &str, schema: &str, db_type: &DatabaseType, where_input: Option<&str>) -> String { - let full_table = qualified_table(table, schema, db_type); +pub fn count_sql_with_where( + table: &str, + schema: &str, + db_type: &DatabaseType, + where_input: Option<&str>, + catalog: Option<&str>, +) -> String { + let full_table = qualified_table(table, schema, db_type, catalog); let predicate = crate::sql_dialect::normalize_where_input(where_input); let where_clause = if predicate.is_empty() { String::new() } else { format!(" WHERE ({predicate})") }; format!("SELECT COUNT(*) FROM {full_table}{where_clause}") @@ -2547,7 +2662,7 @@ pub fn keyset_pagination_sql( last_pk_values: &[serde_json::Value], limit: usize, ) -> String { - let full_table = qualified_table(table, schema, db_type); + let full_table = qualified_table(table, schema, db_type, None); let col_list = columns.iter().map(|c| quote_identifier(c, db_type)).collect::>().join(", "); let order = primary_keys.iter().map(|pk| format!("{} ASC", quote_identifier(pk, db_type))).collect::>().join(", "); @@ -2976,8 +3091,14 @@ async fn execute_on_pool_with_options( state.remove_pool_by_key(¤t_pool_key).await; return result; }; + let catalog = catalog_from_pool_key(¤t_pool_key).map(str::to_string); current_pool_key = state - .reconnect_pool_for_session(connection_id, database.as_deref(), client_session_id.as_deref()) + .reconnect_pool_for_session_with_catalog( + connection_id, + database.as_deref(), + catalog.as_deref(), + client_session_id.as_deref(), + ) .await?; } PoolErrorAction::ReconnectAndRetry => { @@ -3058,13 +3179,14 @@ async fn execute_on_pool_once( } fn database_from_pool_key(pool_key: &str) -> Option<&str> { - pool_key - .split_once(":session:") - .map(|(base, _)| base) - .unwrap_or(pool_key) - .split_once(':') - .map(|(_, database)| database) - .filter(|database| !database.is_empty()) + let base = pool_key.split_once(":session:").map(|(base, _)| base).unwrap_or(pool_key); + let base = base.split_once(":catalog:").map(|(base, _)| base).unwrap_or(base); + base.split_once(':').map(|(_, database)| database).filter(|database| !database.is_empty()) +} + +fn catalog_from_pool_key(pool_key: &str) -> Option<&str> { + let base = pool_key.split_once(":session:").map(|(base, _)| base).unwrap_or(pool_key); + base.split_once(":catalog:").map(|(_, catalog)| catalog).filter(|catalog| !catalog.is_empty()) } pub async fn get_db_type(state: &AppState, connection_id: &str) -> Result { @@ -3079,6 +3201,7 @@ pub async fn get_columns_for_transfer( database: &str, schema: &str, table: &str, + catalog: Option<&str>, ) -> Result, String> { let connections = state.connections.read().await; @@ -3129,8 +3252,14 @@ pub async fn get_columns_for_transfer( match pool { PoolKind::Mysql(p, _) => { let p = p.clone(); + let catalog = normalize_external_catalog_name(catalog).map(str::to_string); drop(connections); - db::mysql::get_columns(&p, &schema, &table).await + if let Some(catalog) = catalog { + // Use 3-part qualified column lookup for Doris/StarRocks external catalogs + db::doris::get_catalog_columns(&p, &catalog, &schema, &table).await + } else { + db::mysql::get_columns(&p, &schema, &table).await + } } PoolKind::Postgres(p) => { let p = p.clone(); @@ -3341,7 +3470,7 @@ async fn bind_postgres_owned_sequences_for_transfer( let owner_sql = format!( "ALTER SEQUENCE {} OWNED BY {}.{}", postgres_sequence_qualified_name(&request.target_schema, &sequence.name), - qualified_table(&sequence.owner_table, &request.target_schema, &DatabaseType::Postgres), + qualified_table(&sequence.owner_table, &request.target_schema, &DatabaseType::Postgres, None), quote_identifier(&sequence.owner_column, &DatabaseType::Postgres) ); execute_on_pool(state, target_pool_key, &owner_sql) @@ -3929,7 +4058,16 @@ where { let total_tables = request.tables.len(); let ResolvedTransferTargetTable { name: target_table, preexisting: target_table_preexisting } = - resolve_transfer_target_table_name(state, request, table, target_pool_key, target_db_type).await; + resolve_transfer_target_table_name( + state, + request, + table, + target_pool_key, + target_db_type, + request.source_catalog.as_deref(), + request.target_catalog.as_deref(), + ) + .await; let batch_size = if request.batch_size == 0 { 1000 } else { request.batch_size }; let mut offset: u64 = 0; let mut total_transferred: u64 = 0; @@ -3991,6 +4129,7 @@ where &request.source_database, &request.source_schema, table, + request.source_catalog.as_deref(), ) .await?; let col_names = columns.iter().map(|column| column.name.clone()).collect::>(); @@ -4007,6 +4146,7 @@ where offset, batch_size, &primary_key_columns, + request.source_catalog.as_deref(), ); let result = execute_on_pool(state, source_pool_key, &sql).await?; sql_rows_to_mongo_documents(&col_names, &result.rows) @@ -4071,6 +4211,7 @@ where target_db_type, source_db_type, None, + request.target_catalog.as_deref(), ); let target_table_created = transfer_create_table_created( execute_on_pool(state, target_pool_key, &ddl).await.map(|_| ()), @@ -4102,7 +4243,12 @@ where } if request.mode == TransferMode::Overwrite { - let full_table = qualified_table(&target_table, &request.target_schema, target_db_type); + let full_table = qualified_table( + &target_table, + &request.target_schema, + target_db_type, + request.target_catalog.as_deref(), + ); let truncate_sql = match target_db_type { DatabaseType::Sqlite | DatabaseType::CloudflareD1 | DatabaseType::DuckDb => { format!("DELETE FROM {full_table}") @@ -4131,6 +4277,7 @@ where &request.target_schema, target_db_type, &[], + request.target_catalog.as_deref(), )?; for (statement_index, batch_sql) in write_statements.iter().enumerate() { execute_on_pool(state, target_pool_key, batch_sql).await.map_err(|e| { @@ -4201,7 +4348,16 @@ where let total_tables = request.tables.len(); let pg_compat_transfer = is_postgres_compat_transfer(source_db_type, target_db_type); let ResolvedTransferTargetTable { name: target_table, preexisting: mut target_table_preexisting } = - resolve_transfer_target_table_name(state, request, table, target_pool_key, target_db_type).await; + resolve_transfer_target_table_name( + state, + request, + table, + target_pool_key, + target_db_type, + request.source_catalog.as_deref(), + request.target_catalog.as_deref(), + ) + .await; let preserves_target_table_name = target_table == table; // Get source columns (deduplicate by name) @@ -4213,6 +4369,7 @@ where &request.source_database, &request.source_schema, table, + request.source_catalog.as_deref(), ) .await?; let mut seen = std::collections::HashSet::new(); @@ -4235,22 +4392,44 @@ where log::info!("[transfer] {} has {} columns, counting rows...", table, columns.len()); // Fetch source table comment - let table_comment: Option = crate::schema::list_tables_core( - state, - &request.source_connection_id, - &request.source_database, - &request.source_schema, - Some(table), - Some(1), - None, - None, - None, - ) - .await - .unwrap_or_default() - .into_iter() - .next() - .and_then(|t| t.comment); + // Route through the catalog-aware path for Doris/StarRocks external catalogs + // so the comment comes from the selected catalog, not the default one. + let table_comment: Option = + if let Some(catalog) = resolve_external_transfer_catalog(request.source_catalog.as_deref(), source_db_type) { + crate::schema::list_doris_catalog_tables_core( + state, + &request.source_connection_id, + catalog, + &request.source_database, + Some(table), + Some(1), + None, + None, + None, + ) + .await + .unwrap_or_default() + .into_iter() + .next() + .and_then(|t| t.comment) + } else { + crate::schema::list_tables_core( + state, + &request.source_connection_id, + &request.source_database, + &request.source_schema, + Some(table), + Some(1), + None, + None, + None, + ) + .await + .unwrap_or_default() + .into_iter() + .next() + .and_then(|t| t.comment) + }; let source_indexes = if request.create_table && pg_compat_transfer && preserves_target_table_name && !target_table_preexisting { @@ -4267,7 +4446,7 @@ where // Count source rows let total_rows = { - let sql = count_sql(table, &request.source_schema, source_db_type); + let sql = count_sql(table, &request.source_schema, source_db_type, request.source_catalog.as_deref()); match execute_on_pool(state, source_pool_key, &sql).await { Ok(result) => result.rows.first().and_then(|r| r.first()).and_then(|v| match v { serde_json::Value::Number(n) => n.as_u64(), @@ -4309,26 +4488,58 @@ where let can_reuse_source_ddl = can_reuse_source_table_ddl(source_db_type, target_db_type, preserves_target_table_name); let ddl = if can_reuse_source_ddl { - let source_ddl = crate::schema::get_table_ddl_core( - state, - &request.source_connection_id, - &request.source_database, - &request.source_schema, - table, - None, - ) - .await - .unwrap_or_else(|_| { - generate_create_table_ddl( - &columns, - &target_table, + let source_ddl = if let Some(catalog) = + resolve_external_transfer_catalog(request.source_catalog.as_deref(), source_db_type) + { + // Doris/StarRocks external catalog: read DDL directly via + // SHOW CREATE TABLE catalog.database.table using the + // existing source pool (bare MySQL — addresses any catalog). + let pool = { + let connections = state.connections.read().await; + let pool = + connections.get(source_pool_key).ok_or_else(|| "Source pool not found".to_string())?; + let PoolKind::Mysql(p, _) = pool else { + return Err("Source pool must be MySQL-family for catalog DDL".to_string()); + }; + p.clone() + }; + db::doris::get_catalog_table_ddl(&pool, catalog, &request.source_database, table).await + .unwrap_or_else(|err| { + log::warn!("[transfer] catalog DDL read failed for {table} in catalog '{catalog}': {err}; falling back to generated DDL"); + generate_create_table_ddl( + &columns, + &target_table, + &request.source_schema, + &request.target_schema, + target_db_type, + source_db_type, + table_comment.as_deref(), + request.target_catalog.as_deref(), + ) + }) + } else { + crate::schema::get_table_ddl_core( + state, + &request.source_connection_id, + &request.source_database, &request.source_schema, - &request.target_schema, - target_db_type, - source_db_type, - table_comment.as_deref(), + table, + None, ) - }); + .await + .unwrap_or_else(|_| { + generate_create_table_ddl( + &columns, + &target_table, + &request.source_schema, + &request.target_schema, + target_db_type, + source_db_type, + table_comment.as_deref(), + request.target_catalog.as_deref(), + ) + }) + }; rewrite_transfer_source_table_ddl( &source_ddl, &request.source_schema, @@ -4345,6 +4556,7 @@ where target_db_type, source_db_type, table_comment.as_deref(), + request.target_catalog.as_deref(), ) }; log::info!("[transfer] creating target table: {}", ddl.chars().take(200).collect::()); @@ -4383,7 +4595,8 @@ where // Truncate target if overwrite mode if request.mode == TransferMode::Overwrite { - let full_table = qualified_table(&target_table, &request.target_schema, target_db_type); + let full_table = + qualified_table(&target_table, &request.target_schema, target_db_type, request.target_catalog.as_deref()); let truncate_sql = match target_db_type { DatabaseType::Sqlite | DatabaseType::CloudflareD1 | DatabaseType::DuckDb => { format!("DELETE FROM {full_table}") @@ -4406,6 +4619,7 @@ where &request.target_database, &request.target_schema, &target_table, + request.target_catalog.as_deref(), ) .await .unwrap_or_default(); @@ -4433,6 +4647,7 @@ where &request.target_database, &request.target_schema, &target_table, + request.target_catalog.as_deref(), ) .await .unwrap_or_default(); @@ -4459,6 +4674,7 @@ where offset, batch_size, &primary_key_columns, + request.source_catalog.as_deref(), ); let result = execute_on_pool(state, source_pool_key, &sql).await?; let row_count = result.rows.len(); @@ -4476,6 +4692,7 @@ where &request.target_schema, target_db_type, &pk_columns, + request.target_catalog.as_deref(), )?; for (statement_index, batch_sql) in write_statements.iter().enumerate() { execute_transfer_write_statement( @@ -4822,7 +5039,7 @@ where error: None, terminal: false, }); - let full_table = qualified_table(&trigger.table_name, &request.target_schema, &DatabaseType::Postgres); + let full_table = qualified_table(&trigger.table_name, &request.target_schema, &DatabaseType::Postgres, None); let drop_sql = format!( "DROP TRIGGER IF EXISTS {} ON {full_table}", quote_identifier(&trigger.trigger_name, &DatabaseType::Postgres) @@ -4996,9 +5213,11 @@ mod tests { source_connection_id: "source".to_string(), source_database: "source_db".to_string(), source_schema: "source_schema".to_string(), + source_catalog: None, target_connection_id: "target".to_string(), target_database: "target_db".to_string(), target_schema: "target_schema".to_string(), + target_catalog: None, tables: tables.into_iter().map(str::to_string).collect(), create_table: true, mode: TransferMode::Append, @@ -5165,7 +5384,8 @@ mod tests { db::ColumnInfo { comment: None, ..test_column("age", "int") }, ]; - let ddl = generate_create_table_ddl(&cols, "users", "", "", &DatabaseType::Mysql, &DatabaseType::Mysql, None); + let ddl = + generate_create_table_ddl(&cols, "users", "", "", &DatabaseType::Mysql, &DatabaseType::Mysql, None, None); assert!(ddl.contains("COMMENT '用户ID'")); assert!(ddl.contains("COMMENT '用户姓名'")); @@ -5205,6 +5425,7 @@ mod tests { &DatabaseType::Postgres, &DatabaseType::Postgres, None, + None, ); assert!(ddl.contains("\"id\" integer GENERATED BY DEFAULT AS IDENTITY NOT NULL")); @@ -5230,6 +5451,7 @@ mod tests { &DatabaseType::Postgres, &DatabaseType::Postgres, None, + None, ); assert!( @@ -5249,6 +5471,7 @@ mod tests { &DatabaseType::Mysql, &DatabaseType::Mysql, Some("用户表"), + None, ); assert!(ddl.contains(") COMMENT='用户表'")); @@ -5259,7 +5482,8 @@ mod tests { let cols = vec![db::ColumnInfo { data_type: "text".to_string(), is_primary_key: true, ..test_column("id", "text") }]; - let ddl = generate_create_table_ddl(&cols, "logs", "", "", &DatabaseType::Mysql, &DatabaseType::Sqlite, None); + let ddl = + generate_create_table_ddl(&cols, "logs", "", "", &DatabaseType::Mysql, &DatabaseType::Sqlite, None, None); assert!(ddl.contains("PRIMARY KEY (`id`(255))")); assert!(ddl.contains("`id` TEXT")); @@ -5269,7 +5493,8 @@ mod tests { fn mysql_int_pk_no_prefix() { let cols = vec![db::ColumnInfo { is_primary_key: true, ..test_column("id", "int") }]; - let ddl = generate_create_table_ddl(&cols, "users", "", "", &DatabaseType::Mysql, &DatabaseType::Sqlite, None); + let ddl = + generate_create_table_ddl(&cols, "users", "", "", &DatabaseType::Mysql, &DatabaseType::Sqlite, None, None); assert!(ddl.contains("PRIMARY KEY (`id`)")); assert!(!ddl.contains("PRIMARY KEY (`id`(255))")); @@ -5379,7 +5604,8 @@ mod tests { let cols = vec![db::ColumnInfo { comment: Some("test".to_string()), ..test_column("col", "text") }]; // PostgreSQL target should NOT have inline COMMENT - let ddl = generate_create_table_ddl(&cols, "t", "", "", &DatabaseType::Postgres, &DatabaseType::Postgres, None); + let ddl = + generate_create_table_ddl(&cols, "t", "", "", &DatabaseType::Postgres, &DatabaseType::Postgres, None, None); assert!(!ddl.contains("COMMENT")); } @@ -5398,6 +5624,7 @@ mod tests { &DatabaseType::ClickHouse, &DatabaseType::ClickHouse, None, + None, ); // Must include ENGINE with ORDER BY using the PK columns @@ -5418,6 +5645,7 @@ mod tests { &DatabaseType::ClickHouse, &DatabaseType::ClickHouse, None, + None, ); assert!(ddl.contains("ENGINE = MergeTree() ORDER BY tuple()")); @@ -5436,6 +5664,7 @@ mod tests { &DatabaseType::ClickHouse, &DatabaseType::Dameng, None, + None, ); assert!(ddl.contains("`created_at` DateTime64(6)"), "ddl: {ddl}"); @@ -5482,6 +5711,7 @@ mod tests { &DatabaseType::Hive, &DatabaseType::Postgres, None, + None, ); assert!(ddl.contains("CREATE TABLE IF NOT EXISTS `warehouse`.`events`")); @@ -5541,6 +5771,7 @@ mod tests { 200, 100, &[String::from("id")], + None, ); assert_eq!(sql, "SELECT \"id\", \"name\" FROM \"public\".\"users\" ORDER BY \"id\" LIMIT 100 OFFSET 200"); @@ -5556,6 +5787,7 @@ mod tests { 200, 100, &[String::from("id")], + None, ); assert_eq!(sql, "SELECT `id`, `name` FROM `users` ORDER BY `id` LIMIT 200, 300"); @@ -5589,6 +5821,7 @@ mod tests { 200, 100, &[String::from("id")], + None, ); assert_eq!( @@ -5663,7 +5896,8 @@ mod tests { #[test] fn filtered_count_preserves_where() { - let sql = count_sql_with_where("users", "public", &DatabaseType::SapHana, Some("WHERE status = 'active'")); + let sql = + count_sql_with_where("users", "public", &DatabaseType::SapHana, Some("WHERE status = 'active'"), None); assert_eq!(sql, "SELECT COUNT(*) FROM \"public\".\"users\" WHERE (status = 'active')"); } @@ -5866,7 +6100,7 @@ mod tests { let owner_sql = format!( "ALTER SEQUENCE {} OWNED BY {}.{}", postgres_sequence_qualified_name("public", &sequence.name), - qualified_table(&sequence.owner_table, "public", &DatabaseType::Postgres), + qualified_table(&sequence.owner_table, "public", &DatabaseType::Postgres, None), quote_identifier(&sequence.owner_column, &DatabaseType::Postgres) ); @@ -5948,7 +6182,7 @@ mod tests { let owner_sql = format!( "ALTER SEQUENCE {} OWNED BY {}.{}", postgres_sequence_qualified_name("archive", &sequence.name), - qualified_table(&sequence.owner_table, "archive", &DatabaseType::Postgres), + qualified_table(&sequence.owner_table, "archive", &DatabaseType::Postgres, None), quote_identifier(&sequence.owner_column, &DatabaseType::Postgres) ); let sequence_sync_sql = generate_postgres_sequence_sync_sql(&columns, "it_quick_entry", "archive"); @@ -6054,23 +6288,22 @@ mod tests { "policies", "", &DatabaseType::Mysql, + None, ); assert_eq!(sql, "INSERT INTO `policies` (`insurance_start_time`) VALUES\n('2026-05-12 00:00:00')"); } #[test] - fn mysql_insert_omits_database_qualified_table_name() { - let sql = generate_insert_typed( - &[String::from("id")], - &[Some(String::from("int"))], - &[vec![json!(1)]], - "users", - "app", - &DatabaseType::Mysql, - ); + fn count_sql_uses_three_part_name_for_mysql_external_catalog() { + let sql = count_sql("t1", "ads", &DatabaseType::Mysql, Some("hive_catalog")); + assert_eq!(sql, "SELECT COUNT(*) FROM `hive_catalog`.`ads`.`t1`"); + } - assert_eq!(sql, "INSERT INTO `users` (`id`) VALUES\n(1)"); + #[test] + fn count_sql_uses_three_part_name_for_starrocks_external_catalog() { + let sql = count_sql("t1", "ads", &DatabaseType::StarRocks, Some("paimon")); + assert_eq!(sql, "SELECT COUNT(*) FROM `paimon`.`ads`.`t1`"); } #[test] @@ -6092,6 +6325,7 @@ mod tests { "policies", "", &DatabaseType::Mysql, + None, ); assert_eq!( @@ -6119,6 +6353,7 @@ mod tests { "events", "APP", &DatabaseType::Oracle, + None, ); assert_eq!( @@ -6169,6 +6404,7 @@ mod tests { "orders", "", &DatabaseType::Mysql, + None, ); assert_eq!( @@ -6187,6 +6423,7 @@ mod tests { "", &DatabaseType::Mysql, &[String::from("id")], + None, ); assert_eq!( @@ -6204,6 +6441,7 @@ mod tests { "customers", "dbo", &DatabaseType::SqlServer, + None, ); assert_eq!(sql, "INSERT INTO [dbo].[customers] ([name], [note]) VALUES\n(N'Tiếng Việt', N'O''Brien')"); @@ -6222,6 +6460,7 @@ mod tests { "notes", "dbo", &DatabaseType::SqlServer, + None, ); assert_eq!( @@ -6249,6 +6488,7 @@ mod tests { "test", "dbo", &DatabaseType::SqlServer, + None, ); assert_eq!( @@ -6266,6 +6506,7 @@ mod tests { "flags", "dbo", &DatabaseType::SqlServer, + None, ); assert_eq!(sql, "INSERT INTO [dbo].[flags] ([enabled], [deleted]) VALUES\n(1, 0)"); @@ -6280,6 +6521,7 @@ mod tests { "files", "dbo", &DatabaseType::SqlServer, + None, ); assert_eq!(sql, "INSERT INTO [dbo].[files] ([payload], [note]) VALUES\n(0x0001ABff, N'0x0001ABff')"); @@ -6294,6 +6536,7 @@ mod tests { "files", "dbo", &DatabaseType::SqlServer, + None, ); assert_eq!(sql, "INSERT INTO [dbo].[files] ([payload]) VALUES\n(CONVERT(varbinary(max), N'O''Brien'))"); @@ -6308,6 +6551,7 @@ mod tests { "flags", "DBX_TEST", &DatabaseType::Dameng, + None, ); assert_eq!( @@ -6328,6 +6572,7 @@ mod tests { "dbo", &DatabaseType::SqlServer, &[String::from("id")], + None, ); assert!(sql.contains("USING (VALUES\n(1, 1)\n)")); @@ -6344,6 +6589,7 @@ mod tests { "events", "public", &DatabaseType::Postgres, + None, ); assert_eq!( @@ -6362,6 +6608,7 @@ mod tests { "files", "public", &DatabaseType::Postgres, + None, ); assert_eq!( @@ -6380,6 +6627,7 @@ mod tests { "files", "public", &DatabaseType::Postgres, + None, ); assert_eq!( @@ -6403,6 +6651,7 @@ mod tests { "files", "", &DatabaseType::Mysql, + None, ); assert_eq!( @@ -6421,6 +6670,7 @@ mod tests { "files", "", &DatabaseType::Mysql, + None, ); assert_eq!( @@ -6439,6 +6689,7 @@ mod tests { "files", "", &DatabaseType::Mysql, + None, ); assert_eq!( @@ -6469,6 +6720,7 @@ mod tests { "notes", "public", &DatabaseType::Postgres, + None, ); assert_eq!( @@ -6487,6 +6739,7 @@ mod tests { "numbers", "public", &DatabaseType::Postgres, + None, ); assert_eq!( @@ -6531,6 +6784,7 @@ mod tests { "INSTR_CATEGORY", "APP", &DatabaseType::Oracle, + None, ); assert_eq!( @@ -6549,6 +6803,7 @@ mod tests { "INSTR_CATEGORY", "APP", &DatabaseType::Oracle, + None, ); assert_eq!( @@ -6572,6 +6827,7 @@ SELECT 1 FROM dual"# "APP", &DatabaseType::Oracle, &[], + None, ) .unwrap(); @@ -6593,6 +6849,7 @@ SELECT 1 FROM dual"# "", &DatabaseType::Mysql, &[], + None, ) .unwrap(); @@ -6612,6 +6869,7 @@ SELECT 1 FROM dual"# "events", "", &DatabaseType::Mysql, + None, limits, ) .unwrap(); @@ -6632,6 +6890,7 @@ SELECT 1 FROM dual"# "events", "", &DatabaseType::Mysql, + None, limits, ) .unwrap_err(); @@ -6651,6 +6910,7 @@ SELECT 1 FROM dual"# "events", "dbo", &DatabaseType::SqlServer, + None, SqlBatchLimits::for_database(&DatabaseType::SqlServer, rows.len()), ) .unwrap(); @@ -6669,6 +6929,7 @@ SELECT 1 FROM dual"# "events", "dbo", &DatabaseType::SqlServer, + None, SqlBatchLimits::for_database(&DatabaseType::SqlServer, rows.len()), ) .unwrap(); @@ -6693,6 +6954,7 @@ SELECT 1 FROM dual"# "", &DatabaseType::Mysql, &[String::from("id")], + None, ) .unwrap(); @@ -6705,6 +6967,82 @@ SELECT 1 FROM dual"# assert_eq!(database_from_pool_key("conn:analytics"), Some("analytics")); assert_eq!(database_from_pool_key("conn:analytics:session:editor-1"), Some("analytics")); assert_eq!(database_from_pool_key("conn"), None); + assert_eq!(database_from_pool_key("conn:analytics:catalog:hive"), Some("analytics")); + assert_eq!(database_from_pool_key("conn:catalog:hive"), None); + assert_eq!(catalog_from_pool_key("conn:catalog:hive"), Some("hive")); + assert_eq!(catalog_from_pool_key("conn:ads:catalog:paimon"), Some("paimon")); + assert_eq!(catalog_from_pool_key("conn:ads"), None); + } + + #[test] + fn resolve_external_transfer_catalog_skips_builtin_catalogs() { + assert_eq!(resolve_external_transfer_catalog(Some("hive"), &DatabaseType::StarRocks), Some("hive")); + assert_eq!(resolve_external_transfer_catalog(Some("default_catalog"), &DatabaseType::StarRocks), None); + assert_eq!(resolve_external_transfer_catalog(Some("internal"), &DatabaseType::Doris), None); + // MySQL db_type is included so StarRocks saved as mysql+driver_profile still + // gets 3-part catalog qualification (UI only sends catalog for capable engines). + assert_eq!(resolve_external_transfer_catalog(Some("hive"), &DatabaseType::Mysql), Some("hive")); + assert_eq!(resolve_external_transfer_catalog(Some("hive"), &DatabaseType::Postgres), None); + } + + #[test] + fn resolve_external_transfer_catalog_for_config_accepts_starrocks_driver_profile() { + let config = crate::models::connection::ConnectionConfig { + id: "sr".to_string(), + name: "sr".to_string(), + note: String::new(), + db_type: DatabaseType::Mysql, + driver_profile: Some("starrocks".to_string()), + driver_label: None, + url_params: None, + agent_java_options: Vec::new(), + host: "127.0.0.1".to_string(), + port: 9030, + username: String::new(), + password: String::new(), + database: None, + visible_databases: None, + visible_schemas: None, + show_system_schemas: false, + attached_databases: Vec::new(), + init_script: None, + color: None, + transport_layers: Vec::new(), + connect_timeout_secs: 5, + query_timeout_secs: 30, + idle_timeout_secs: 60, + keepalive_interval_secs: 0, + ssl: false, + ca_cert_path: String::new(), + client_cert_path: String::new(), + client_key_path: String::new(), + sysdba: false, + oracle_connection_type: None, + connection_string: None, + redis_connection_mode: None, + redis_sentinel_master: String::new(), + redis_sentinel_nodes: String::new(), + redis_sentinel_username: String::new(), + redis_sentinel_password: String::new(), + redis_sentinel_tls: false, + redis_cluster_nodes: String::new(), + redis_key_separator: crate::models::connection::default_redis_key_separator(), + redis_scan_page_size: None, + redis_database_aliases: Default::default(), + etcd_endpoints: String::new(), + gbase_server: String::new(), + informix_server: String::new(), + external_config: None, + jdbc_driver_class: None, + jdbc_driver_paths: Vec::new(), + one_time: false, + read_only: false, + is_production: false, + production_databases: vec![], + database_info: None, + }; + assert_eq!(resolve_external_transfer_catalog_for_config(Some("paimon"), &config), Some("paimon")); + assert_eq!(resolve_external_transfer_catalog_for_config(Some("default_catalog"), &config), None); } #[test] @@ -6809,7 +7147,8 @@ SELECT 1 FROM dual"# db::ColumnInfo { is_nullable: false, ..test_column("name", "varchar(64)") }, ]; - let ddl = generate_create_table_ddl(&cols, "users", "", "", &DatabaseType::Mysql, &DatabaseType::Mysql, None); + let ddl = + generate_create_table_ddl(&cols, "users", "", "", &DatabaseType::Mysql, &DatabaseType::Mysql, None, None); assert!(ddl.contains("`id` INT NOT NULL AUTO_INCREMENT"), "ddl: {ddl}"); assert!(ddl.contains("PRIMARY KEY (`id`)"), "ddl: {ddl}"); @@ -6823,7 +7162,8 @@ SELECT 1 FROM dual"# ..test_column("status", "tinyint") }]; - let ddl = generate_create_table_ddl(&cols, "items", "", "", &DatabaseType::Mysql, &DatabaseType::Mysql, None); + let ddl = + generate_create_table_ddl(&cols, "items", "", "", &DatabaseType::Mysql, &DatabaseType::Mysql, None, None); assert!(ddl.contains("DEFAULT 0"), "ddl: {ddl}"); assert!(!ddl.contains("'0'"), "ddl should not quote numeric default: {ddl}"); @@ -6834,7 +7174,8 @@ SELECT 1 FROM dual"# let cols = vec![db::ColumnInfo { column_default: Some("o'clock".to_string()), ..test_column("label", "varchar(32)") }]; - let ddl = generate_create_table_ddl(&cols, "items", "", "", &DatabaseType::Mysql, &DatabaseType::Mysql, None); + let ddl = + generate_create_table_ddl(&cols, "items", "", "", &DatabaseType::Mysql, &DatabaseType::Mysql, None, None); assert!(ddl.contains("DEFAULT 'o''clock'"), "ddl: {ddl}"); } @@ -6848,7 +7189,8 @@ SELECT 1 FROM dual"# ..test_column("updated_at", "timestamp") }]; - let ddl = generate_create_table_ddl(&cols, "items", "", "", &DatabaseType::Mysql, &DatabaseType::Mysql, None); + let ddl = + generate_create_table_ddl(&cols, "items", "", "", &DatabaseType::Mysql, &DatabaseType::Mysql, None, None); assert!(ddl.contains("DEFAULT CURRENT_TIMESTAMP"), "ddl: {ddl}"); assert!(ddl.contains("ON UPDATE CURRENT_TIMESTAMP"), "ddl: {ddl}"); @@ -6864,7 +7206,8 @@ SELECT 1 FROM dual"# ..test_column("created_at", "timestamp(6)") }]; - let ddl = generate_create_table_ddl(&cols, "items", "", "", &DatabaseType::Mysql, &DatabaseType::Mysql, None); + let ddl = + generate_create_table_ddl(&cols, "items", "", "", &DatabaseType::Mysql, &DatabaseType::Mysql, None, None); assert!(ddl.contains("DEFAULT CURRENT_TIMESTAMP(6)"), "ddl: {ddl}"); } @@ -6877,7 +7220,8 @@ SELECT 1 FROM dual"# ..test_column("touched_at", "timestamp(3)") }]; - let ddl = generate_create_table_ddl(&cols, "items", "", "", &DatabaseType::Mysql, &DatabaseType::Mysql, None); + let ddl = + generate_create_table_ddl(&cols, "items", "", "", &DatabaseType::Mysql, &DatabaseType::Mysql, None, None); assert!(ddl.contains("ON UPDATE CURRENT_TIMESTAMP(3)"), "ddl: {ddl}"); assert!(!ddl.contains("DEFAULT"), "ddl should not emit DEFAULT when none was set: {ddl}"); @@ -6892,7 +7236,8 @@ SELECT 1 FROM dual"# ..test_column("id", "int") }]; - let ddl = generate_create_table_ddl(&cols, "users", "", "", &DatabaseType::Sqlite, &DatabaseType::Mysql, None); + let ddl = + generate_create_table_ddl(&cols, "users", "", "", &DatabaseType::Sqlite, &DatabaseType::Mysql, None, None); assert!(!ddl.contains("AUTO_INCREMENT"), "non-mysql target should not emit AUTO_INCREMENT: {ddl}"); } @@ -6915,6 +7260,7 @@ SELECT 1 FROM dual"# &DatabaseType::Postgres, &DatabaseType::Postgres, None, + None, ); assert!(ddl.contains("GENERATED BY DEFAULT AS IDENTITY"), "ddl: {ddl}"); @@ -6938,6 +7284,7 @@ SELECT 1 FROM dual"# &DatabaseType::Postgres, &DatabaseType::Postgres, None, + None, ); assert!(ddl.contains("\"id\" integer generated by default as identity NOT NULL"), "ddl: {ddl}"); @@ -6969,6 +7316,7 @@ SELECT 1 FROM dual"# &DatabaseType::Kingbase, &DatabaseType::Postgres, None, + None, ); assert!(ddl.contains("GENERATED BY DEFAULT AS IDENTITY"), "ddl: {ddl}"); @@ -6984,6 +7332,7 @@ SELECT 1 FROM dual"# "public", &DatabaseType::Kingbase, &[String::from("id")], + None, ); assert!(sql.contains("ON CONFLICT (\"id\") DO UPDATE SET \"name\" = EXCLUDED.\"name\""), "sql: {sql}"); diff --git a/crates/dbx-core/tests/database_capabilities.rs b/crates/dbx-core/tests/database_capabilities.rs index 31585d88f0..1d09790d60 100644 --- a/crates/dbx-core/tests/database_capabilities.rs +++ b/crates/dbx-core/tests/database_capabilities.rs @@ -182,8 +182,9 @@ fn identifies_single_connection_pool_types() { #[test] fn identifies_metadata_connections_that_drop_database_scope() { assert!(is_metadata_connection_scoped(&DatabaseType::Mysql)); - assert!(!is_metadata_connection_scoped(&DatabaseType::Doris)); - assert!(!is_metadata_connection_scoped(&DatabaseType::StarRocks)); + assert!(is_metadata_connection_scoped(&DatabaseType::Doris)); + assert!(is_metadata_connection_scoped(&DatabaseType::StarRocks)); + assert!(is_metadata_connection_scoped(&DatabaseType::ManticoreSearch)); assert!(!is_metadata_connection_scoped(&DatabaseType::Postgres)); assert!(!is_metadata_connection_scoped(&DatabaseType::Kwdb)); assert!(!is_metadata_connection_scoped(&DatabaseType::Oracle)); diff --git a/crates/dbx-core/tests/live_postgres_transfer.rs b/crates/dbx-core/tests/live_postgres_transfer.rs index b56547c15e..321af5f0cd 100644 --- a/crates/dbx-core/tests/live_postgres_transfer.rs +++ b/crates/dbx-core/tests/live_postgres_transfer.rs @@ -211,9 +211,11 @@ async fn live_postgres_transfer_preserves_data_and_schema_objects() { source_connection_id: source_connection_id.to_string(), source_database: source_database.clone(), source_schema: source_schema.clone(), + source_catalog: None, target_connection_id: target_connection_id.to_string(), target_database: target_database.clone(), target_schema: target_schema.clone(), + target_catalog: None, tables: vec!["users".to_string(), "audit_logs".to_string(), "files".to_string()], create_table: true, mode: TransferMode::Append, @@ -492,9 +494,11 @@ async fn live_postgres_transfer_skips_create_ddl_for_existing_target_table() { source_connection_id: source_connection_id.to_string(), source_database: source_database.clone(), source_schema: source_schema.clone(), + source_catalog: None, target_connection_id: target_connection_id.to_string(), target_database: target_database.clone(), target_schema: target_schema.clone(), + target_catalog: None, tables: vec!["items".to_string()], create_table: true, mode: TransferMode::Append, diff --git a/crates/dbx-core/tests/live_sqlserver_completion.rs b/crates/dbx-core/tests/live_sqlserver_completion.rs index 555234fffd..aeb0fb895b 100644 --- a/crates/dbx-core/tests/live_sqlserver_completion.rs +++ b/crates/dbx-core/tests/live_sqlserver_completion.rs @@ -1354,9 +1354,11 @@ async fn live_sqlserver_transfer_table_skips_rowversion_insert_column() { source_connection_id: "live-sqlserver-rowversion".to_string(), source_database: database.clone(), source_schema: "dbo".to_string(), + source_catalog: None, target_connection_id: "live-sqlserver-rowversion".to_string(), target_database: target_database.clone(), target_schema: "dbo".to_string(), + target_catalog: None, tables: vec![source_table.clone()], create_table: true, mode: dbx_core::transfer::TransferMode::Append, diff --git a/crates/dbx-web/src/routes/transfer.rs b/crates/dbx-web/src/routes/transfer.rs index 5195fd0c35..c000bcf8ae 100644 --- a/crates/dbx-web/src/routes/transfer.rs +++ b/crates/dbx-web/src/routes/transfer.rs @@ -116,7 +116,15 @@ pub async fn start_transfer( } }; - let source_pool_key = match app.get_or_create_pool(&req.source_connection_id, Some(&req.source_database)).await + // External Doris/StarRocks catalogs: pool is created with `catalog=` URL + // setup (SET catalog) and without USE . See ensure_transfer_pool. + let source_pool_key = match transfer::ensure_transfer_pool( + &app, + &req.source_connection_id, + &req.source_database, + req.source_catalog.as_deref(), + ) + .await { Ok(k) => k, Err(e) => { @@ -125,7 +133,13 @@ pub async fn start_transfer( return; } }; - let target_pool_key = match app.get_or_create_pool(&req.target_connection_id, Some(&req.target_database)).await + let target_pool_key = match transfer::ensure_transfer_pool( + &app, + &req.target_connection_id, + &req.target_database, + req.target_catalog.as_deref(), + ) + .await { Ok(k) => k, Err(e) => { @@ -137,19 +151,36 @@ pub async fn start_transfer( let tables = req.tables.clone(); // Sort by FK dependency so referenced tables are transferred first. - let tables = transfer::sort_tables_by_fk_dependency( - &app, - &req.source_connection_id, - &req.source_database, - &req.source_schema, - &tables, - true, - ) - .await - .unwrap_or_else(|e| { - log::warn!("[transfer] failed to sort tables by FK dependency, using original order: {e}"); - tables - }); + // Skip for external Doris/StarRocks catalogs — the database name does + // not exist in the default catalog and sorting is unnecessary. + let tables = { + let skip_fk_sort = { + let configs = app.configs.read().await; + configs + .get(&req.source_connection_id) + .and_then(|config| { + transfer::resolve_external_transfer_catalog_for_config(req.source_catalog.as_deref(), config) + }) + .is_some() + }; + if skip_fk_sort { + tables + } else { + transfer::sort_tables_by_fk_dependency( + &app, + &req.source_connection_id, + &req.source_database, + &req.source_schema, + &tables, + true, + ) + .await + .unwrap_or_else(|e| { + log::warn!("[transfer] failed to sort tables by FK dependency, using original order: {e}"); + tables + }) + } + }; let mut failed_tables: Vec = Vec::new(); if matches!(source_db_type, dbx_core::models::connection::DatabaseType::Postgres) @@ -375,16 +406,22 @@ pub async fn preview_transfer_ownership( transfer::validate_transfer_target_table_names(&req).map_err(AppError::from)?; let source_db_type = transfer::get_db_type(&state.app, &req.source_connection_id).await.map_err(AppError::from)?; let target_db_type = transfer::get_db_type(&state.app, &req.target_connection_id).await.map_err(AppError::from)?; - let source_pool_key = state - .app - .get_or_create_pool(&req.source_connection_id, Some(&req.source_database)) - .await - .map_err(AppError::from)?; - let target_pool_key = state - .app - .get_or_create_pool(&req.target_connection_id, Some(&req.target_database)) - .await - .map_err(AppError::from)?; + let source_pool_key = transfer::ensure_transfer_pool( + &state.app, + &req.source_connection_id, + &req.source_database, + req.source_catalog.as_deref(), + ) + .await + .map_err(AppError::from)?; + let target_pool_key = transfer::ensure_transfer_pool( + &state.app, + &req.target_connection_id, + &req.target_database, + req.target_catalog.as_deref(), + ) + .await + .map_err(AppError::from)?; let preview = transfer::preview_transfer_ownership( &state.app, &req, diff --git a/packages/app-tests/treeRefreshTarget.test.ts b/packages/app-tests/treeRefreshTarget.test.ts index 84fe38b684..5d6a54739a 100644 --- a/packages/app-tests/treeRefreshTarget.test.ts +++ b/packages/app-tests/treeRefreshTarget.test.ts @@ -34,3 +34,42 @@ test("finds database refresh targets inside grouped sidebar trees", () => { test("returns null when the target database node is not loaded", () => { assert.equal(findDatabaseTreeNode([], "conn-1", "app"), null); }); + +test("keeps same-named databases isolated by catalog", () => { + const hive: TreeNode = { + id: "conn-1:doris-catalog:hive:sales", + label: "sales", + type: "database", + connectionId: "conn-1", + catalog: "hive", + database: "sales", + }; + const iceberg: TreeNode = { + id: "conn-1:doris-catalog:iceberg:sales", + label: "sales", + type: "database", + connectionId: "conn-1", + catalog: "iceberg", + database: "sales", + }; + const internal: TreeNode = { + id: "conn-1:sales", + label: "sales", + type: "database", + connectionId: "conn-1", + database: "sales", + }; + const nodes: TreeNode[] = [ + { + id: "conn-1", + label: "StarRocks", + type: "connection", + connectionId: "conn-1", + children: [internal, { id: "catalog-hive", label: "hive", type: "doris-catalog", connectionId: "conn-1", catalog: "hive", children: [hive] }, { id: "catalog-iceberg", label: "iceberg", type: "doris-catalog", connectionId: "conn-1", catalog: "iceberg", children: [iceberg] }], + }, + ]; + + assert.equal(findDatabaseTreeNode(nodes, "conn-1", "sales"), internal); + assert.equal(findDatabaseTreeNode(nodes, "conn-1", "sales", "hive"), hive); + assert.equal(findDatabaseTreeNode(nodes, "conn-1", "sales", "iceberg"), iceberg); +}); diff --git a/src-tauri/src/commands/transfer.rs b/src-tauri/src/commands/transfer.rs index 6a940139a4..a69742f2d3 100644 --- a/src-tauri/src/commands/transfer.rs +++ b/src-tauri/src/commands/transfer.rs @@ -29,27 +29,58 @@ pub async fn start_transfer( let target_db_type = get_db_type(&state, &request.target_connection_id).await?; dbx_core::transfer::validate_transfer_target_table_names(&request)?; - // Ensure pools - let source_pool_key = - state.get_or_create_pool(&request.source_connection_id, Some(&request.source_database)).await?; - let target_pool_key = - state.get_or_create_pool(&request.target_connection_id, Some(&request.target_database)).await?; + // External Doris/StarRocks catalogs: pool is created with `catalog=` URL + // setup (SET catalog) and without USE . See ensure_transfer_pool. + let source_pool_key = dbx_core::transfer::ensure_transfer_pool( + &state, + &request.source_connection_id, + &request.source_database, + request.source_catalog.as_deref(), + ) + .await?; + let target_pool_key = dbx_core::transfer::ensure_transfer_pool( + &state, + &request.target_connection_id, + &request.target_database, + request.target_catalog.as_deref(), + ) + .await?; tokio::spawn(async move { // Sort tables by FK dependency so referenced tables are transferred first. - let sorted_tables = dbx_core::transfer::sort_tables_by_fk_dependency( - &state, - &request.source_connection_id, - &request.source_database, - &request.source_schema, - &request.tables, - true, - ) - .await - .unwrap_or_else(|e| { - log::warn!("[transfer] failed to sort tables by FK dependency, using original order: {e}"); - request.tables.clone() - }); + // Skip for external Doris/StarRocks catalogs — the database name does not + // exist in the default catalog and sorting is unnecessary (no FK constraints). + let sorted_tables = { + let skip_fk_sort = { + let configs = state.configs.read().await; + configs + .get(&request.source_connection_id) + .and_then(|config| { + dbx_core::transfer::resolve_external_transfer_catalog_for_config( + request.source_catalog.as_deref(), + config, + ) + }) + .is_some() + }; + if skip_fk_sort { + request.tables.clone() + } else { + dbx_core::transfer::sort_tables_by_fk_dependency( + &state, + &request.source_connection_id, + &request.source_database, + &request.source_schema, + &request.tables, + true, + ) + .await + .unwrap_or_else(|e| { + log::warn!("[transfer] failed to sort tables by FK dependency, using original order: {e}"); + request.tables.clone() + }) + } + }; let total_tables = sorted_tables.len(); log::info!("[transfer] starting transfer_id={} tables={}", transfer_id, total_tables); @@ -291,10 +322,20 @@ pub async fn preview_transfer_ownership( let source_db_type = get_db_type(&state, &request.source_connection_id).await?; let target_db_type = get_db_type(&state, &request.target_connection_id).await?; dbx_core::transfer::validate_transfer_target_table_names(&request)?; - let source_pool_key = - state.get_or_create_pool(&request.source_connection_id, Some(&request.source_database)).await?; - let target_pool_key = - state.get_or_create_pool(&request.target_connection_id, Some(&request.target_database)).await?; + let source_pool_key = dbx_core::transfer::ensure_transfer_pool( + &state, + &request.source_connection_id, + &request.source_database, + request.source_catalog.as_deref(), + ) + .await?; + let target_pool_key = dbx_core::transfer::ensure_transfer_pool( + &state, + &request.target_connection_id, + &request.target_database, + request.target_catalog.as_deref(), + ) + .await?; dbx_core::transfer::preview_transfer_ownership( &state,