fix(data): harden transfer and comparison workflows

This commit is contained in:
t8y2
2026-10-13 16:15:53 +08:00
parent aaa8e15791
commit 47b7b492d3
88 changed files with 1564 additions and 1156 deletions
@@ -833,6 +833,7 @@ function toggleShowAll(table: DataCompareTableResult, kind: DiffKind) {
function setDiffSelection(kind: DiffKind, selected: boolean) {
if (executionLocked.value) return;
batchResults.value.forEach((table) => {
if (!table.syncEligible) return;
table.diff[kind].forEach((row) => {
row.selected = selected;
});
@@ -841,7 +842,7 @@ function setDiffSelection(kind: DiffKind, selected: boolean) {
}
function setTableDiffSelection(table: DataCompareTableResult, kind: DiffKind, selected: boolean) {
if (executionLocked.value) return;
if (executionLocked.value || !table.syncEligible) return;
table.diff[kind].forEach((row) => {
row.selected = selected;
});
@@ -860,8 +861,8 @@ function clearAllSelections() {
rebuildSyncPlan().catch((e) => toast(String(e), 5000));
}
function toggleRowSelection(row: SelectableDataCompareRow | SelectableDataCompareModifiedRow) {
if (executionLocked.value) return;
function toggleRowSelection(table: DataCompareTableResult, row: SelectableDataCompareRow | SelectableDataCompareModifiedRow) {
if (executionLocked.value || !table.syncEligible) return;
row.selected = !row.selected;
rebuildSyncPlan().catch((e) => toast(String(e), 5000));
}
@@ -1631,10 +1632,10 @@ onBeforeUnmount(() => {
<span class="font-medium">{{ t(`diff.${kind}`) }}</span>
<span class="text-muted-foreground">{{ selectedRows(item, kind) }}/{{ item.diff[kind].length }}</span>
<span class="flex-1" />
<Button size="sm" variant="ghost" class="h-6 px-2 text-xs" @click="setTableDiffSelection(item, kind, true)">
<Button size="sm" variant="ghost" class="h-6 px-2 text-xs" :disabled="!item.syncEligible" @click="setTableDiffSelection(item, kind, true)">
{{ t("dataCompare.selectAllKind", { kind: t(`diff.${kind}`) }) }}
</Button>
<Button size="sm" variant="ghost" class="h-6 px-2 text-xs" @click="setTableDiffSelection(item, kind, false)">
<Button size="sm" variant="ghost" class="h-6 px-2 text-xs" :disabled="!item.syncEligible" @click="setTableDiffSelection(item, kind, false)">
{{ t("dataCompare.clearKind", { kind: t(`diff.${kind}`) }) }}
</Button>
<Button v-if="item.diff[kind].length > detailPreviewLimitNumber" size="sm" variant="ghost" class="h-6 px-2 text-xs" @click="toggleShowAll(item, kind)">
@@ -1643,7 +1644,14 @@ onBeforeUnmount(() => {
</div>
<div class="max-h-72 overflow-auto divide-y">
<button v-for="row in rowsForDisplay(item, kind)" :key="`${item.sourceTable}:${kind}:${row.key}`" type="button" class="flex w-full items-start gap-3 px-3 py-2 text-left text-xs hover:bg-muted/40" @click="toggleRowSelection(row)">
<button
v-for="row in rowsForDisplay(item, kind)"
:key="`${item.sourceTable}:${kind}:${row.key}`"
type="button"
class="flex w-full items-start gap-3 px-3 py-2 text-left text-xs hover:bg-muted/40 disabled:cursor-not-allowed disabled:opacity-60"
:disabled="!item.syncEligible"
@click="toggleRowSelection(item, row)"
>
<CheckSquare v-if="row.selected" class="mt-0.5 h-3.5 w-3.5 shrink-0 text-primary" />
<Square v-else class="mt-0.5 h-3.5 w-3.5 shrink-0 text-muted-foreground/40" />
<div class="min-w-0 flex-1">
@@ -206,6 +206,7 @@ function preparationFor(table: string) {
sourceTruncated: false,
targetTruncated: false,
rowBudget: 100_000,
syncEligible: true,
};
}
@@ -130,6 +130,7 @@ function completedSession(): DataCompareSession {
sourceTruncated: false,
targetTruncated: false,
rowBudget: 100_000,
syncEligible: true,
databaseType: "oracle",
diff: {
added: [
@@ -20,7 +20,7 @@ vi.mock("@/lib/backend/api", () => apiMock);
vi.mock("@/composables/useDialogSources", () => ({ openDataCompareSession: openMock }));
vi.mock("@/composables/useExportTracker", () => ({ useExportTracker: () => trackerMock }));
const { normalizeKeyColumnOverrides, startDataCompareSession } = await import("../useDataCompareSession.ts");
const { buildDataCompareSyncPlanTables, normalizeKeyColumnOverrides, startDataCompareSession } = await import("../useDataCompareSession.ts");
function column(name: string, isPrimaryKey = false): ColumnInfo {
return { name, data_type: "TEXT", is_nullable: true, column_default: null, is_primary_key: isPrimaryKey, extra: null };
@@ -45,6 +45,7 @@ function preparationFor(table: string) {
sourceTruncated: false,
targetTruncated: false,
rowBudget: 100_000,
syncEligible: true,
};
}
@@ -228,13 +229,16 @@ test("keeps the configured per-table match columns after cloning the session con
});
test("carries the row budget and truncation flags from the backend into the table result", async () => {
apiMock.prepareDataCompareFromTables.mockImplementation((options: { targetTable: string }) => Promise.resolve({ ...preparationFor(options.targetTable), sourceRowCount: 4_000_000, sourceTruncated: true, targetTruncated: true, rowBudget: 100_000 }));
apiMock.prepareDataCompareFromTables.mockImplementation((options: { targetTable: string }) => Promise.resolve({ ...preparationFor(options.targetTable), sourceRowCount: 4_000_000, sourceTruncated: true, targetTruncated: true, rowBudget: 100_000, syncEligible: false }));
const session = await runSession(baseConfig({ selectedSourceTables: ["users"], targetTable: "users" }), [{ sourceTable: "users", targetTable: "users" }]);
assert.equal(session.batchResults[0]?.rowBudget, 100_000);
assert.equal(session.batchResults[0]?.sourceTruncated, true);
assert.equal(session.batchResults[0]?.targetTruncated, true);
assert.equal(session.batchResults[0]?.syncEligible, false);
assert.equal(buildDataCompareSyncPlanTables(session.batchResults, "public").length, 0);
assert.equal(session.batchResults[0]?.diff.added[0]?.selected, false);
});
test("keeps a truncated missing-target table on the same row budget", async () => {
@@ -243,12 +247,14 @@ test("keeps a truncated missing-target table on the same row budget", async () =
sourceRowCount: 8_000_000,
sourceTruncated: true,
rowBudget: 100_000,
syncEligible: false,
});
const session = await runSession(baseConfig({ selectedSourceTables: ["users"], targetTable: "users", targetTables: ["orders"] }), [{ sourceTable: "users", targetTable: "users" }]);
assert.equal(session.batchResults[0]?.rowBudget, 100_000);
assert.equal(session.batchResults[0]?.sourceTruncated, true);
assert.equal(buildDataCompareSyncPlanTables(session.batchResults, "public").length, 0);
});
test("normalizes a per-table match-column map", () => {
@@ -59,8 +59,9 @@ export interface DataCompareTableResult {
targetRowCount: number;
sourceTruncated: boolean;
targetTruncated: boolean;
/** Rows the backend was allowed to read per side; the diff only covers this many rows when a side is truncated. */
/** Rows the backend was allowed to inspect per side; incomplete results are never synchronization-eligible. */
rowBudget: number;
syncEligible: boolean;
databaseType?: DatabaseType;
preSyncStatements?: string[];
diff: SelectableDataCompareResult;
@@ -179,17 +180,17 @@ function compareErrorMessage(dependencies: DataCompareSessionDependencies, kind:
return dependencies.formatError?.(kind, columns) ?? fallbackCompareErrorMessage(kind, columns);
}
function toSelectableDiff(diff: DataCompareResult): SelectableDataCompareResult {
function toSelectableDiff(diff: DataCompareResult, selected: boolean): SelectableDataCompareResult {
return {
added: diff.added.map((row) => ({ ...row, selected: true })),
removed: diff.removed.map((row) => ({ ...row, selected: true })),
modified: diff.modified.map((row) => ({ ...row, selected: true })),
added: diff.added.map((row) => ({ ...row, selected })),
removed: diff.removed.map((row) => ({ ...row, selected })),
modified: diff.modified.map((row) => ({ ...row, selected })),
};
}
export function buildDataCompareSyncPlanTables(results: DataCompareTableResult[], targetSchema: string): DataCompareSyncPlanTableOptions[] {
return results
.filter((table) => table.status === "different")
.filter((table) => table.status === "different" && table.syncEligible)
.map((table) => ({
tableName: table.targetTable,
schema: targetSchema,
@@ -203,6 +204,7 @@ export function buildDataCompareSyncPlanTables(results: DataCompareTableResult[]
},
databaseType: table.databaseType,
preSyncStatements: table.preSyncStatements ?? [],
syncEligible: true,
}))
.filter((table) => table.preSyncStatements.length > 0 || table.diff.added.length > 0 || table.diff.removed.length > 0 || table.diff.modified.length > 0);
}
@@ -267,9 +269,10 @@ function buildTableResult(task: DataCompareTableTask, preparation: DataCompareFr
sourceTruncated: preparation.sourceTruncated,
targetTruncated: preparation.targetTruncated,
rowBudget: preparation.rowBudget,
syncEligible: preparation.syncEligible,
databaseType,
preSyncStatements: preparation.preSyncStatements,
diff: toSelectableDiff(preparation.result),
diff: toSelectableDiff(preparation.result, preparation.syncEligible),
expanded: status === "different",
showAll: { added: false, removed: false, modified: false },
};
@@ -291,9 +294,10 @@ function buildMissingTargetResult(task: DataCompareTableTask, preparation: DataC
sourceTruncated: preparation.sourceTruncated,
targetTruncated: false,
rowBudget: preparation.rowBudget,
syncEligible: preparation.syncEligible,
databaseType,
preSyncStatements: preparation.preSyncStatements,
diff: toSelectableDiff(preparation.result),
diff: toSelectableDiff(preparation.result, preparation.syncEligible),
expanded: preparation.result.added.length > 0,
showAll: { added: false, removed: false, modified: false },
};
@@ -315,6 +319,7 @@ function buildErrorResult(task: DataCompareTableTask, keyColumns: string[], data
sourceTruncated: false,
targetTruncated: false,
rowBudget: 0,
syncEligible: false,
databaseType,
preSyncStatements: [],
diff: { added: [], removed: [], modified: [] },
+2 -2
View File
@@ -8223,9 +8223,9 @@ export default {
noDifferences: "Data is identical, no sync SQL needed",
syncSuccess: "Data sync executed successfully",
truncatedTitle: "Some tables were not compared in full",
truncatedHint: "Tables with more rows than a single compare can read were compared only over their first rows, so added, modified, or deleted rows beyond that point are not reported. Narrow the range with a filter and compare in batches.",
truncatedHint: "Tables above the single-compare limit were inspected only over a subset, so changes outside it are not reported. Synchronization SQL is disabled for these results; narrow the range and run complete comparisons in batches.",
truncatedTable: "{table}: {source} source rows, {target} target rows, compared the first {budget} rows only",
truncatedBadge: "Compared the first {budget} rows only",
truncatedBadge: "Inspected a subset (up to {budget} rows per side); synchronization disabled",
},
settings: {
historyRetentionLimit: "Query history retention",
+2 -2
View File
@@ -8206,9 +8206,9 @@ export default withEnglishFallback({
noDifferences: "数据完全一致,无需生成同步 SQL",
syncSuccess: "数据同步执行成功",
truncatedTitle: "部分表未完整比较",
truncatedHint: "行数超过单次比较上限的表只比较了前若干行,超出部分的新增、修改和删除不会被发现。建议加上过滤条件缩小范围后分批比较。",
truncatedHint: "行数超过单次比较上限的表只检查了部分数据,未检查范围内的新增、修改和删除不会被发现,因此不会生成或执行同步 SQL。建议缩小范围后分批完整比较。",
truncatedTable: "{table}:源 {source} 行、目标 {target} 行,仅比较前 {budget} 行",
truncatedBadge: "仅比较前 {budget} 行",
truncatedBadge: "仅检查部分数据(单侧上限 {budget} 行),同步已禁用",
},
settings: {
historyRetentionLimit: "查询历史保留条数",
@@ -1,5 +1,16 @@
import { describe, expect, it } from "vitest";
import { databaseConnectionFormKind, databaseDefaultPort, databaseManifestEntry, databaseProductCapabilities, databaseRuntimeMode, supportsDatabaseFeature, usesAgentCursorForQuery, usesAgentCursorForTableData } from "@/lib/database/databaseDriverManifest";
import {
databaseConnectionFormKind,
databaseDefaultPort,
databaseManifestEntry,
databaseProductCapabilities,
databaseRuntimeMode,
databaseTransferSupportLevel,
supportsDatabaseCompareMode,
supportsDatabaseFeature,
usesAgentCursorForQuery,
usesAgentCursorForTableData,
} from "@/lib/database/databaseDriverManifest";
describe("databaseDriverManifest", () => {
it("advertises implemented Xugu capabilities without enabling unsupported features", () => {
@@ -61,6 +72,20 @@ describe("databaseDriverManifest", () => {
});
});
it("exposes transfer compatibility evidence from the shared manifest", () => {
expect(databaseTransferSupportLevel("mysql")).toBe("verified");
expect(databaseTransferSupportLevel("kingbase")).toBe("family-compatible");
expect(databaseTransferSupportLevel("db2")).toBe("best-effort");
expect(databaseTransferSupportLevel("gbase")).toBeUndefined();
});
it("exposes schema and data comparison support from the shared manifest", () => {
expect(supportsDatabaseCompareMode("mysql", "schema")).toBe(true);
expect(supportsDatabaseCompareMode("kingbase", "data")).toBe(true);
expect(supportsDatabaseCompareMode("mongodb", "data")).toBe(false);
expect(supportsDatabaseCompareMode("redis", "schema")).toBe(false);
});
it("offers only the supported NebulaGraph 3.x profile", () => {
expect(databaseManifestEntry("nebula")?.driverProfiles).toEqual([{ profile: "nebula-v3", agentKey: "nebula", label: "NebulaGraph 3.x", storeVisible: false }]);
});
+3 -1
View File
@@ -95,8 +95,9 @@ export interface DataCompareFromTablesPreparation extends DataComparePreparation
targetRowCount: number;
sourceTruncated: boolean;
targetTruncated: boolean;
/** Rows per side this compare was allowed to read; fewer than `sourceRowCount`/`targetRowCount` when truncated. */
/** Rows per side this compare was allowed to inspect; fewer than the full table when truncated or sampled. */
rowBudget: number;
syncEligible: boolean;
}
export interface DataCompareSyncPlanTableOptions {
@@ -108,6 +109,7 @@ export interface DataCompareSyncPlanTableOptions {
diff: DataCompareResult;
databaseType?: DatabaseType;
preSyncStatements?: string[];
syncEligible: boolean;
}
export interface DataCompareSyncPlanOptions {
@@ -0,0 +1,16 @@
import { describe, expect, it } from "vitest";
import { supportsDatabaseCompare } from "@/lib/database/databaseCompareCapabilities";
describe("databaseCompareCapabilities", () => {
it("reads comparison support from the shared driver manifest", () => {
expect(supportsDatabaseCompare({ db_type: "mysql" }, "schema")).toBe(true);
expect(supportsDatabaseCompare({ db_type: "kingbase" }, "data")).toBe(true);
expect(supportsDatabaseCompare({ db_type: "mongodb" }, "data")).toBe(false);
expect(supportsDatabaseCompare({ db_type: "redis" }, "schema")).toBe(false);
});
it("applies manifest support after resolving a JDBC product dialect", () => {
expect(supportsDatabaseCompare({ db_type: "jdbc", connection_string: "jdbc:postgresql://localhost:5432/app" }, "data")).toBe(true);
expect(supportsDatabaseCompare({ db_type: "jdbc", driver_profile: "unknown-product" }, "data")).toBe(false);
});
});
@@ -1,84 +1,14 @@
import type { ConnectionConfig, DatabaseType } from "@/types/database";
import type { ConnectionConfig } from "@/types/database";
import { CONNECTION_PROFILES, type ConnectionProfileDefinition } from "@/types/generated/connectionProfiles";
import { databaseManifestEntry } from "@/lib/database/databaseDriverManifest";
import { databaseManifestEntry, supportsDatabaseCompareMode, type DatabaseCompareMode } from "@/lib/database/databaseDriverManifest";
import { effectiveDatabaseTypeForConnection } from "@/lib/database/jdbcDialect";
export type DatabaseCompareMode = "schema" | "data";
// Comparison loads relational metadata and, for data, generates COUNT/SELECT
// statements. Unknown providers must opt in here rather than pass a blacklist.
const SQL_COMPARE_TYPES = new Set<DatabaseType>([
"mysql",
"postgres",
"sqlite",
"rqlite",
"turso",
"cloudflare-d1",
"duckdb",
"clickhouse",
"sqlserver",
"oracle",
"doris",
"starrocks",
"manticoresearch",
"redshift",
"dameng",
"kingbase",
"highgo",
"uxdb",
"vastbase",
"goldendb",
"databend",
"gaussdb",
"kwdb",
"yashandb",
"databricks",
"saphana",
"teradata",
"vertica",
"firebird",
"exasol",
"opengauss",
"questdb",
"oceanbase-oracle",
"gbase",
"access",
"h2",
"snowflake",
"trino",
"prestosql",
"hive",
"argo",
"transwarp",
"kyuubi",
"impala",
"db2",
"informix",
"bigquery",
"spanner",
"kylin",
"ignite",
"ignite3",
"sundb",
"oscar",
"tdengine",
"xugu",
"iotdb",
"iris",
"influxdb3",
"spark",
]);
export type { DatabaseCompareMode } from "@/lib/database/databaseDriverManifest";
export function supportsDatabaseCompare(connection: ConnectionConfig | undefined, mode: DatabaseCompareMode): boolean {
if (!connection) return false;
const type = effectiveDatabaseTypeForConnection(connection);
// Keep the two paths explicit: a future driver may support metadata compare
// without supporting the generated data comparison SQL (or vice versa).
switch (mode) {
case "schema":
case "data":
return !!type && SQL_COMPARE_TYPES.has(type);
}
return supportsDatabaseCompareMode(type, mode);
}
const profiles: Readonly<Record<string, ConnectionProfileDefinition>> = CONNECTION_PROFILES;
@@ -3,6 +3,9 @@ import driverManifest from "../../../../../crates/dbx-core/assets/database-drive
export type DatabaseSupportLevel = "connect" | "browse" | "understand" | "operate";
export type DatabaseRuntimeMode = "native" | "file" | "agent" | "external";
export type DatabaseTransferObjectFamily = "mysql" | "postgres" | "oracle" | "sqlserver";
export type DatabaseTransferSupportLevel = "verified" | "family-compatible" | "best-effort";
export type DatabaseCompareMode = "schema" | "data";
export type ConnectionFormKind = "standard" | "jdbc" | "mq" | "mqtt" | "nacos";
export const DATABASE_PRODUCT_CAPABILITY_KEYS = [
@@ -50,6 +53,9 @@ export interface ManagedDatabaseDriver {
export interface DatabaseDriverManifestEntry {
dbType: DatabaseType;
compareModes?: DatabaseCompareMode[];
transferObjectFamily?: DatabaseTransferObjectFamily;
transferSupportLevel?: DatabaseTransferSupportLevel;
label: string;
dialect?: string;
runtimeMode: DatabaseRuntimeMode;
@@ -140,6 +146,18 @@ export function databaseRuntimeMode(dbType?: DatabaseType): DatabaseRuntimeMode
return dbType ? DATABASE_DRIVER_BY_TYPE.get(dbType)?.runtimeMode : undefined;
}
export function databaseTransferObjectFamily(dbType?: DatabaseType): DatabaseTransferObjectFamily | undefined {
return dbType ? DATABASE_DRIVER_BY_TYPE.get(dbType)?.transferObjectFamily : undefined;
}
export function databaseTransferSupportLevel(dbType?: DatabaseType): DatabaseTransferSupportLevel | undefined {
return dbType ? DATABASE_DRIVER_BY_TYPE.get(dbType)?.transferSupportLevel : undefined;
}
export function supportsDatabaseCompareMode(dbType: DatabaseType | undefined, mode: DatabaseCompareMode): boolean {
return dbType ? DATABASE_DRIVER_BY_TYPE.get(dbType)?.compareModes?.includes(mode) === true : false;
}
export function databaseProductCapabilities(dbType?: DatabaseType): DatabaseProductCapabilities {
if (!dbType) return DEFAULT_CAPABILITIES;
return (DATABASE_DRIVER_BY_TYPE.get(dbType)?.capabilities ?? DEFAULT_CAPABILITIES) as DatabaseProductCapabilities;
@@ -1,5 +1,6 @@
import type { DatabaseType } from "@/types/database";
import { supportsTransfer } from "@/lib/database/databaseFeatureSupport";
import { databaseTransferObjectFamily } from "@/lib/database/databaseDriverManifest";
export enum TransferObjectFamily {
Mysql = "mysql",
@@ -15,22 +16,8 @@ const POSTGRES_KINDS: TransferObjectKind[] = ["TABLE", "VIEW", "MATERIALIZED_VIE
const ORACLE_KINDS: TransferObjectKind[] = ["TABLE", "VIEW", "MATERIALIZED_VIEW", "PROCEDURE", "FUNCTION", "TRIGGER", "SEQUENCE"];
const SQLSERVER_KINDS: TransferObjectKind[] = ["TABLE", "VIEW", "PROCEDURE", "FUNCTION", "TRIGGER", "SEQUENCE"];
const FAMILY_BY_DB = new Map<DatabaseType, TransferObjectFamily>([
["mysql", TransferObjectFamily.Mysql],
["gbase", TransferObjectFamily.Mysql],
["postgres", TransferObjectFamily.Postgres],
["kingbase", TransferObjectFamily.Postgres],
["gaussdb", TransferObjectFamily.Postgres],
["kwdb", TransferObjectFamily.Postgres],
["opengauss", TransferObjectFamily.Postgres],
["oracle", TransferObjectFamily.Oracle],
["dameng", TransferObjectFamily.Oracle],
["oceanbase-oracle", TransferObjectFamily.Oracle],
["sqlserver", TransferObjectFamily.SqlServer],
]);
export function transferObjectFamily(dbType?: DatabaseType): TransferObjectFamily | undefined {
return dbType ? FAMILY_BY_DB.get(dbType) : undefined;
return databaseTransferObjectFamily(dbType) as TransferObjectFamily | undefined;
}
export function isSameTransferFamily(a?: DatabaseType, b?: DatabaseType): boolean {
@@ -3,6 +3,12 @@
"drivers": [
{
"dbType": "mysql",
"compareModes": [
"schema",
"data"
],
"transferSupportLevel": "verified",
"transferObjectFamily": "mysql",
"label": "MySQL",
"dialect": "MySQL",
"runtimeMode": "native",
@@ -36,6 +42,12 @@
},
{
"dbType": "postgres",
"compareModes": [
"schema",
"data"
],
"transferSupportLevel": "verified",
"transferObjectFamily": "postgres",
"label": "PostgreSQL",
"dialect": "PostgreSQL",
"runtimeMode": "native",
@@ -73,6 +85,11 @@
},
{
"dbType": "sqlite",
"compareModes": [
"schema",
"data"
],
"transferSupportLevel": "verified",
"label": "SQLite",
"dialect": "SQLite",
"runtimeMode": "file",
@@ -114,6 +131,11 @@
},
{
"dbType": "rqlite",
"compareModes": [
"schema",
"data"
],
"transferSupportLevel": "best-effort",
"label": "RQLite",
"dialect": "Rqlite",
"runtimeMode": "native",
@@ -147,6 +169,11 @@
},
{
"dbType": "turso",
"compareModes": [
"schema",
"data"
],
"transferSupportLevel": "best-effort",
"label": "Turso",
"dialect": "Turso",
"runtimeMode": "native",
@@ -180,6 +207,11 @@
},
{
"dbType": "cloudflare-d1",
"compareModes": [
"schema",
"data"
],
"transferSupportLevel": "best-effort",
"label": "Cloudflare D1",
"dialect": "SQLite",
"runtimeMode": "native",
@@ -242,6 +274,11 @@
},
{
"dbType": "duckdb",
"compareModes": [
"schema",
"data"
],
"transferSupportLevel": "best-effort",
"label": "DuckDB",
"dialect": "DuckDB",
"runtimeMode": "file",
@@ -284,6 +321,11 @@
},
{
"dbType": "clickhouse",
"compareModes": [
"schema",
"data"
],
"transferSupportLevel": "best-effort",
"label": "ClickHouse",
"dialect": "ClickHouse",
"runtimeMode": "native",
@@ -314,6 +356,12 @@
},
{
"dbType": "sqlserver",
"compareModes": [
"schema",
"data"
],
"transferSupportLevel": "verified",
"transferObjectFamily": "sqlserver",
"label": "SQL Server",
"dialect": "SQL Server",
"runtimeMode": "native",
@@ -359,6 +407,7 @@
},
{
"dbType": "mongodb",
"transferSupportLevel": "best-effort",
"label": "MongoDB (Legacy)",
"runtimeMode": "agent",
"mcpMode": "bridge",
@@ -428,6 +477,12 @@
},
{
"dbType": "oracle",
"compareModes": [
"schema",
"data"
],
"transferSupportLevel": "verified",
"transferObjectFamily": "oracle",
"label": "Oracle",
"dialect": "Oracle",
"runtimeMode": "agent",
@@ -779,6 +834,10 @@
},
{
"dbType": "doris",
"compareModes": [
"schema",
"data"
],
"label": "Doris",
"dialect": "Doris",
"runtimeMode": "native",
@@ -809,6 +868,10 @@
},
{
"dbType": "starrocks",
"compareModes": [
"schema",
"data"
],
"label": "StarRocks",
"dialect": "StarRocks",
"runtimeMode": "native",
@@ -839,6 +902,10 @@
},
{
"dbType": "manticoresearch",
"compareModes": [
"schema",
"data"
],
"label": "Manticore Search",
"dialect": "ManticoreSearch",
"runtimeMode": "native",
@@ -869,6 +936,10 @@
},
{
"dbType": "redshift",
"compareModes": [
"schema",
"data"
],
"label": "Redshift",
"dialect": "Redshift",
"runtimeMode": "native",
@@ -905,6 +976,12 @@
},
{
"dbType": "dameng",
"compareModes": [
"schema",
"data"
],
"transferSupportLevel": "family-compatible",
"transferObjectFamily": "oracle",
"label": "达梦 DM8",
"dialect": "Dameng",
"runtimeMode": "agent",
@@ -945,6 +1022,12 @@
},
{
"dbType": "kingbase",
"compareModes": [
"schema",
"data"
],
"transferSupportLevel": "family-compatible",
"transferObjectFamily": "postgres",
"label": "金仓KingbaseES",
"dialect": "Kingbase",
"runtimeMode": "agent",
@@ -983,6 +1066,11 @@
},
{
"dbType": "highgo",
"compareModes": [
"schema",
"data"
],
"transferSupportLevel": "best-effort",
"label": "瀚高 HighGo",
"dialect": "HighGo",
"runtimeMode": "agent",
@@ -1021,6 +1109,10 @@
},
{
"dbType": "uxdb",
"compareModes": [
"schema",
"data"
],
"label": "优炫 UXDB",
"dialect": "PostgreSQL",
"runtimeMode": "agent",
@@ -1059,6 +1151,11 @@
},
{
"dbType": "vastbase",
"compareModes": [
"schema",
"data"
],
"transferSupportLevel": "best-effort",
"label": "海量 Vastbase",
"dialect": "Vastbase",
"runtimeMode": "agent",
@@ -1097,6 +1194,11 @@
},
{
"dbType": "goldendb",
"compareModes": [
"schema",
"data"
],
"transferSupportLevel": "best-effort",
"label": "金篆 GoldenDB",
"dialect": "GoldenDB",
"runtimeMode": "agent",
@@ -1130,6 +1232,10 @@
},
{
"dbType": "databend",
"compareModes": [
"schema",
"data"
],
"label": "Databend",
"dialect": "Databend",
"runtimeMode": "agent",
@@ -1166,6 +1272,12 @@
},
{
"dbType": "gaussdb",
"compareModes": [
"schema",
"data"
],
"transferSupportLevel": "family-compatible",
"transferObjectFamily": "postgres",
"label": "GaussDB",
"dialect": "GaussDB",
"runtimeMode": "native",
@@ -1204,6 +1316,12 @@
},
{
"dbType": "kwdb",
"compareModes": [
"schema",
"data"
],
"transferSupportLevel": "family-compatible",
"transferObjectFamily": "postgres",
"label": "KWDB",
"dialect": "KwDB",
"runtimeMode": "native",
@@ -1241,6 +1359,11 @@
},
{
"dbType": "yashandb",
"compareModes": [
"schema",
"data"
],
"transferSupportLevel": "best-effort",
"label": "崖山 YashanDB",
"dialect": "YashanDB",
"runtimeMode": "agent",
@@ -1278,6 +1401,10 @@
},
{
"dbType": "databricks",
"compareModes": [
"schema",
"data"
],
"label": "Databricks SQL",
"runtimeMode": "agent",
"mcpMode": "bridge",
@@ -1314,6 +1441,10 @@
},
{
"dbType": "saphana",
"compareModes": [
"schema",
"data"
],
"label": "SAP HANA",
"runtimeMode": "agent",
"mcpMode": "bridge",
@@ -1350,6 +1481,10 @@
},
{
"dbType": "teradata",
"compareModes": [
"schema",
"data"
],
"label": "Teradata",
"runtimeMode": "agent",
"mcpMode": "bridge",
@@ -1386,6 +1521,10 @@
},
{
"dbType": "vertica",
"compareModes": [
"schema",
"data"
],
"label": "Vertica",
"dialect": "Vertica",
"runtimeMode": "agent",
@@ -1423,6 +1562,10 @@
},
{
"dbType": "firebird",
"compareModes": [
"schema",
"data"
],
"label": "Firebird",
"dialect": "Firebird",
"runtimeMode": "agent",
@@ -1459,6 +1602,10 @@
},
{
"dbType": "exasol",
"compareModes": [
"schema",
"data"
],
"label": "Exasol",
"dialect": "Exasol",
"runtimeMode": "agent",
@@ -1496,6 +1643,12 @@
},
{
"dbType": "opengauss",
"compareModes": [
"schema",
"data"
],
"transferSupportLevel": "family-compatible",
"transferObjectFamily": "postgres",
"label": "openGauss",
"dialect": "openGauss",
"runtimeMode": "native",
@@ -1533,6 +1686,11 @@
},
{
"dbType": "questdb",
"compareModes": [
"schema",
"data"
],
"transferSupportLevel": "best-effort",
"label": "QuestDB",
"dialect": "QuestDB",
"runtimeMode": "native",
@@ -1568,6 +1726,12 @@
},
{
"dbType": "oceanbase-oracle",
"compareModes": [
"schema",
"data"
],
"transferSupportLevel": "family-compatible",
"transferObjectFamily": "oracle",
"label": "OceanBase Oracle Mode",
"dialect": "OceanBase",
"runtimeMode": "agent",
@@ -1608,6 +1772,11 @@
},
{
"dbType": "gbase",
"compareModes": [
"schema",
"data"
],
"transferObjectFamily": "mysql",
"label": "南大通用 GBase 8a",
"dialect": "GBase",
"runtimeMode": "agent",
@@ -1661,6 +1830,10 @@
},
{
"dbType": "access",
"compareModes": [
"schema",
"data"
],
"label": "Microsoft Access",
"dialect": "Access",
"runtimeMode": "agent",
@@ -1697,6 +1870,11 @@
},
{
"dbType": "h2",
"compareModes": [
"schema",
"data"
],
"transferSupportLevel": "best-effort",
"label": "H2",
"dialect": "H2",
"runtimeMode": "agent",
@@ -1768,6 +1946,10 @@
},
{
"dbType": "snowflake",
"compareModes": [
"schema",
"data"
],
"label": "Snowflake",
"runtimeMode": "agent",
"mcpMode": "bridge",
@@ -1805,6 +1987,10 @@
},
{
"dbType": "trino",
"compareModes": [
"schema",
"data"
],
"label": "Trino",
"runtimeMode": "agent",
"mcpMode": "bridge",
@@ -1842,6 +2028,10 @@
},
{
"dbType": "prestosql",
"compareModes": [
"schema",
"data"
],
"label": "PrestoSQL",
"runtimeMode": "external",
"mcpMode": "bridge",
@@ -1876,6 +2066,11 @@
},
{
"dbType": "hive",
"compareModes": [
"schema",
"data"
],
"transferSupportLevel": "best-effort",
"label": "Apache Hive",
"runtimeMode": "agent",
"mcpMode": "bridge",
@@ -1911,6 +2106,11 @@
},
{
"dbType": "argo",
"compareModes": [
"schema",
"data"
],
"transferSupportLevel": "best-effort",
"label": "星环Argo",
"runtimeMode": "agent",
"mcpMode": "bridge",
@@ -1946,6 +2146,11 @@
},
{
"dbType": "transwarp",
"compareModes": [
"schema",
"data"
],
"transferSupportLevel": "best-effort",
"label": "星环Inceptor",
"dialect": "Transwarp",
"runtimeMode": "agent",
@@ -1990,6 +2195,11 @@
},
{
"dbType": "kyuubi",
"compareModes": [
"schema",
"data"
],
"transferSupportLevel": "best-effort",
"label": "Apache Kyuubi",
"runtimeMode": "agent",
"mcpMode": "bridge",
@@ -2024,6 +2234,11 @@
},
{
"dbType": "impala",
"compareModes": [
"schema",
"data"
],
"transferSupportLevel": "best-effort",
"label": "Apache Impala",
"runtimeMode": "agent",
"mcpMode": "bridge",
@@ -2058,6 +2273,11 @@
},
{
"dbType": "db2",
"compareModes": [
"schema",
"data"
],
"transferSupportLevel": "best-effort",
"label": "IBM DB2",
"runtimeMode": "agent",
"mcpMode": "bridge",
@@ -2094,6 +2314,10 @@
},
{
"dbType": "informix",
"compareModes": [
"schema",
"data"
],
"label": "IBM Informix",
"dialect": "Informix",
"runtimeMode": "agent",
@@ -2235,6 +2459,10 @@
},
{
"dbType": "bigquery",
"compareModes": [
"schema",
"data"
],
"label": "Google BigQuery",
"runtimeMode": "agent",
"mcpMode": "bridge",
@@ -2266,6 +2494,10 @@
},
{
"dbType": "spanner",
"compareModes": [
"schema",
"data"
],
"label": "Google Cloud Spanner",
"runtimeMode": "agent",
"mcpMode": "bridge",
@@ -2302,6 +2534,10 @@
},
{
"dbType": "kylin",
"compareModes": [
"schema",
"data"
],
"label": "Apache Kylin",
"runtimeMode": "agent",
"mcpMode": "bridge",
@@ -2334,6 +2570,10 @@
},
{
"dbType": "ignite",
"compareModes": [
"schema",
"data"
],
"label": "Apache Ignite",
"runtimeMode": "agent",
"mcpMode": "bridge",
@@ -2370,6 +2610,10 @@
},
{
"dbType": "ignite3",
"compareModes": [
"schema",
"data"
],
"label": "Apache Ignite 3",
"runtimeMode": "agent",
"mcpMode": "bridge",
@@ -2406,6 +2650,10 @@
},
{
"dbType": "sundb",
"compareModes": [
"schema",
"data"
],
"label": "科蓝 SUNDB",
"dialect": "SunDB",
"runtimeMode": "agent",
@@ -2443,6 +2691,10 @@
},
{
"dbType": "oscar",
"compareModes": [
"schema",
"data"
],
"label": "神通 OSCAR",
"runtimeMode": "agent",
"mcpMode": "bridge",
@@ -2479,6 +2731,10 @@
},
{
"dbType": "tdengine",
"compareModes": [
"schema",
"data"
],
"label": "TDengine",
"runtimeMode": "agent",
"mcpMode": "bridge",
@@ -2511,6 +2767,11 @@
},
{
"dbType": "xugu",
"compareModes": [
"schema",
"data"
],
"transferSupportLevel": "best-effort",
"label": "虚谷 XuguDB",
"dialect": "Xugu",
"runtimeMode": "agent",
@@ -2766,6 +3027,10 @@
},
{
"dbType": "iotdb",
"compareModes": [
"schema",
"data"
],
"label": "Apache IoTDB",
"runtimeMode": "agent",
"mcpMode": "bridge",
@@ -2801,6 +3066,11 @@
},
{
"dbType": "iris",
"compareModes": [
"schema",
"data"
],
"transferSupportLevel": "best-effort",
"label": "InterSystems IRIS",
"dialect": "Iris",
"runtimeMode": "agent",
@@ -2878,6 +3148,10 @@
},
{
"dbType": "influxdb3",
"compareModes": [
"schema",
"data"
],
"label": "InfluxDB 3.x",
"runtimeMode": "native",
"mcpMode": "bridge",
@@ -2973,6 +3247,11 @@
},
{
"dbType": "spark",
"compareModes": [
"schema",
"data"
],
"transferSupportLevel": "best-effort",
"label": "Apache Spark",
"runtimeMode": "agent",
"mcpMode": "bridge",
+1
View File
@@ -513,6 +513,7 @@ mod tests {
source_truncated: false,
target_truncated: false,
row_budget: 100_000,
sync_eligible: true,
degradation_level: Some("full".to_string()),
sampling_rate: Some(1.0),
confidence_score: Some(1.0),
+107 -13
View File
@@ -11,11 +11,13 @@ use crate::models::connection::DatabaseType;
use crate::query::{execute_sql_statement_with_options, QueryExecutionOptions};
use crate::schema::get_columns_core;
use crate::sql_dialect::{
build_count_table_sql, firebird_rows_clause, pagination_strategy, qualified_table_name, quote_table_identifier,
PaginationContext, TablePaginationStrategy,
build_count_table_sql, build_iris_table_select_sql, firebird_rows_clause, pagination_strategy,
qualified_table_name, quote_table_identifier, PaginationContext, TablePaginationStrategy,
};
use crate::transfer::{generate_comment_ddl, generate_create_table_ddl};
use super::sample_pairing::retain_shared_sample_rows_by_key;
const DATA_SYNC_INSERT_BATCH_SIZE: usize = 500;
const DATA_SYNC_CONDITION_BATCH_SIZE: usize = 200;
@@ -149,6 +151,8 @@ pub struct DataCompareSyncPlanTableOptions {
pub database_type: Option<DatabaseType>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub pre_sync_statements: Vec<String>,
#[serde(default)]
pub sync_eligible: bool,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
@@ -182,11 +186,13 @@ pub struct DataCompareFromTablesPreparation {
pub source_truncated: bool,
pub target_truncated: bool,
/// Rows per side this compare was allowed to read. When a side holds more rows than
/// this, the diff only covers the first `row_budget` rows and the matching
/// `*_truncated` flag is set, so the UI can warn instead of showing a partial plan
/// as if it were complete.
/// this, the diff covers only a bounded subset and the matching `*_truncated`
/// flag is set. `sync_eligible` remains false so a partial result cannot become
/// executable synchronization SQL.
#[serde(default)]
pub row_budget: usize,
#[serde(default)]
pub sync_eligible: bool,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub degradation_level: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
@@ -331,6 +337,8 @@ pub async fn prepare_data_compare_from_tables(
) -> Result<DataCompareFromTablesPreparation, String> {
let source_database_type = connection_database_type(state, &options.source_connection_id).await?;
let target_database_type = connection_database_type(state, &options.target_connection_id).await?;
ensure_data_compare_supported(source_database_type, "source")?;
ensure_data_compare_supported(target_database_type, "target")?;
let fetch_batch_size = options.fetch_batch_size.unwrap_or(1000).max(1);
let source_count_sql =
@@ -435,14 +443,24 @@ pub async fn prepare_data_compare_from_tables(
threshold.sample_size,
)
)?;
let column_indexes = column_index_map(&options.columns);
let (src, tgt) = retain_shared_sample_rows_by_key(src, tgt, |row| {
key_for_row(row, &options.key_columns, &column_indexes)
});
let max_count = source_row_count.max(target_row_count);
let sample_rate = if max_count > 0 { threshold.sample_size as f64 / max_count as f64 } else { 1.0 };
let method = "sampled".to_string();
(src, tgt, sample_rate.min(1.0), method, false, false)
let method = "paired_key_sample".to_string();
let source_truncated = source_row_count > src.len() as u64;
let target_truncated = target_row_count > tgt.len() as u64;
(src, tgt, sample_rate.min(1.0), method, source_truncated, target_truncated)
}
DegradationLevel::SkipWithRisk => {
(Vec::new(), Vec::new(), 0.0, "skipped".to_string(), source_row_count > 0, target_row_count > 0)
}
DegradationLevel::SkipWithRisk => (Vec::new(), Vec::new(), 0.0, "skipped".to_string(), false, false),
};
let sync_eligible = matches!(degradation_level, DegradationLevel::Full) && !source_truncated && !target_truncated;
let target_columns = get_columns_core(
state,
&options.target_connection_id,
@@ -478,16 +496,20 @@ pub async fn prepare_data_compare_from_tables(
database_type: Some(target_database_type),
})?;
let (sync_statements, sync_sql) =
if sync_eligible { (preparation.sync_statements, preparation.sync_sql) } else { (Vec::new(), String::new()) };
Ok(DataCompareFromTablesPreparation {
result: preparation.result,
sync_statements: preparation.sync_statements,
sync_sql: preparation.sync_sql,
sync_statements,
sync_sql,
pre_sync_statements: Vec::new(),
source_row_count,
target_row_count,
source_truncated,
target_truncated,
row_budget,
sync_eligible,
degradation_level: Some(degradation_level.to_string()),
sampling_rate: Some(sampling_rate),
confidence_score: Some(confidence_score),
@@ -503,6 +525,8 @@ pub async fn prepare_data_compare_missing_target(
) -> Result<DataCompareFromTablesPreparation, String> {
let source_database_type = connection_database_type(state, &options.source_connection_id).await?;
let target_database_type = connection_database_type(state, &options.target_connection_id).await?;
ensure_data_compare_supported(source_database_type, "source")?;
ensure_data_compare_supported(target_database_type, "target")?;
let fetch_batch_size = options.fetch_batch_size.unwrap_or(1000).max(1);
let source_columns = get_columns_core(
state,
@@ -546,6 +570,7 @@ pub async fn prepare_data_compare_missing_target(
)
.await?;
let source_truncated = source_row_count > row_budget as u64;
let sync_eligible = !source_truncated;
let result = missing_target_diff(&column_names, &options.key_columns, source_rows);
let mut pre_sync_statements = Vec::new();
pre_sync_statements.push(format!(
@@ -587,14 +612,15 @@ pub async fn prepare_data_compare_missing_target(
Ok(DataCompareFromTablesPreparation {
result,
sync_statements: sync_plan.sync_statements,
sync_sql: sync_plan.sync_sql,
sync_statements: if sync_eligible { sync_plan.sync_statements } else { Vec::new() },
sync_sql: if sync_eligible { sync_plan.sync_sql } else { String::new() },
pre_sync_statements,
source_row_count,
target_row_count: 0,
source_truncated,
target_truncated: false,
row_budget,
sync_eligible,
degradation_level: Some(should_degrade(source_row_count, 0, &threshold).to_string()),
sampling_rate: Some(if source_truncated { row_budget as f64 / source_row_count.max(1) as f64 } else { 1.0 }),
confidence_score: Some(if source_truncated { 0.5 } else { 1.0 }),
@@ -610,6 +636,7 @@ pub fn build_data_compare_sync_plan(options: DataCompareSyncPlanOptions) -> Data
let tables = options
.tables
.iter()
.filter(|table| table.sync_eligible)
.map(|table| DataCompareSyncPlanTableRef {
table_name: &table.table_name,
schema: table.schema.as_deref(),
@@ -1069,6 +1096,16 @@ async fn connection_database_type(state: &AppState, connection_id: &str) -> Resu
.ok_or_else(|| format!("Connection config not found: {connection_id}"))
}
fn ensure_data_compare_supported(database_type: DatabaseType, endpoint: &str) -> Result<(), String> {
if dbx_types::database_manifest::supports_compare_mode(
&database_type,
dbx_types::database_manifest::DatabaseCompareMode::Data,
) {
return Ok(());
}
Err(format!("{endpoint} database {} does not support data comparison", database_type.as_str()))
}
fn first_count(rows: &[Vec<Value>]) -> Result<u64, String> {
let value = rows.first().and_then(|row| row.first()).ok_or_else(|| "COUNT query returned no rows".to_string())?;
match value {
@@ -1164,7 +1201,9 @@ fn build_data_compare_select_sql(
"WITH {page_alias} AS (SELECT {select_columns}, ROW_NUMBER() OVER (ORDER BY {order_expression}) AS {row_number_alias} FROM {table}) SELECT {select_columns} FROM {page_alias} WHERE {row_number_alias} > {offset} AND {row_number_alias} <= {end} ORDER BY {row_number_alias}"
)
}
TablePaginationStrategy::IrisTop => format!("SELECT TOP {row_limit} {select_columns} FROM {table}{order_by}"),
TablePaginationStrategy::IrisTop => {
build_iris_table_select_sql(&select_columns, &table, "", &order_by, row_limit, offset as u64)
}
TablePaginationStrategy::InformixFirst => {
let row_limit_clause =
if offset > 0 { format!("SKIP {offset} FIRST {row_limit}") } else { format!("FIRST {row_limit}") };
@@ -1991,6 +2030,7 @@ mod tests {
},
database_type: Some(DatabaseType::Postgres),
pre_sync_statements: Vec::new(),
sync_eligible: true,
}],
});
@@ -2038,6 +2078,7 @@ mod tests {
},
database_type: Some(DatabaseType::Postgres),
pre_sync_statements: Vec::new(),
sync_eligible: true,
}],
});
@@ -2090,6 +2131,7 @@ mod tests {
diff: DataCompareResult { added: Vec::new(), removed: Vec::new(), modified },
database_type: Some(DatabaseType::Postgres),
pre_sync_statements: Vec::new(),
sync_eligible: true,
}],
});
@@ -2174,6 +2216,7 @@ mod tests {
},
database_type: Some(DatabaseType::Postgres),
pre_sync_statements: Vec::new(),
sync_eligible: true,
}],
});
@@ -2183,6 +2226,35 @@ mod tests {
assert_eq!(plan.statement_count, 2);
}
#[test]
fn incomplete_comparisons_cannot_build_sync_statements() {
let plan = build_data_compare_sync_plan(DataCompareSyncPlanOptions {
tables: vec![DataCompareSyncPlanTableOptions {
table_name: "users".to_string(),
schema: Some("public".to_string()),
columns: vec!["id".to_string()],
key_columns: vec!["id".to_string()],
column_info: Vec::new(),
diff: DataCompareResult {
added: vec![DataCompareRow {
key: "1".to_string(),
key_values: HashMap::from([(String::from("id"), json!(1))]),
values: HashMap::from([(String::from("id"), json!(1))]),
}],
removed: Vec::new(),
modified: Vec::new(),
},
database_type: Some(DatabaseType::Postgres),
pre_sync_statements: Vec::new(),
sync_eligible: false,
}],
});
assert_eq!(plan.statement_count, 0);
assert!(plan.sync_statements.is_empty());
assert!(plan.sync_sql.is_empty());
}
#[test]
fn batches_added_rows_into_multi_value_insert_statements() {
let plan = build_data_compare_sync_plan(DataCompareSyncPlanOptions {
@@ -2216,6 +2288,7 @@ mod tests {
},
database_type: Some(DatabaseType::Postgres),
pre_sync_statements: Vec::new(),
sync_eligible: true,
}],
});
@@ -2270,6 +2343,7 @@ mod tests {
},
database_type: Some(DatabaseType::Mysql),
pre_sync_statements: Vec::new(),
sync_eligible: true,
}],
});
@@ -2308,6 +2382,7 @@ mod tests {
},
database_type: Some(DatabaseType::Postgres),
pre_sync_statements: Vec::new(),
sync_eligible: true,
}],
});
@@ -2356,6 +2431,7 @@ mod tests {
diff: diff.clone(),
database_type: Some(DatabaseType::Postgres),
pre_sync_statements: pre_sync_statements.clone(),
sync_eligible: true,
}],
});
let borrowed_plan = build_data_compare_sync_plan_from_refs(&[DataCompareSyncPlanTableRef {
@@ -2397,6 +2473,7 @@ mod tests {
},
database_type: Some(DatabaseType::Postgres),
pre_sync_statements: vec!["CREATE TABLE \"public\".\"users\" (\"id\" integer);".to_string()],
sync_eligible: true,
}],
});
@@ -2440,6 +2517,7 @@ mod tests {
pre_sync_statements: vec![
"CREATE TABLE \"public\".\"generated_column_sync_test\" (\"id\" bigint);".to_string()
],
sync_eligible: true,
}],
});
@@ -2874,6 +2952,22 @@ mod tests {
);
}
#[test]
fn builds_backend_table_select_sql_for_iris_offset_pages() {
assert_eq!(
build_data_compare_select_sql(
DatabaseType::Iris,
"APP",
"USERS",
&["ID".to_string(), "NAME".to_string()],
&["ID".to_string()],
25,
50,
),
"SELECT * FROM (SELECT TOP 75 ID, NAME FROM APP.USERS ORDER BY ID ASC) WHERE %VID > 50"
);
}
#[test]
fn builds_backend_table_select_sql_for_oceanbase_oracle_rownum_pages() {
assert_eq!(
+2
View File
@@ -10,6 +10,7 @@ pub mod grid_clipboard_guard;
pub mod mongodb_dump;
pub mod mongodb_import_export;
pub mod query_result_export;
mod sample_pairing;
pub mod script_generator;
pub mod sql_file_import;
pub mod sqlite_backup;
@@ -17,3 +18,4 @@ pub mod table_export;
pub mod table_import;
pub mod transfer;
pub mod transfer_rebuild;
pub mod transfer_runner;
@@ -0,0 +1,18 @@
use std::collections::HashSet;
use std::hash::Hash;
pub(super) fn retain_shared_sample_rows_by_key<Row, Key, KeyFn>(
source_rows: Vec<Row>,
target_rows: Vec<Row>,
key_for_row: KeyFn,
) -> (Vec<Row>, Vec<Row>)
where
Key: Eq + Hash,
KeyFn: Fn(&Row) -> Key,
{
let source_keys = source_rows.iter().map(&key_for_row).collect::<HashSet<_>>();
let target_keys = target_rows.iter().map(&key_for_row).collect::<HashSet<_>>();
let source_rows = source_rows.into_iter().filter(|row| target_keys.contains(&key_for_row(row))).collect();
let target_rows = target_rows.into_iter().filter(|row| source_keys.contains(&key_for_row(row))).collect();
(source_rows, target_rows)
}
+20 -19
View File
@@ -224,28 +224,15 @@ pub struct TransferObjectSelection {
pub names: Vec<String>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TransferObjectFamily {
Mysql,
Postgres,
Oracle,
SqlServer,
}
pub use dbx_types::database_manifest::DatabaseTransferObjectFamily as TransferObjectFamily;
pub use dbx_types::database_manifest::DatabaseTransferSupportLevel as TransferSupportLevel;
pub fn transfer_object_family(db_type: &DatabaseType) -> Option<TransferObjectFamily> {
match db_type {
DatabaseType::Mysql | DatabaseType::Gbase => Some(TransferObjectFamily::Mysql),
DatabaseType::Postgres
| DatabaseType::Kingbase
| DatabaseType::Gaussdb
| DatabaseType::Kwdb
| DatabaseType::OpenGauss => Some(TransferObjectFamily::Postgres),
DatabaseType::Oracle | DatabaseType::Dameng | DatabaseType::OceanbaseOracle => {
Some(TransferObjectFamily::Oracle)
}
DatabaseType::SqlServer => Some(TransferObjectFamily::SqlServer),
_ => None,
dbx_types::database_manifest::transfer_object_family(db_type)
}
pub fn transfer_support_level(db_type: &DatabaseType) -> Option<TransferSupportLevel> {
dbx_types::database_manifest::transfer_support_level(db_type)
}
pub fn is_same_transfer_family(a: &DatabaseType, b: &DatabaseType) -> bool {
@@ -1045,6 +1032,12 @@ pub fn validate_transfer_database_pair(
source_db_type: &DatabaseType,
target_db_type: &DatabaseType,
) -> Result<(), String> {
if transfer_support_level(source_db_type).is_none() {
return Err(format!("{} 暂未声明数据传输兼容级别", source_db_type.as_str()));
}
if transfer_support_level(target_db_type).is_none() {
return Err(format!("{} 暂未声明数据传输兼容级别", target_db_type.as_str()));
}
if !matches!(source_db_type, DatabaseType::Xugu) && !matches!(target_db_type, DatabaseType::Xugu) {
return Ok(());
}
@@ -14368,6 +14361,14 @@ CREATE TABLE "Other"."prefix""Source"."NAME" ("ID" INT);"#;
assert!(validate_transfer_database_pair(&rebuild, &DatabaseType::Xugu, &DatabaseType::Xugu).is_err());
}
#[test]
fn transfer_pair_requires_declared_endpoint_compatibility() {
let request = base_request();
let error = validate_transfer_database_pair(&request, &DatabaseType::Gbase, &DatabaseType::Mysql)
.expect_err("GBase data transfer is disabled in the shared manifest");
assert!(error.contains("gbase"));
}
#[test]
fn xugu_structure_transfer_rejects_renamed_tables_and_non_table_objects() {
let request = TransferRequest { target_table_name_case: TransferTableNameCase::Upper, ..base_request() };
+694
View File
@@ -0,0 +1,694 @@
use std::sync::{Arc, Mutex};
use crate::{
connection::AppState,
models::connection::DatabaseType,
persistence::task_history::{TaskItemStatus, TransferTaskJournal},
};
use super::transfer::{TransferProgress, TransferRequest, TransferStatus};
struct ProgressEmitter<F> {
callback: Mutex<F>,
terminal: Mutex<Option<TransferProgress>>,
}
impl<F> ProgressEmitter<F>
where
F: FnMut(TransferProgress) + Send,
{
fn new(callback: F) -> Self {
Self { callback: Mutex::new(callback), terminal: Mutex::new(None) }
}
fn emit(&self, progress: TransferProgress) {
(self.callback.lock().unwrap_or_else(std::sync::PoisonError::into_inner))(progress);
}
fn emit_terminal(&self, progress: TransferProgress) {
*self.terminal.lock().unwrap_or_else(std::sync::PoisonError::into_inner) = Some(progress);
}
fn take_terminal(&self, transfer_id: &str, total_tables: usize) -> TransferProgress {
self.terminal.lock().unwrap_or_else(std::sync::PoisonError::into_inner).take().unwrap_or_else(|| {
TransferProgress {
transfer_id: transfer_id.to_string(),
table: String::new(),
table_index: total_tables,
total_tables,
rows_transferred: 0,
total_rows: None,
status: TransferStatus::Error,
error: Some("Transfer ended without a terminal result".to_string()),
terminal: true,
}
})
}
}
fn emit_progress<F>(emitter: &ProgressEmitter<F>, progress: TransferProgress)
where
F: FnMut(TransferProgress) + Send,
{
emitter.emit(progress);
}
async fn emit_terminal_progress<F>(
emitter: &ProgressEmitter<F>,
_history: Option<&TransferTaskJournal>,
progress: TransferProgress,
) where
F: FnMut(TransferProgress) + Send,
{
emitter.emit_terminal(progress);
}
#[allow(clippy::too_many_arguments)]
pub async fn run_transfer<F>(
state: Arc<AppState>,
request: TransferRequest,
source_db_type: DatabaseType,
target_db_type: DatabaseType,
source_pool_key: String,
target_pool_key: String,
history: Option<TransferTaskJournal>,
progress_callback: F,
) -> TransferProgress
where
F: FnMut(TransferProgress) + Send,
{
let transfer_id = request.transfer_id.clone();
let total_tables = request.tables.len();
let app = ProgressEmitter::new(progress_callback);
async {
// Sort tables by FK dependency so referenced tables are transferred first,
// and keep the foreign key metadata fetched along the way — MySQL-family
// targets reuse it per table below instead of re-querying it.
// 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, known_foreign_keys) = {
let skip_fk_sort = {
let configs = state.configs.read().await;
configs
.get(&request.source_connection_id)
.and_then(|config| {
super::transfer::resolve_external_transfer_catalog_for_config(
request.source_catalog.as_deref(),
config,
)
})
.is_some()
};
if skip_fk_sort {
(request.tables.clone(), std::collections::HashMap::new())
} else {
super::transfer::sort_tables_by_fk_dependency_with_foreign_keys(
&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(), std::collections::HashMap::new())
})
}
};
let total_tables = sorted_tables.len();
log::info!("[transfer] starting transfer_id={} tables={}", transfer_id, total_tables);
let mut failed_tables: Vec<String> = Vec::new();
let mut pending_fk_alters: Vec<(String, String)> = Vec::new();
let mut last_rows_transferred = 0_u64;
let mut last_total_rows = None;
// When drop_target_before_create is true, perform a rename pre-pass in
// parents_first=false order (children first) so that foreign keys on the
// backup tables remain intact. The main create/insert pass then runs in
// parents_first=true order (parents first).
// Children-first order used by the rename pre-pass, kept for the post-loop backup
// cleanup: a backup can still be referenced by another backup, so the drops must
// follow the same order.
let mut backup_drop_order: Vec<String> = Vec::new();
let backup_names = if request.drop_target_before_create {
// Re-sort tables in children-first order for the rename pre-pass
let (tables_for_rename, _) = super::transfer::sort_tables_by_fk_dependency_with_foreign_keys(
&state,
&request.target_connection_id,
&request.target_database,
&request.target_schema,
&sorted_tables,
false, // parents_first=false: children first
)
.await
.unwrap_or_else(|e| {
log::warn!("[transfer] failed to sort tables for rename pre-pass, using original order: {e}");
(sorted_tables.clone(), std::collections::HashMap::new())
});
backup_drop_order = tables_for_rename.clone();
match super::transfer::rename_tables_to_backup(
&state,
&request,
&tables_for_rename,
target_db_type,
&target_pool_key,
|progress| {
last_rows_transferred = progress.rows_transferred;
last_total_rows = progress.total_rows;
emit_progress(&app, progress);
},
)
.await
{
Ok(names) => Some(names),
Err(e) if e == "Cancelled" => {
emit_terminal_progress(
&app,
history.as_ref(),
TransferProgress {
transfer_id: transfer_id.clone(),
table: "rename pre-pass".to_string(),
table_index: 0,
total_tables,
rows_transferred: last_rows_transferred,
total_rows: last_total_rows,
status: TransferStatus::Cancelled,
error: None,
terminal: true,
},
)
.await;
super::transfer::clear_cancelled(&transfer_id).await;
return;
}
Err(e) => {
emit_terminal_progress(
&app,
history.as_ref(),
TransferProgress {
transfer_id: transfer_id.clone(),
table: "rename pre-pass".to_string(),
table_index: 0,
total_tables,
rows_transferred: last_rows_transferred,
total_rows: last_total_rows,
status: TransferStatus::Error,
error: Some(e),
terminal: true,
},
)
.await;
super::transfer::clear_cancelled(&transfer_id).await;
return;
}
}
} else {
None
};
if matches!(source_db_type, DatabaseType::Postgres) && matches!(target_db_type, DatabaseType::Postgres) {
match super::transfer::transfer_postgres_schema_dependencies(
&state,
&request,
&source_pool_key,
&target_pool_key,
|progress| {
last_rows_transferred = progress.rows_transferred;
last_total_rows = progress.total_rows;
emit_progress(&app, progress);
},
)
.await
{
Ok(()) => {}
Err(e) if e == "Cancelled" => {
emit_terminal_progress(
&app,
history.as_ref(),
TransferProgress {
transfer_id: transfer_id.clone(),
table: "schema dependencies".to_string(),
table_index: 0,
total_tables,
rows_transferred: last_rows_transferred,
total_rows: last_total_rows,
status: TransferStatus::Cancelled,
error: None,
terminal: true,
},
)
.await;
super::transfer::clear_cancelled(&transfer_id).await;
return;
}
Err(e) => {
emit_terminal_progress(
&app,
history.as_ref(),
TransferProgress {
transfer_id: transfer_id.clone(),
table: "schema dependencies".to_string(),
table_index: 0,
total_tables,
rows_transferred: last_rows_transferred,
total_rows: last_total_rows,
status: TransferStatus::Error,
error: Some(e),
terminal: true,
},
)
.await;
super::transfer::clear_cancelled(&transfer_id).await;
return;
}
}
}
// Overwrite clears each target right before copying it, parents first, which cannot
// clear a table another selected table references. Empty those children first now.
let overwrite_cleared = match super::transfer::clear_foreign_key_linked_overwrite_targets(
&state,
&request,
&sorted_tables,
target_db_type,
&target_pool_key,
)
.await
{
Ok(cleared) => cleared,
Err(e) => {
emit_terminal_progress(
&app,
history.as_ref(),
TransferProgress {
transfer_id: transfer_id.clone(),
table: "overwrite pre-pass".to_string(),
table_index: 0,
total_tables,
rows_transferred: last_rows_transferred,
total_rows: last_total_rows,
status: TransferStatus::Error,
error: Some(e),
terminal: true,
},
)
.await;
super::transfer::clear_cancelled(&transfer_id).await;
return;
}
};
for (i, table) in sorted_tables.iter().enumerate() {
if super::transfer::is_cancelled(&transfer_id).await {
emit_terminal_progress(
&app,
history.as_ref(),
TransferProgress {
transfer_id: transfer_id.clone(),
table: table.clone(),
table_index: i,
total_tables,
rows_transferred: last_rows_transferred,
total_rows: last_total_rows,
status: TransferStatus::Cancelled,
error: None,
terminal: true,
},
)
.await;
super::transfer::clear_cancelled(&transfer_id).await;
return;
}
log::info!("[transfer] table {}/{}: {}", i + 1, total_tables, table);
let history_item_index =
history.as_ref().and_then(|journal| journal.item_index_for_table(table)).unwrap_or(i);
if let Some(journal) = history.as_ref() {
journal.start_table(history_item_index).await;
}
let mut source_row_count = None;
let mut moved_row_count = None;
let history_for_progress = history.as_ref();
let history_for_count = history.as_ref();
let result = super::transfer::transfer_table_with_result(
&state,
&request,
table,
i,
&source_db_type,
&target_db_type,
&source_pool_key,
&target_pool_key,
&known_foreign_keys,
&mut pending_fk_alters,
backup_names.as_ref(),
overwrite_cleared.contains(table),
|progress| {
last_rows_transferred = progress.rows_transferred;
last_total_rows = progress.total_rows;
moved_row_count = Some(progress.rows_transferred);
if let Some(journal) = history_for_progress {
journal.observe_table_progress(history_item_index, progress.rows_transferred);
}
emit_progress(&app, progress);
},
|source_count| {
source_row_count = source_count;
if let Some(journal) = history_for_count {
journal.observe_source_count(history_item_index, source_count);
}
},
)
.await;
match result {
Ok(result) => {
if let Some(journal) = history.as_ref() {
journal
.finish_table(
history_item_index,
TaskItemStatus::Succeeded,
result.source_row_count,
Some(result.moved_rows),
)
.await;
}
emit_progress(
&app,
TransferProgress {
transfer_id: transfer_id.clone(),
table: table.clone(),
table_index: i,
total_tables,
rows_transferred: result.moved_rows,
total_rows: last_total_rows.or(Some(result.moved_rows)),
status: TransferStatus::TableDone,
error: None,
terminal: false,
},
);
}
Err(e) => {
if e == "Cancelled" {
if let Some(journal) = history.as_ref() {
journal
.finish_table(
history_item_index,
TaskItemStatus::Cancelled,
source_row_count,
moved_row_count,
)
.await;
}
emit_terminal_progress(
&app,
history.as_ref(),
TransferProgress {
transfer_id: transfer_id.clone(),
table: table.clone(),
table_index: i,
total_tables,
rows_transferred: 0,
total_rows: None,
status: TransferStatus::Cancelled,
error: None,
terminal: true,
},
)
.await;
super::transfer::clear_cancelled(&transfer_id).await;
return;
}
if let Some(journal) = history.as_ref() {
journal
.finish_table(history_item_index, TaskItemStatus::Failed, source_row_count, moved_row_count)
.await;
}
failed_tables.push(table.clone());
emit_progress(
&app,
TransferProgress {
transfer_id: transfer_id.clone(),
table: table.clone(),
table_index: i,
total_tables,
rows_transferred: last_rows_transferred,
total_rows: last_total_rows,
status: TransferStatus::Error,
error: Some(e),
terminal: false,
},
);
}
}
}
// Add any foreign keys deferred during MySQL-family table creation now that
// every selected table exists — see transfer_table's use of
// strip_inline_foreign_key_constraint_lines for why these can't be created
// inline (a foreign key cycle has no valid CREATE TABLE order at all). The
// rename pre-pass already freed the backup constraint names, so these re-create
// the original names without colliding.
let mut failed_fk_tables: Vec<String> = Vec::new();
let mut failed_fk_count = 0usize;
for (table, alter_sql) in &pending_fk_alters {
if let Err(e) = super::transfer::execute_on_pool(&state, &target_pool_key, alter_sql).await {
log::warn!("[transfer] failed to add deferred foreign key constraint for {table}: {e}");
failed_fk_count += 1;
failed_fk_tables.push(table.clone());
}
}
if failed_fk_count > 0 {
failed_fk_tables.sort();
failed_fk_tables.dedup();
failed_tables.push(format!("{} foreign key(s) on: {}", failed_fk_count, failed_fk_tables.join(", ")));
}
// Transfer selected non-table objects (views, procedures, functions,
// triggers, sequences, events) after the per-table loop. The shared
// Core decision handles all content modes: DataOnly never
// transfers schema objects; PG→PG keeps the legacy empty-selection
// default only when structure participates in the transfer.
let mut object_outcome = super::transfer::TransferObjectOutcome::default();
match super::transfer::transfer_schema_objects(
&state,
&request,
&source_pool_key,
&target_pool_key,
|progress| emit_progress(&app, progress),
)
.await
{
Ok(outcome) => {
if let Some(journal) = history.as_ref() {
journal.record_object_outcome(&outcome).await;
}
object_outcome = outcome;
}
Err(e) if e == "Cancelled" => {
if let Some(journal) = history.as_ref() {
journal.record_schema_objects_error(true).await;
}
emit_terminal_progress(
&app,
history.as_ref(),
TransferProgress {
transfer_id: transfer_id.clone(),
table: "schema objects".to_string(),
table_index: total_tables,
total_tables,
rows_transferred: 0,
total_rows: None,
status: TransferStatus::Cancelled,
error: None,
terminal: true,
},
)
.await;
super::transfer::clear_cancelled(&transfer_id).await;
return;
}
Err(e) => {
if let Some(journal) = history.as_ref() {
journal.record_schema_objects_error(false).await;
}
failed_tables.push("schema objects".to_string());
emit_progress(
&app,
TransferProgress {
transfer_id: transfer_id.clone(),
table: "schema objects".to_string(),
table_index: total_tables,
total_tables,
rows_transferred: 0,
total_rows: None,
status: TransferStatus::Error,
error: Some(e),
terminal: false,
},
);
}
}
if !object_outcome.failed.is_empty() {
failed_tables.push(format!("schema objects ({})", object_outcome.failed.len()));
}
// The rename pre-pass left one backup per rebuilt table. Drop them only now that
// every table, every deferred foreign key and every selected schema object has
// succeeded — a failure anywhere above keeps the originals recoverable under
// their backup names.
if let Some(backup_names) = backup_names.as_ref() {
if failed_tables.is_empty() {
if let Err(e) = super::transfer::drop_backup_tables(
&state,
&request,
target_db_type,
&target_pool_key,
backup_names,
&backup_drop_order,
)
.await
{
failed_tables.push(e);
}
} else {
log::warn!(
"[transfer] keeping {} backup table(s) because {} step(s) failed",
backup_names.len(),
failed_tables.len()
);
}
}
let skip_suffix = if !object_outcome.skipped.is_empty() && failed_tables.is_empty() {
format!(",跳过 {} 个已存在对象", object_outcome.skipped.len())
} else if !object_outcome.skipped.is_empty() {
format!(";跳过 {} 个已存在对象", object_outcome.skipped.len())
} else {
String::new()
};
emit_terminal_progress(
&app,
history.as_ref(),
TransferProgress {
transfer_id: transfer_id.clone(),
table: String::new(),
table_index: total_tables,
total_tables,
rows_transferred: last_rows_transferred,
total_rows: last_total_rows,
status: if failed_tables.is_empty() { TransferStatus::Done } else { TransferStatus::Error },
error: if failed_tables.is_empty() {
if skip_suffix.is_empty() {
None
} else {
Some(skip_suffix.clone())
}
} else {
Some(format!(
"{} table(s) failed: {}{}",
failed_tables.len(),
failed_tables.iter().take(5).cloned().collect::<Vec<_>>().join(", "),
skip_suffix
))
},
terminal: true,
},
)
.await;
super::transfer::clear_cancelled(&transfer_id).await;
}
.await;
super::transfer::clear_cancelled(&transfer_id).await;
app.take_terminal(&transfer_id, total_tables)
}
#[cfg(test)]
mod tests {
use super::*;
use crate::models::connection::ConnectionConfig;
use serde_json::json;
#[tokio::test]
async fn empty_data_only_transfer_completes_without_schema_object_progress() {
let directory = tempfile::tempdir().unwrap();
let storage = crate::persistence::test_storage::open(&directory.path().join("storage.db")).await.unwrap();
let state = Arc::new(AppState::new_with_plugin_dir(storage, directory.path().join("plugins")));
for id in ["source", "target"] {
let path = directory.path().join(format!("{id}.db"));
crate::db::sqlite::connect_path_create_if_missing(path.to_str().unwrap()).await.unwrap();
let config: ConnectionConfig = serde_json::from_value(json!({
"id": id,
"name": id,
"db_type": "sqlite",
"host": path.to_str().unwrap(),
"port": 0,
"username": "",
"password": "",
"database": null,
"one_time": false,
"save_password": false,
"read_only": false
}))
.unwrap();
state.configs.write().await.insert(id.to_string(), config);
}
let source_pool = super::super::transfer::ensure_transfer_pool(&state, "source", "main", None).await.unwrap();
let target_pool = super::super::transfer::ensure_transfer_pool(&state, "target", "main", None).await.unwrap();
let request: TransferRequest = serde_json::from_value(json!({
"transferId": uuid::Uuid::new_v4().to_string(),
"sourceConnectionId": "source",
"sourceDatabase": "main",
"sourceSchema": "main",
"targetConnectionId": "target",
"targetDatabase": "main",
"targetSchema": "main",
"tables": [],
"createTable": false,
"dropTargetBeforeCreate": false,
"dropTargetConfirmed": false,
"content": "dataOnly",
"objects": [],
"mode": "append",
"batchSize": 1000
}))
.unwrap();
let transfer_id = request.transfer_id.clone();
let observed = Arc::new(Mutex::new(Vec::new()));
let observed_for_callback = observed.clone();
let terminal = run_transfer(
state,
request,
DatabaseType::Sqlite,
DatabaseType::Sqlite,
source_pool,
target_pool,
None,
move |progress| {
observed_for_callback.lock().unwrap_or_else(std::sync::PoisonError::into_inner).push(progress);
},
)
.await;
assert!(matches!(terminal.status, TransferStatus::Done));
assert!(terminal.terminal);
assert!(terminal.error.is_none());
assert!(
observed
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.iter()
.all(|progress| progress.table != "schema objects"),
"data-only no-op must not emit schema-object progress"
);
assert!(!super::super::transfer::is_cancelled(&transfer_id).await);
}
}
+1
View File
@@ -101,6 +101,7 @@ pub use data::table_export;
pub use data::table_import;
pub use data::transfer;
pub use data::transfer_rebuild;
pub use data::transfer_runner;
pub use dbx_drivers::ssh_config;
pub use dbx_formats::sql_file_zip_package;
pub use dbx_formats::temporal_format;
@@ -1,4 +1,9 @@
use dbx_core::data_compare::{DataCompareFromTablesPreparation, DataComparePreparation, DataCompareResult};
use std::collections::HashMap;
use dbx_core::data_compare::{
build_data_compare_sync_plan, DataCompareFromTablesPreparation, DataComparePreparation, DataCompareResult,
DataCompareRow, DataCompareSyncPlanOptions, DataCompareSyncPlanTableOptions,
};
use dbx_core::models::connection::DatabaseType;
use dbx_core::schema_diff::{
generate_schema_sync_sql, prepare_schema_diff, SchemaDiffPreparation, SchemaDiffPreparationOptions, TableDiff,
@@ -242,6 +247,7 @@ fn data_compare_from_tables_preparation_optional_fields_default() {
source_truncated: false,
target_truncated: false,
row_budget: 100_000,
sync_eligible: true,
degradation_level: None,
sampling_rate: None,
confidence_score: None,
@@ -257,6 +263,7 @@ fn data_compare_from_tables_preparation_optional_fields_default() {
assert!(json.get("syncSql").is_some(), "syncSql must be present");
// The row budget travels with the preparation so the UI can flag a partial compare.
assert_eq!(json.get("rowBudget").and_then(|value| value.as_u64()), Some(100_000));
assert_eq!(json.get("syncEligible").and_then(|value| value.as_bool()), Some(true));
// New optional fields should be absent when None
assert!(json.get("degradationLevel").is_none(), "degradationLevel should be absent");
@@ -286,11 +293,41 @@ fn data_compare_from_tables_preparation_deserializes_old_format() {
assert_eq!(prep.source_row_count, 100);
assert_eq!(prep.target_row_count, 100);
assert!(!prep.sync_eligible, "legacy payloads without completeness evidence must fail closed");
assert!(prep.degradation_level.is_none());
assert!(prep.sampling_rate.is_none());
assert!(prep.confidence_score.is_none());
}
#[test]
fn incomplete_data_compare_cannot_generate_sync_sql() {
let plan = build_data_compare_sync_plan(DataCompareSyncPlanOptions {
tables: vec![DataCompareSyncPlanTableOptions {
table_name: "users".to_string(),
schema: Some("public".to_string()),
columns: vec!["id".to_string()],
key_columns: vec!["id".to_string()],
column_info: Vec::new(),
diff: DataCompareResult {
added: vec![DataCompareRow {
key: "1".to_string(),
key_values: HashMap::from([(String::from("id"), serde_json::json!(1))]),
values: HashMap::from([(String::from("id"), serde_json::json!(1))]),
}],
removed: Vec::new(),
modified: Vec::new(),
},
database_type: Some(DatabaseType::Postgres),
pre_sync_statements: Vec::new(),
sync_eligible: false,
}],
});
assert_eq!(plan.statement_count, 0);
assert!(plan.sync_statements.is_empty());
assert!(plan.sync_sql.is_empty());
}
/// Verify that DataComparePreparation (the original struct) remains unchanged
#[test]
fn data_compare_preparation_format_stable() {
+26
View File
@@ -0,0 +1,26 @@
#[path = "../src/data/sample_pairing.rs"]
mod sample_pairing;
use sample_pairing::retain_shared_sample_rows_by_key;
#[test]
fn independently_sampled_rows_keep_only_shared_keys() {
let source = vec![(1, "source-only"), (2, "shared")];
let target = vec![(2, "shared"), (3, "target-only")];
let (source, target) = retain_shared_sample_rows_by_key(source, target, |row| row.0);
assert_eq!(source, vec![(2, "shared")]);
assert_eq!(target, vec![(2, "shared")]);
}
#[test]
fn paired_samples_preserve_real_duplicate_keys() {
let source = vec![(1, "first"), (1, "duplicate")];
let target = vec![(1, "target")];
let (source, target) = retain_shared_sample_rows_by_key(source, target, |row| row.0);
assert_eq!(source, vec![(1, "first"), (1, "duplicate")]);
assert_eq!(target, vec![(1, "target")]);
}
+71
View File
@@ -20,6 +20,30 @@ pub enum DatabaseRuntimeMode {
External,
}
#[derive(Debug, Clone, Copy, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "lowercase")]
pub enum DatabaseTransferObjectFamily {
Mysql,
Postgres,
Oracle,
SqlServer,
}
#[derive(Debug, Clone, Copy, Deserialize, PartialEq, Eq, PartialOrd, Ord)]
#[serde(rename_all = "kebab-case")]
pub enum DatabaseTransferSupportLevel {
BestEffort,
FamilyCompatible,
Verified,
}
#[derive(Debug, Clone, Copy, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "lowercase")]
pub enum DatabaseCompareMode {
Schema,
Data,
}
#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct DatabaseDriverProfile {
@@ -49,6 +73,12 @@ pub struct ManagedDriverEntry {
#[serde(rename_all = "camelCase")]
pub struct DatabaseManifestEntry {
pub db_type: DatabaseType,
#[serde(default)]
pub compare_modes: Vec<DatabaseCompareMode>,
#[serde(default)]
pub transfer_object_family: Option<DatabaseTransferObjectFamily>,
#[serde(default)]
pub transfer_support_level: Option<DatabaseTransferSupportLevel>,
pub label: String,
#[serde(default)]
pub dialect: Option<String>,
@@ -106,6 +136,18 @@ pub fn dialect_name(db_type: &DatabaseType) -> Option<&'static str> {
entry(db_type).and_then(|candidate| candidate.dialect.as_deref())
}
pub fn transfer_object_family(db_type: &DatabaseType) -> Option<DatabaseTransferObjectFamily> {
entry(db_type).and_then(|candidate| candidate.transfer_object_family)
}
pub fn transfer_support_level(db_type: &DatabaseType) -> Option<DatabaseTransferSupportLevel> {
entry(db_type).and_then(|candidate| candidate.transfer_support_level)
}
pub fn supports_compare_mode(db_type: &DatabaseType, mode: DatabaseCompareMode) -> bool {
entry(db_type).is_some_and(|candidate| candidate.compare_modes.contains(&mode))
}
pub fn is_agent_runtime(db_type: &DatabaseType) -> bool {
entry(db_type).is_some_and(|candidate| candidate.runtime_mode == DatabaseRuntimeMode::Agent)
}
@@ -142,4 +184,33 @@ mod tests {
assert_eq!(entry(&DatabaseType::Sqlite).map(|entry| entry.skip_tcp_probe), Some(true));
assert_eq!(entry(&DatabaseType::H2).map(|entry| entry.agent_key.as_deref()), Some(Some("h2")));
}
#[test]
fn transfer_object_families_come_from_the_shared_manifest() {
assert_eq!(transfer_object_family(&DatabaseType::Mysql), Some(DatabaseTransferObjectFamily::Mysql));
assert_eq!(transfer_object_family(&DatabaseType::Kingbase), Some(DatabaseTransferObjectFamily::Postgres));
assert_eq!(transfer_object_family(&DatabaseType::Dameng), Some(DatabaseTransferObjectFamily::Oracle));
assert_eq!(transfer_object_family(&DatabaseType::SqlServer), Some(DatabaseTransferObjectFamily::SqlServer));
assert_eq!(transfer_object_family(&DatabaseType::Sqlite), None);
}
#[test]
fn transfer_support_levels_distinguish_evidence_strength() {
assert_eq!(transfer_support_level(&DatabaseType::Mysql), Some(DatabaseTransferSupportLevel::Verified));
assert_eq!(
transfer_support_level(&DatabaseType::Kingbase),
Some(DatabaseTransferSupportLevel::FamilyCompatible)
);
assert_eq!(transfer_support_level(&DatabaseType::Db2), Some(DatabaseTransferSupportLevel::BestEffort));
assert_eq!(transfer_support_level(&DatabaseType::Gbase), None);
}
#[test]
fn compare_modes_come_from_the_shared_manifest() {
assert!(supports_compare_mode(&DatabaseType::Mysql, DatabaseCompareMode::Schema));
assert!(supports_compare_mode(&DatabaseType::Mysql, DatabaseCompareMode::Data));
assert!(supports_compare_mode(&DatabaseType::Kingbase, DatabaseCompareMode::Data));
assert!(!supports_compare_mode(&DatabaseType::MongoDb, DatabaseCompareMode::Data));
assert!(!supports_compare_mode(&DatabaseType::Redis, DatabaseCompareMode::Schema));
}
}
+21 -485
View File
@@ -3,9 +3,7 @@ use std::sync::Arc;
use axum::extract::{Path, State};
use axum::response::sse::{Event, Sse};
use axum::Json;
use dbx_core::persistence::task_history::{
TaskHistoryStorageError, TaskItemStatus, TaskLifecycleOwner, TransferTaskJournal,
};
use dbx_core::persistence::task_history::{TaskHistoryStorageError, TaskLifecycleOwner, TransferTaskJournal};
use dbx_core::transfer::{self, TransferRequest, TransferStatus};
use futures::stream::Stream;
use serde::Deserialize;
@@ -147,10 +145,6 @@ pub async fn start_transfer(
let state_clone = state.clone();
tokio::spawn(async move {
// Cross-family object transfers are validated inside transfer_schema_objects:
// only mechanically rewriteable kinds (views, sequences) are allowed; any
// other selection fails with a descriptive error. Structure-only data
// transfer is unsupported for MongoDB.
if matches!(req.content, transfer::TransferContent::StructureOnly)
&& (matches!(source_db_type, dbx_core::models::connection::DatabaseType::MongoDb)
|| matches!(target_db_type, dbx_core::models::connection::DatabaseType::MongoDb))
@@ -163,8 +157,6 @@ pub async fn start_transfer(
return;
}
// External Doris/StarRocks catalogs: pool is created with `catalog=` URL
// setup (SET catalog) and without USE <external-db>. See ensure_transfer_pool.
let source_pool_key = match transfer::ensure_transfer_pool(
&app,
&req.source_connection_id,
@@ -173,9 +165,9 @@ pub async fn start_transfer(
)
.await
{
Ok(k) => k,
Err(e) => {
send_transfer_progress(&progress_channel, &terminal_transfer_error(&req, e));
Ok(key) => key,
Err(error) => {
send_transfer_progress(&progress_channel, &terminal_transfer_error(&req, error));
finish_transfer_channel(&state_clone, &req.transfer_id, &progress_channel).await;
return;
}
@@ -188,491 +180,35 @@ pub async fn start_transfer(
)
.await
{
Ok(k) => k,
Err(e) => {
send_transfer_progress(&progress_channel, &terminal_transfer_error(&req, e));
Ok(key) => key,
Err(error) => {
send_transfer_progress(&progress_channel, &terminal_transfer_error(&req, error));
finish_transfer_channel(&state_clone, &req.transfer_id, &progress_channel).await;
return;
}
};
if let Err(e) = transfer::ensure_transfer_source_types_supported(&app, &req, &source_pool_key).await {
send_transfer_progress(&progress_channel, &terminal_transfer_error(&req, e));
if let Err(error) = transfer::ensure_transfer_source_types_supported(&app, &req, &source_pool_key).await {
send_transfer_progress(&progress_channel, &terminal_transfer_error(&req, error));
finish_transfer_channel(&state_clone, &req.transfer_id, &progress_channel).await;
return;
}
let tables = req.tables.clone();
// Sort by FK dependency so referenced tables are transferred first, and
// keep the foreign key metadata fetched along the way — MySQL-family
// targets reuse it per table below instead of re-querying it.
// Skip for external Doris/StarRocks catalogs — the database name does
// not exist in the default catalog and sorting is unnecessary.
let (tables, known_foreign_keys) = {
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, std::collections::HashMap::new())
} else {
transfer::sort_tables_by_fk_dependency_with_foreign_keys(
&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, std::collections::HashMap::new())
})
}
};
let mut failed_tables: Vec<String> = Vec::new();
let mut pending_fk_alters: Vec<(String, String)> = Vec::new();
// When drop_target_before_create is true, perform a rename pre-pass in
// parents_first=false order (children first) so that foreign keys on the
// backup tables remain intact. The main create/insert pass then runs in
// parents_first=true order (parents first).
// Children-first order used by the rename pre-pass, kept for the post-loop backup
// cleanup: a backup can still be referenced by another backup, so the drops must
// follow the same order.
let mut backup_drop_order: Vec<String> = Vec::new();
let backup_names = if req.drop_target_before_create {
// Re-sort tables in children-first order for the rename pre-pass
let (tables_for_rename, _) = transfer::sort_tables_by_fk_dependency_with_foreign_keys(
&app,
&req.target_connection_id,
&req.target_database,
&req.target_schema,
&tables,
false, // parents_first=false: children first
)
.await
.unwrap_or_else(|e| {
log::warn!("[transfer] failed to sort tables for rename pre-pass, using original order: {e}");
(tables.clone(), std::collections::HashMap::new())
});
backup_drop_order = tables_for_rename.clone();
let progress_channel_clone = progress_channel.clone();
match transfer::rename_tables_to_backup(
&app,
&req,
&tables_for_rename,
let task_transfer_id = req.transfer_id.clone();
let progress_channel_for_run = progress_channel.clone();
let terminal = dbx_core::transfer_runner::run_transfer(
app,
req,
source_db_type,
target_db_type,
&target_pool_key,
|progress| {
send_transfer_progress(&progress_channel_clone, &progress);
},
)
.await
{
Ok(names) => Some(names),
Err(e) if e == "Cancelled" => {
let progress = transfer::TransferProgress {
transfer_id: req.transfer_id.clone(),
table: "rename pre-pass".to_string(),
table_index: 0,
total_tables: tables.len(),
rows_transferred: 0,
total_rows: None,
status: TransferStatus::Cancelled,
error: None,
terminal: true,
};
send_transfer_progress(&progress_channel, &progress);
finish_transfer_channel(&state_clone, &req.transfer_id, &progress_channel).await;
return;
}
Err(e) => {
let progress = transfer::TransferProgress {
transfer_id: req.transfer_id.clone(),
table: "rename pre-pass".to_string(),
table_index: 0,
total_tables: tables.len(),
rows_transferred: 0,
total_rows: None,
status: TransferStatus::Error,
error: Some(e),
terminal: true,
};
send_transfer_progress(&progress_channel, &progress);
finish_transfer_channel(&state_clone, &req.transfer_id, &progress_channel).await;
return;
}
}
} else {
None
};
if matches!(source_db_type, dbx_core::models::connection::DatabaseType::Postgres)
&& matches!(target_db_type, dbx_core::models::connection::DatabaseType::Postgres)
{
let progress_channel_clone = progress_channel.clone();
match transfer::transfer_postgres_schema_dependencies(
&app,
&req,
&source_pool_key,
&target_pool_key,
|progress| {
send_transfer_progress(&progress_channel_clone, &progress);
},
)
.await
{
Ok(()) => {}
Err(e) if e == "Cancelled" => {
let progress = transfer::TransferProgress {
transfer_id: req.transfer_id.clone(),
table: "schema dependencies".to_string(),
table_index: 0,
total_tables: tables.len(),
rows_transferred: 0,
total_rows: None,
status: TransferStatus::Cancelled,
error: None,
terminal: true,
};
send_transfer_progress(&progress_channel, &progress);
finish_transfer_channel(&state_clone, &req.transfer_id, &progress_channel).await;
return;
}
Err(e) => {
let progress = transfer::TransferProgress {
transfer_id: req.transfer_id.clone(),
table: "schema dependencies".to_string(),
table_index: 0,
total_tables: tables.len(),
rows_transferred: 0,
total_rows: None,
status: TransferStatus::Error,
error: Some(e),
terminal: true,
};
send_transfer_progress(&progress_channel, &progress);
finish_transfer_channel(&state_clone, &req.transfer_id, &progress_channel).await;
return;
}
}
}
// Overwrite clears each target right before copying it, parents first, which cannot
// clear a table another selected table references. Empty those children first now.
let overwrite_cleared = match transfer::clear_foreign_key_linked_overwrite_targets(
&app,
&req,
&tables,
target_db_type,
&target_pool_key,
)
.await
{
Ok(cleared) => cleared,
Err(e) => {
let progress = transfer::TransferProgress {
transfer_id: req.transfer_id.clone(),
table: "overwrite pre-pass".to_string(),
table_index: 0,
total_tables: tables.len(),
rows_transferred: 0,
total_rows: None,
status: TransferStatus::Error,
error: Some(e),
terminal: true,
};
send_transfer_progress(&progress_channel, &progress);
finish_transfer_channel(&state_clone, &req.transfer_id, &progress_channel).await;
return;
}
};
for (i, table) in tables.iter().enumerate() {
if transfer::is_cancelled(&req.transfer_id).await {
let progress = transfer::TransferProgress {
transfer_id: req.transfer_id.clone(),
table: table.clone(),
table_index: i,
total_tables: tables.len(),
rows_transferred: 0,
total_rows: None,
status: TransferStatus::Cancelled,
error: None,
terminal: true,
};
send_transfer_progress(&progress_channel, &progress);
finish_transfer_channel(&state_clone, &req.transfer_id, &progress_channel).await;
return;
}
let history_item_index =
history.as_ref().and_then(|journal| journal.item_index_for_table(table)).unwrap_or(i);
if let Some(journal) = history.as_ref() {
journal.start_table(history_item_index).await;
}
let progress_channel_clone = progress_channel.clone();
let mut last_rows_transferred = 0_u64;
let mut last_total_rows = None;
let mut source_row_count = None;
let mut moved_row_count = None;
let history_for_progress = history.as_ref();
let history_for_count = history.as_ref();
let result = transfer::transfer_table_with_result(
&app,
&req,
table,
i,
&source_db_type,
&target_db_type,
&source_pool_key,
&target_pool_key,
&known_foreign_keys,
&mut pending_fk_alters,
backup_names.as_ref(),
overwrite_cleared.contains(table),
|progress| {
last_rows_transferred = progress.rows_transferred;
last_total_rows = progress.total_rows;
moved_row_count = Some(progress.rows_transferred);
if let Some(journal) = history_for_progress {
journal.observe_table_progress(history_item_index, progress.rows_transferred);
}
send_transfer_progress(&progress_channel_clone, &progress);
},
|source_count| {
source_row_count = source_count;
if let Some(journal) = history_for_count {
journal.observe_source_count(history_item_index, source_count);
}
},
source_pool_key,
target_pool_key,
history,
move |progress| send_transfer_progress(&progress_channel_for_run, &progress),
)
.await;
match result {
Ok(result) => {
if let Some(journal) = history.as_ref() {
journal
.finish_table(
history_item_index,
TaskItemStatus::Succeeded,
result.source_row_count,
Some(result.moved_rows),
)
.await;
}
let progress = transfer::TransferProgress {
transfer_id: req.transfer_id.clone(),
table: table.clone(),
table_index: i,
total_tables: tables.len(),
rows_transferred: result.moved_rows,
total_rows: last_total_rows.or(Some(result.moved_rows)),
status: TransferStatus::TableDone,
error: None,
terminal: false,
};
send_transfer_progress(&progress_channel, &progress);
}
Err(e) => {
if e == "Cancelled" {
if let Some(journal) = history.as_ref() {
journal
.finish_table(
history_item_index,
TaskItemStatus::Cancelled,
source_row_count,
moved_row_count,
)
.await;
}
let progress = transfer::TransferProgress {
transfer_id: req.transfer_id.clone(),
table: table.clone(),
table_index: i,
total_tables: tables.len(),
rows_transferred: 0,
total_rows: None,
status: TransferStatus::Cancelled,
error: None,
terminal: true,
};
send_transfer_progress(&progress_channel, &progress);
finish_transfer_channel(&state_clone, &req.transfer_id, &progress_channel).await;
return;
}
if let Some(journal) = history.as_ref() {
journal
.finish_table(history_item_index, TaskItemStatus::Failed, source_row_count, moved_row_count)
.await;
}
failed_tables.push(table.clone());
let progress = transfer::TransferProgress {
transfer_id: req.transfer_id.clone(),
table: table.clone(),
table_index: i,
total_tables: tables.len(),
rows_transferred: last_rows_transferred,
total_rows: last_total_rows,
status: TransferStatus::Error,
error: Some(e),
terminal: false,
};
send_transfer_progress(&progress_channel, &progress);
}
}
}
// Add any foreign keys deferred during MySQL-family table creation now that
// every selected table exists — see transfer_table's use of
// strip_inline_foreign_key_constraint_lines for why these can't be created
// inline (a foreign key cycle has no valid CREATE TABLE order at all). The
// rename pre-pass already freed the backup constraint names, so these re-create
// the original names without colliding.
let mut failed_fk_tables: Vec<String> = Vec::new();
let mut failed_fk_count = 0usize;
for (table, alter_sql) in &pending_fk_alters {
if let Err(e) = transfer::execute_on_pool(&app, &target_pool_key, alter_sql).await {
log::warn!("[transfer] failed to add deferred foreign key constraint for {table}: {e}");
failed_fk_count += 1;
failed_fk_tables.push(table.clone());
}
}
if failed_fk_count > 0 {
failed_fk_tables.sort();
failed_fk_tables.dedup();
failed_tables.push(format!("{} foreign key(s) on: {}", failed_fk_count, failed_fk_tables.join(", ")));
}
// Transfer selected non-table objects (views, procedures, functions,
// triggers, sequences, events) after the per-table loop. The shared
// Core decision handles all content modes: DataOnly never
// transfers schema objects; PG→PG keeps the legacy empty-selection
// default only when structure participates in the transfer.
let mut object_outcome = transfer::TransferObjectOutcome::default();
let progress_channel_clone = progress_channel.clone();
match transfer::transfer_schema_objects(&app, &req, &source_pool_key, &target_pool_key, |progress| {
send_transfer_progress(&progress_channel_clone, &progress);
})
.await
{
Ok(outcome) => {
if let Some(journal) = history.as_ref() {
journal.record_object_outcome(&outcome).await;
}
object_outcome = outcome;
}
Err(e) if e == "Cancelled" => {
if let Some(journal) = history.as_ref() {
journal.record_schema_objects_error(true).await;
}
let progress = transfer::TransferProgress {
transfer_id: req.transfer_id.clone(),
table: "schema objects".to_string(),
table_index: tables.len(),
total_tables: tables.len(),
rows_transferred: 0,
total_rows: None,
status: TransferStatus::Cancelled,
error: None,
terminal: true,
};
send_transfer_progress(&progress_channel, &progress);
finish_transfer_channel(&state_clone, &req.transfer_id, &progress_channel).await;
return;
}
Err(e) => {
if let Some(journal) = history.as_ref() {
journal.record_schema_objects_error(false).await;
}
failed_tables.push("schema objects".to_string());
let progress = transfer::TransferProgress {
transfer_id: req.transfer_id.clone(),
table: "schema objects".to_string(),
table_index: tables.len(),
total_tables: tables.len(),
rows_transferred: 0,
total_rows: None,
status: TransferStatus::Error,
error: Some(e),
terminal: false,
};
send_transfer_progress(&progress_channel, &progress);
}
}
// Send done
if !object_outcome.failed.is_empty() {
failed_tables.push(format!("schema objects ({})", object_outcome.failed.len()));
}
// The rename pre-pass left one backup per rebuilt table. Drop them only now that
// every table, every deferred foreign key and every selected schema object has
// succeeded — a failure anywhere above keeps the originals recoverable under
// their backup names.
if let Some(backup_names) = backup_names.as_ref() {
if failed_tables.is_empty() {
if let Err(e) = transfer::drop_backup_tables(
&app,
&req,
target_db_type,
&target_pool_key,
backup_names,
&backup_drop_order,
)
.await
{
failed_tables.push(e);
}
} else {
log::warn!(
"[transfer] keeping {} backup table(s) because {} step(s) failed",
backup_names.len(),
failed_tables.len()
);
}
}
let skip_suffix = if !object_outcome.skipped.is_empty() && failed_tables.is_empty() {
format!(",跳过 {} 个已存在对象", object_outcome.skipped.len())
} else if !object_outcome.skipped.is_empty() {
format!(";跳过 {} 个已存在对象", object_outcome.skipped.len())
} else {
String::new()
};
let done = transfer::TransferProgress {
transfer_id: req.transfer_id.clone(),
table: String::new(),
table_index: tables.len(),
total_tables: tables.len(),
rows_transferred: 0,
total_rows: None,
status: if failed_tables.is_empty() { TransferStatus::Done } else { TransferStatus::Error },
error: if failed_tables.is_empty() {
if skip_suffix.is_empty() {
None
} else {
Some(skip_suffix.clone())
}
} else {
Some(format!(
"{} table(s) failed: {}{}",
failed_tables.len(),
failed_tables.iter().take(5).cloned().collect::<Vec<_>>().join(", "),
skip_suffix
))
},
terminal: true,
};
send_transfer_progress(&progress_channel, &done);
finish_transfer_channel(&state_clone, &req.transfer_id, &progress_channel).await;
send_transfer_progress(&progress_channel, &terminal);
finish_transfer_channel(&state_clone, &task_transfer_id, &progress_channel).await;
});
Ok(Json(serde_json::json!({ "transferId": transfer_id })))
+1
View File
@@ -1,6 +1,7 @@
schemaVersion: 1
order: 460
dbType: access
compareModes: [schema, data]
rustVariant: Access
label: Microsoft Access
dialect: Access
+2
View File
@@ -1,6 +1,8 @@
schemaVersion: 1
order: 515
dbType: argo
compareModes: [schema, data]
transferSupportLevel: best-effort
rustVariant: Argo
label: 星环Argo
runtimeMode: agent
+1
View File
@@ -1,6 +1,7 @@
schemaVersion: 1
order: 580
dbType: bigquery
compareModes: [schema, data]
rustVariant: Bigquery
label: Google BigQuery
runtimeMode: agent
+2
View File
@@ -1,6 +1,8 @@
schemaVersion: 1
order: 90
dbType: clickhouse
compareModes: [schema, data]
transferSupportLevel: best-effort
rustVariant: ClickHouse
label: ClickHouse
dialect: ClickHouse
@@ -1,6 +1,8 @@
schemaVersion: 1
order: 60
dbType: cloudflare-d1
compareModes: [schema, data]
transferSupportLevel: best-effort
rustVariant: CloudflareD1
label: Cloudflare D1
dialect: SQLite
+3
View File
@@ -1,6 +1,9 @@
schemaVersion: 1
order: 260
dbType: dameng
compareModes: [schema, data]
transferSupportLevel: family-compatible
transferObjectFamily: oracle
rustVariant: Dameng
label: 达梦 DM8
dialect: Dameng
+1
View File
@@ -1,6 +1,7 @@
schemaVersion: 1
order: 320
dbType: databend
compareModes: [schema, data]
rustVariant: Databend
label: Databend
dialect: Databend
+1
View File
@@ -1,6 +1,7 @@
schemaVersion: 1
order: 360
dbType: databricks
compareModes: [schema, data]
rustVariant: Databricks
label: Databricks SQL
runtimeMode: agent
+2
View File
@@ -1,6 +1,8 @@
schemaVersion: 1
order: 540
dbType: db2
compareModes: [schema, data]
transferSupportLevel: best-effort
rustVariant: Db2
label: IBM DB2
runtimeMode: agent
+1
View File
@@ -1,6 +1,7 @@
schemaVersion: 1
order: 220
dbType: doris
compareModes: [schema, data]
rustVariant: Doris
label: Doris
dialect: Doris
+2
View File
@@ -1,6 +1,8 @@
schemaVersion: 1
order: 80
dbType: duckdb
compareModes: [schema, data]
transferSupportLevel: best-effort
rustVariant: DuckDb
label: DuckDB
dialect: DuckDB
+1
View File
@@ -1,6 +1,7 @@
schemaVersion: 1
order: 410
dbType: exasol
compareModes: [schema, data]
rustVariant: Exasol
label: Exasol
dialect: Exasol
+1
View File
@@ -1,6 +1,7 @@
schemaVersion: 1
order: 400
dbType: firebird
compareModes: [schema, data]
rustVariant: Firebird
label: Firebird
dialect: Firebird
+3
View File
@@ -1,6 +1,9 @@
schemaVersion: 1
order: 330
dbType: gaussdb
compareModes: [schema, data]
transferSupportLevel: family-compatible
transferObjectFamily: postgres
rustVariant: Gaussdb
label: GaussDB
dialect: GaussDB
+2
View File
@@ -1,6 +1,8 @@
schemaVersion: 1
order: 450
dbType: gbase
compareModes: [schema, data]
transferObjectFamily: mysql
rustVariant: Gbase
label: 南大通用 GBase 8a
dialect: GBase
+2
View File
@@ -1,6 +1,8 @@
schemaVersion: 1
order: 310
dbType: goldendb
compareModes: [schema, data]
transferSupportLevel: best-effort
rustVariant: Goldendb
label: 金篆 GoldenDB
dialect: GoldenDB
+2
View File
@@ -1,6 +1,8 @@
schemaVersion: 1
order: 470
dbType: h2
compareModes: [schema, data]
transferSupportLevel: best-effort
rustVariant: H2
label: H2
dialect: H2
+2
View File
@@ -1,6 +1,8 @@
schemaVersion: 1
order: 280
dbType: highgo
compareModes: [schema, data]
transferSupportLevel: best-effort
rustVariant: Highgo
label: 瀚高 HighGo
dialect: HighGo
+2
View File
@@ -1,6 +1,8 @@
schemaVersion: 1
order: 510
dbType: hive
compareModes: [schema, data]
transferSupportLevel: best-effort
rustVariant: Hive
label: Apache Hive
runtimeMode: agent
+1
View File
@@ -1,6 +1,7 @@
schemaVersion: 1
order: 610
dbType: ignite
compareModes: [schema, data]
rustVariant: Ignite
label: Apache Ignite
runtimeMode: agent
+1
View File
@@ -1,6 +1,7 @@
schemaVersion: 1
order: 620
dbType: ignite3
compareModes: [schema, data]
rustVariant: Ignite3
label: Apache Ignite 3
runtimeMode: agent
+2
View File
@@ -1,6 +1,8 @@
schemaVersion: 1
order: 530
dbType: impala
compareModes: [schema, data]
transferSupportLevel: best-effort
rustVariant: Impala
label: Apache Impala
runtimeMode: agent
+1
View File
@@ -1,6 +1,7 @@
schemaVersion: 1
order: 755
dbType: influxdb3
compareModes: [schema, data]
rustVariant: InfluxDb3
label: InfluxDB 3.x
runtimeMode: native
+1
View File
@@ -1,6 +1,7 @@
schemaVersion: 1
order: 550
dbType: informix
compareModes: [schema, data]
rustVariant: Informix
label: IBM Informix
dialect: Informix
+1
View File
@@ -1,6 +1,7 @@
schemaVersion: 1
order: 730
dbType: iotdb
compareModes: [schema, data]
rustVariant: Iotdb
label: Apache IoTDB
runtimeMode: agent
+2
View File
@@ -1,6 +1,8 @@
schemaVersion: 1
order: 740
dbType: iris
compareModes: [schema, data]
transferSupportLevel: best-effort
rustVariant: Iris
label: InterSystems IRIS
dialect: Iris
+3
View File
@@ -1,6 +1,9 @@
schemaVersion: 1
order: 270
dbType: kingbase
compareModes: [schema, data]
transferSupportLevel: family-compatible
transferObjectFamily: postgres
rustVariant: Kingbase
label: 金仓KingbaseES
dialect: Kingbase
+3
View File
@@ -1,6 +1,9 @@
schemaVersion: 1
order: 340
dbType: kwdb
compareModes: [schema, data]
transferSupportLevel: family-compatible
transferObjectFamily: postgres
rustVariant: Kwdb
label: KWDB
dialect: KwDB
+1
View File
@@ -1,6 +1,7 @@
schemaVersion: 1
order: 600
dbType: kylin
compareModes: [schema, data]
rustVariant: Kylin
label: Apache Kylin
runtimeMode: agent
+2
View File
@@ -1,6 +1,8 @@
schemaVersion: 1
order: 520
dbType: kyuubi
compareModes: [schema, data]
transferSupportLevel: best-effort
rustVariant: Kyuubi
label: Apache Kyuubi
runtimeMode: agent
@@ -1,6 +1,7 @@
schemaVersion: 1
order: 240
dbType: manticoresearch
compareModes: [schema, data]
rustVariant: ManticoreSearch
label: Manticore Search
dialect: ManticoreSearch
+1
View File
@@ -1,6 +1,7 @@
schemaVersion: 1
order: 110
dbType: mongodb
transferSupportLevel: best-effort
rustVariant: MongoDb
label: MongoDB (Legacy)
runtimeMode: agent
+3
View File
@@ -1,6 +1,9 @@
schemaVersion: 1
order: 10
dbType: mysql
compareModes: [schema, data]
transferSupportLevel: verified
transferObjectFamily: mysql
rustVariant: Mysql
label: MySQL
dialect: MySQL
@@ -1,6 +1,9 @@
schemaVersion: 1
order: 440
dbType: oceanbase-oracle
compareModes: [schema, data]
transferSupportLevel: family-compatible
transferObjectFamily: oracle
rustVariant: OceanbaseOracle
label: OceanBase Oracle Mode
dialect: OceanBase
+3
View File
@@ -1,6 +1,9 @@
schemaVersion: 1
order: 420
dbType: opengauss
compareModes: [schema, data]
transferSupportLevel: family-compatible
transferObjectFamily: postgres
rustVariant: OpenGauss
label: openGauss
dialect: openGauss
+3
View File
@@ -1,6 +1,9 @@
schemaVersion: 1
order: 130
dbType: oracle
compareModes: [schema, data]
transferSupportLevel: verified
transferObjectFamily: oracle
rustVariant: Oracle
label: Oracle
dialect: Oracle
+1
View File
@@ -1,6 +1,7 @@
schemaVersion: 1
order: 640
dbType: oscar
compareModes: [schema, data]
rustVariant: Oscar
label: 神通 OSCAR
runtimeMode: agent
+3
View File
@@ -1,6 +1,9 @@
schemaVersion: 1
order: 20
dbType: postgres
compareModes: [schema, data]
transferSupportLevel: verified
transferObjectFamily: postgres
rustVariant: Postgres
label: PostgreSQL
dialect: PostgreSQL
+1
View File
@@ -1,6 +1,7 @@
schemaVersion: 1
order: 500
dbType: prestosql
compareModes: [schema, data]
rustVariant: PrestoSql
label: PrestoSQL
runtimeMode: external
+2
View File
@@ -1,6 +1,8 @@
schemaVersion: 1
order: 430
dbType: questdb
compareModes: [schema, data]
transferSupportLevel: best-effort
rustVariant: Questdb
label: QuestDB
dialect: QuestDB
+1
View File
@@ -1,6 +1,7 @@
schemaVersion: 1
order: 250
dbType: redshift
compareModes: [schema, data]
rustVariant: Redshift
label: Redshift
dialect: Redshift
+2
View File
@@ -1,6 +1,8 @@
schemaVersion: 1
order: 40
dbType: rqlite
compareModes: [schema, data]
transferSupportLevel: best-effort
rustVariant: Rqlite
label: RQLite
dialect: Rqlite
+1
View File
@@ -1,6 +1,7 @@
schemaVersion: 1
order: 370
dbType: saphana
compareModes: [schema, data]
rustVariant: SapHana
label: SAP HANA
runtimeMode: agent
+1
View File
@@ -1,6 +1,7 @@
schemaVersion: 1
order: 480
dbType: snowflake
compareModes: [schema, data]
rustVariant: Snowflake
label: Snowflake
runtimeMode: agent
+1
View File
@@ -1,6 +1,7 @@
schemaVersion: 1
order: 590
dbType: spanner
compareModes: [schema, data]
rustVariant: Spanner
label: Google Cloud Spanner
runtimeMode: agent
+2
View File
@@ -1,6 +1,8 @@
schemaVersion: 1
order: 780
dbType: spark
compareModes: [schema, data]
transferSupportLevel: best-effort
rustVariant: Spark
label: Apache Spark
runtimeMode: agent
+2
View File
@@ -1,6 +1,8 @@
schemaVersion: 1
order: 30
dbType: sqlite
compareModes: [schema, data]
transferSupportLevel: verified
rustVariant: Sqlite
label: SQLite
dialect: SQLite
+3
View File
@@ -1,6 +1,9 @@
schemaVersion: 1
order: 100
dbType: sqlserver
compareModes: [schema, data]
transferSupportLevel: verified
transferObjectFamily: sqlserver
rustVariant: SqlServer
label: SQL Server
dialect: SQL Server
+1
View File
@@ -1,6 +1,7 @@
schemaVersion: 1
order: 230
dbType: starrocks
compareModes: [schema, data]
rustVariant: StarRocks
label: StarRocks
dialect: StarRocks
+1
View File
@@ -1,6 +1,7 @@
schemaVersion: 1
order: 630
dbType: sundb
compareModes: [schema, data]
rustVariant: Sundb
label: 科蓝 SUNDB
dialect: SunDB
+1
View File
@@ -1,6 +1,7 @@
schemaVersion: 1
order: 650
dbType: tdengine
compareModes: [schema, data]
rustVariant: Tdengine
label: TDengine
runtimeMode: agent
+1
View File
@@ -1,6 +1,7 @@
schemaVersion: 1
order: 380
dbType: teradata
compareModes: [schema, data]
rustVariant: Teradata
label: Teradata
runtimeMode: agent
+2
View File
@@ -2,6 +2,8 @@ schemaVersion: 1
order: 516
dbType: transwarp
compareModes: [schema, data]
transferSupportLevel: best-effort
rustVariant: Transwarp
label: 星环Inceptor
dialect: Transwarp
+1
View File
@@ -1,6 +1,7 @@
schemaVersion: 1
order: 490
dbType: trino
compareModes: [schema, data]
rustVariant: Trino
label: Trino
runtimeMode: agent
+2
View File
@@ -1,6 +1,8 @@
schemaVersion: 1
order: 50
dbType: turso
compareModes: [schema, data]
transferSupportLevel: best-effort
rustVariant: Turso
label: Turso
dialect: Turso
+1
View File
@@ -1,6 +1,7 @@
schemaVersion: 1
order: 290
dbType: uxdb
compareModes: [schema, data]
rustVariant: Uxdb
label: 优炫 UXDB
dialect: PostgreSQL
+2
View File
@@ -1,6 +1,8 @@
schemaVersion: 1
order: 300
dbType: vastbase
compareModes: [schema, data]
transferSupportLevel: best-effort
rustVariant: Vastbase
label: 海量 Vastbase
dialect: Vastbase
+1
View File
@@ -1,6 +1,7 @@
schemaVersion: 1
order: 390
dbType: vertica
compareModes: [schema, data]
rustVariant: Vertica
label: Vertica
dialect: Vertica
+2
View File
@@ -1,6 +1,8 @@
schemaVersion: 1
order: 660
dbType: xugu
compareModes: [schema, data]
transferSupportLevel: best-effort
rustVariant: Xugu
label: 虚谷 XuguDB
dialect: Xugu
+2
View File
@@ -1,6 +1,8 @@
schemaVersion: 1
order: 350
dbType: yashandb
compareModes: [schema, data]
transferSupportLevel: best-effort
rustVariant: Yashandb
label: 崖山 YashanDB
dialect: YashanDB
+32
View File
@@ -65,6 +65,7 @@ test("core owns application orchestration in real business directories", () => {
"schema/mod.rs",
"schema/table_structure_sql/sqlite_rebuild.rs",
"data/transfer.rs",
"data/transfer_runner.rs",
"data/table_import.rs",
"data/docs/export.rs",
"data/cloudflare_d1/import.rs",
@@ -88,6 +89,37 @@ test("core owns application orchestration in real business directories", () => {
assert.ok(!existsSync(path.join(root, "crates/dbx-core/build.rs")));
});
test("transfer entrypoints delegate orchestration to the core runner", () => {
const entrypoints = [
"src-tauri/src/commands/transfer.rs",
"crates/dbx-web/src/routes/transfer.rs",
];
const orchestrationCalls = [
"rename_tables_to_backup",
"transfer_postgres_schema_dependencies",
"clear_foreign_key_linked_overwrite_targets",
"transfer_table_with_result",
"transfer_schema_objects",
"drop_backup_tables",
];
for (const entrypoint of entrypoints) {
const source = read(entrypoint);
assert.match(source, /transfer_runner::run_transfer\(/, `${entrypoint} must use the shared transfer runner`);
for (const call of orchestrationCalls) {
const directCall = new RegExp(`(?:dbx_core::)?transfer::${call}\\(`);
assert.doesNotMatch(source, directCall, `${entrypoint} must not own ${call} orchestration`);
}
}
});
test("database comparison admission comes from the shared driver manifest", () => {
const source = read("apps/desktop/src/lib/database/databaseCompareCapabilities.ts");
assert.match(source, /supportsDatabaseCompareMode\(type, mode\)/);
assert.doesNotMatch(source, /new Set<DatabaseType>/);
assert.doesNotMatch(source, /SQL_COMPARE_TYPES/);
});
test("core forwards capability flags without enabling driver defaults or test hooks", () => {
for (const feature of ["duckdb-sidecar", "dynamodb", "mq-admin", "sqlite-bundled", "sqlite-sqlcipher", "sqlite-multiple-ciphers"]) {
assert.ok(core.features[feature].includes(`dbx-drivers/${feature}`), `missing driver forwarding for ${feature}`);
+30
View File
@@ -15,12 +15,18 @@ const checkOnly = process.argv.includes("--check");
const runtimeModes = new Set(["native", "file", "agent", "external"]);
const mcpModes = new Set(["direct", "bridge", "unsupported"]);
const supportLevels = new Set(["connect", "browse", "understand", "operate"]);
const transferObjectFamilies = new Set(["mysql", "postgres", "oracle", "sqlserver"]);
const transferSupportLevels = new Set(["verified", "family-compatible", "best-effort"]);
const compareModes = new Set(["schema", "data"]);
const formKinds = new Set(["standard", "jdbc", "mq", "mqtt", "nacos"]);
const profileCategories = new Set(["sql", "analytics", "domestic", "lightweight", "document", "graph_ai", "timeseries", "mq", "registry_config"]);
const descriptorKeys = new Set([
"schemaVersion",
"order",
"dbType",
"compareModes",
"transferObjectFamily",
"transferSupportLevel",
"rustVariant",
"label",
"dialect",
@@ -95,6 +101,18 @@ function validateDescriptor(descriptor, file, knownDialects) {
if (descriptor.schemaVersion !== 1) throw new Error(`${location}: schemaVersion must be 1`);
if (!Number.isInteger(descriptor.order) || descriptor.order <= 0) throw new Error(`${location}: order must be a positive integer`);
if (!/^[a-z0-9]+(?:-[a-z0-9]+)*$/.test(descriptor.dbType ?? "")) throw new Error(`${location}: invalid dbType`);
if (descriptor.compareModes !== undefined) {
if (!Array.isArray(descriptor.compareModes) || descriptor.compareModes.length === 0 || descriptor.compareModes.some((mode) => !compareModes.has(mode))) {
throw new Error(`${location}: invalid compareModes`);
}
if (new Set(descriptor.compareModes).size !== descriptor.compareModes.length) throw new Error(`${location}: duplicate compareModes`);
}
if (descriptor.transferObjectFamily !== undefined && !transferObjectFamilies.has(descriptor.transferObjectFamily)) {
throw new Error(`${location}: invalid transferObjectFamily`);
}
if (descriptor.transferSupportLevel !== undefined && !transferSupportLevels.has(descriptor.transferSupportLevel)) {
throw new Error(`${location}: invalid transferSupportLevel`);
}
if (!/^[A-Z][A-Za-z0-9]*$/.test(descriptor.rustVariant ?? "")) throw new Error(`${location}: invalid rustVariant`);
if (typeof descriptor.label !== "string" || descriptor.label.trim() === "") throw new Error(`${location}: label is required`);
if (!runtimeModes.has(descriptor.runtimeMode)) throw new Error(`${location}: invalid runtimeMode`);
@@ -109,6 +127,18 @@ function validateDescriptor(descriptor, file, knownDialects) {
}
if (!descriptor.capabilities || typeof descriptor.capabilities !== "object") throw new Error(`${location}: capabilities are required`);
validateBooleanMap(descriptor.capabilities, capabilityKeys, `${location}: capabilities`, true);
if (descriptor.capabilities.dataTransfer !== (descriptor.transferSupportLevel !== undefined)) {
throw new Error(`${location}: transferSupportLevel must be set exactly when dataTransfer is true`);
}
if (descriptor.compareModes?.includes("schema") && !descriptor.capabilities.metadataBrowse) {
throw new Error(`${location}: schema comparison requires metadataBrowse`);
}
if (descriptor.compareModes?.includes("data") && (!descriptor.capabilities.metadataBrowse || !descriptor.capabilities.queryExecution)) {
throw new Error(`${location}: data comparison requires metadataBrowse and queryExecution`);
}
if (descriptor.transferSupportLevel === "family-compatible" && descriptor.transferObjectFamily === undefined) {
throw new Error(`${location}: family-compatible transfer support requires transferObjectFamily`);
}
if (!Object.values(descriptor.capabilities).some(Boolean) && descriptor.specializedSurface !== true) {
throw new Error(`${location}: enable a product capability or set specializedSurface: true`);
}
+10 -525
View File
@@ -5,12 +5,8 @@ use crate::commands::connection::{ensure_connection_writable, AppState};
// Re-export types and functions used by other modules
use dbx_core::models::connection::DatabaseType;
use dbx_core::persistence::task_history::{
TaskHistoryStorageError, TaskItemStatus, TaskLifecycleOwner, TransferTaskJournal,
};
pub use dbx_core::transfer::{
get_db_type, TransferOwnershipPreview, TransferProgress, TransferRequest, TransferStatus,
};
use dbx_core::persistence::task_history::{TaskHistoryStorageError, TaskLifecycleOwner, TransferTaskJournal};
pub use dbx_core::transfer::{get_db_type, TransferOwnershipPreview, TransferProgress, TransferRequest};
fn emit_progress(app: &AppHandle, progress: TransferProgress) {
let _ = app.emit("transfer-progress", progress);
@@ -30,8 +26,6 @@ pub async fn start_transfer(
request: TransferRequest,
) -> Result<(), String> {
let state = state.inner().clone();
let transfer_id = request.transfer_id.clone();
// Reject transfer early if the target connection is read-only — writing to it is inherently required
ensure_connection_writable(&state, &request.target_connection_id, "Transfer").await?;
@@ -88,527 +82,18 @@ pub async fn start_transfer(
};
tokio::spawn(async move {
// Sort tables by FK dependency so referenced tables are transferred first,
// and keep the foreign key metadata fetched along the way — MySQL-family
// targets reuse it per table below instead of re-querying it.
// 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, known_foreign_keys) = {
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(), std::collections::HashMap::new())
} else {
dbx_core::transfer::sort_tables_by_fk_dependency_with_foreign_keys(
&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(), std::collections::HashMap::new())
})
}
};
let total_tables = sorted_tables.len();
log::info!("[transfer] starting transfer_id={} tables={}", transfer_id, total_tables);
let mut failed_tables: Vec<String> = Vec::new();
let mut pending_fk_alters: Vec<(String, String)> = Vec::new();
let mut last_rows_transferred = 0_u64;
let mut last_total_rows = None;
// When drop_target_before_create is true, perform a rename pre-pass in
// parents_first=false order (children first) so that foreign keys on the
// backup tables remain intact. The main create/insert pass then runs in
// parents_first=true order (parents first).
// Children-first order used by the rename pre-pass, kept for the post-loop backup
// cleanup: a backup can still be referenced by another backup, so the drops must
// follow the same order.
let mut backup_drop_order: Vec<String> = Vec::new();
let backup_names = if request.drop_target_before_create {
// Re-sort tables in children-first order for the rename pre-pass
let (tables_for_rename, _) = dbx_core::transfer::sort_tables_by_fk_dependency_with_foreign_keys(
&state,
&request.target_connection_id,
&request.target_database,
&request.target_schema,
&sorted_tables,
false, // parents_first=false: children first
)
.await
.unwrap_or_else(|e| {
log::warn!("[transfer] failed to sort tables for rename pre-pass, using original order: {e}");
(sorted_tables.clone(), std::collections::HashMap::new())
});
backup_drop_order = tables_for_rename.clone();
match dbx_core::transfer::rename_tables_to_backup(
&state,
&request,
&tables_for_rename,
let terminal = dbx_core::transfer_runner::run_transfer(
state,
request,
source_db_type,
target_db_type,
&target_pool_key,
|progress| {
last_rows_transferred = progress.rows_transferred;
last_total_rows = progress.total_rows;
emit_progress(&app, progress);
},
)
.await
{
Ok(names) => Some(names),
Err(e) if e == "Cancelled" => {
emit_terminal_progress(
&app,
history.as_ref(),
TransferProgress {
transfer_id: transfer_id.clone(),
table: "rename pre-pass".to_string(),
table_index: 0,
total_tables,
rows_transferred: last_rows_transferred,
total_rows: last_total_rows,
status: TransferStatus::Cancelled,
error: None,
terminal: true,
},
)
.await;
dbx_core::transfer::clear_cancelled(&transfer_id).await;
return;
}
Err(e) => {
emit_terminal_progress(
&app,
history.as_ref(),
TransferProgress {
transfer_id: transfer_id.clone(),
table: "rename pre-pass".to_string(),
table_index: 0,
total_tables,
rows_transferred: last_rows_transferred,
total_rows: last_total_rows,
status: TransferStatus::Error,
error: Some(e),
terminal: true,
},
)
.await;
dbx_core::transfer::clear_cancelled(&transfer_id).await;
return;
}
}
} else {
None
};
if matches!(source_db_type, dbx_core::models::connection::DatabaseType::Postgres)
&& matches!(target_db_type, dbx_core::models::connection::DatabaseType::Postgres)
{
match dbx_core::transfer::transfer_postgres_schema_dependencies(
&state,
&request,
&source_pool_key,
&target_pool_key,
|progress| {
last_rows_transferred = progress.rows_transferred;
last_total_rows = progress.total_rows;
emit_progress(&app, progress);
},
)
.await
{
Ok(()) => {}
Err(e) if e == "Cancelled" => {
emit_terminal_progress(
&app,
history.as_ref(),
TransferProgress {
transfer_id: transfer_id.clone(),
table: "schema dependencies".to_string(),
table_index: 0,
total_tables,
rows_transferred: last_rows_transferred,
total_rows: last_total_rows,
status: TransferStatus::Cancelled,
error: None,
terminal: true,
},
)
.await;
dbx_core::transfer::clear_cancelled(&transfer_id).await;
return;
}
Err(e) => {
emit_terminal_progress(
&app,
history.as_ref(),
TransferProgress {
transfer_id: transfer_id.clone(),
table: "schema dependencies".to_string(),
table_index: 0,
total_tables,
rows_transferred: last_rows_transferred,
total_rows: last_total_rows,
status: TransferStatus::Error,
error: Some(e),
terminal: true,
},
)
.await;
dbx_core::transfer::clear_cancelled(&transfer_id).await;
return;
}
}
}
// Overwrite clears each target right before copying it, parents first, which cannot
// clear a table another selected table references. Empty those children first now.
let overwrite_cleared = match dbx_core::transfer::clear_foreign_key_linked_overwrite_targets(
&state,
&request,
&sorted_tables,
target_db_type,
&target_pool_key,
)
.await
{
Ok(cleared) => cleared,
Err(e) => {
emit_terminal_progress(
&app,
history.as_ref(),
TransferProgress {
transfer_id: transfer_id.clone(),
table: "overwrite pre-pass".to_string(),
table_index: 0,
total_tables,
rows_transferred: last_rows_transferred,
total_rows: last_total_rows,
status: TransferStatus::Error,
error: Some(e),
terminal: true,
},
)
.await;
dbx_core::transfer::clear_cancelled(&transfer_id).await;
return;
}
};
for (i, table) in sorted_tables.iter().enumerate() {
if dbx_core::transfer::is_cancelled(&transfer_id).await {
emit_terminal_progress(
&app,
history.as_ref(),
TransferProgress {
transfer_id: transfer_id.clone(),
table: table.clone(),
table_index: i,
total_tables,
rows_transferred: last_rows_transferred,
total_rows: last_total_rows,
status: TransferStatus::Cancelled,
error: None,
terminal: true,
},
)
.await;
dbx_core::transfer::clear_cancelled(&transfer_id).await;
return;
}
log::info!("[transfer] table {}/{}: {}", i + 1, total_tables, table);
let history_item_index =
history.as_ref().and_then(|journal| journal.item_index_for_table(table)).unwrap_or(i);
if let Some(journal) = history.as_ref() {
journal.start_table(history_item_index).await;
}
let mut source_row_count = None;
let mut moved_row_count = None;
let history_for_progress = history.as_ref();
let history_for_count = history.as_ref();
let result = dbx_core::transfer::transfer_table_with_result(
&state,
&request,
table,
i,
&source_db_type,
&target_db_type,
&source_pool_key,
&target_pool_key,
&known_foreign_keys,
&mut pending_fk_alters,
backup_names.as_ref(),
overwrite_cleared.contains(table),
|progress| {
last_rows_transferred = progress.rows_transferred;
last_total_rows = progress.total_rows;
moved_row_count = Some(progress.rows_transferred);
if let Some(journal) = history_for_progress {
journal.observe_table_progress(history_item_index, progress.rows_transferred);
}
emit_progress(&app, progress);
},
|source_count| {
source_row_count = source_count;
if let Some(journal) = history_for_count {
journal.observe_source_count(history_item_index, source_count);
}
},
)
.await;
match result {
Ok(result) => {
if let Some(journal) = history.as_ref() {
journal
.finish_table(
history_item_index,
TaskItemStatus::Succeeded,
result.source_row_count,
Some(result.moved_rows),
)
.await;
}
emit_progress(
&app,
TransferProgress {
transfer_id: transfer_id.clone(),
table: table.clone(),
table_index: i,
total_tables,
rows_transferred: result.moved_rows,
total_rows: last_total_rows.or(Some(result.moved_rows)),
status: TransferStatus::TableDone,
error: None,
terminal: false,
},
);
}
Err(e) => {
if e == "Cancelled" {
if let Some(journal) = history.as_ref() {
journal
.finish_table(
history_item_index,
TaskItemStatus::Cancelled,
source_row_count,
moved_row_count,
)
.await;
}
emit_terminal_progress(
&app,
history.as_ref(),
TransferProgress {
transfer_id: transfer_id.clone(),
table: table.clone(),
table_index: i,
total_tables,
rows_transferred: 0,
total_rows: None,
status: TransferStatus::Cancelled,
error: None,
terminal: true,
},
)
.await;
dbx_core::transfer::clear_cancelled(&transfer_id).await;
return;
}
if let Some(journal) = history.as_ref() {
journal
.finish_table(history_item_index, TaskItemStatus::Failed, source_row_count, moved_row_count)
.await;
}
failed_tables.push(table.clone());
emit_progress(
&app,
TransferProgress {
transfer_id: transfer_id.clone(),
table: table.clone(),
table_index: i,
total_tables,
rows_transferred: last_rows_transferred,
total_rows: last_total_rows,
status: TransferStatus::Error,
error: Some(e),
terminal: false,
},
);
}
}
}
// Add any foreign keys deferred during MySQL-family table creation now that
// every selected table exists — see transfer_table's use of
// strip_inline_foreign_key_constraint_lines for why these can't be created
// inline (a foreign key cycle has no valid CREATE TABLE order at all). The
// rename pre-pass already freed the backup constraint names, so these re-create
// the original names without colliding.
let mut failed_fk_tables: Vec<String> = Vec::new();
let mut failed_fk_count = 0usize;
for (table, alter_sql) in &pending_fk_alters {
if let Err(e) = dbx_core::transfer::execute_on_pool(&state, &target_pool_key, alter_sql).await {
log::warn!("[transfer] failed to add deferred foreign key constraint for {table}: {e}");
failed_fk_count += 1;
failed_fk_tables.push(table.clone());
}
}
if failed_fk_count > 0 {
failed_fk_tables.sort();
failed_fk_tables.dedup();
failed_tables.push(format!("{} foreign key(s) on: {}", failed_fk_count, failed_fk_tables.join(", ")));
}
// Transfer selected non-table objects (views, procedures, functions,
// triggers, sequences, events) after the per-table loop. The shared
// Core decision handles all content modes: DataOnly never
// transfers schema objects; PG→PG keeps the legacy empty-selection
// default only when structure participates in the transfer.
let mut object_outcome = dbx_core::transfer::TransferObjectOutcome::default();
match dbx_core::transfer::transfer_schema_objects(
&state,
&request,
&source_pool_key,
&target_pool_key,
source_pool_key,
target_pool_key,
history.clone(),
|progress| emit_progress(&app, progress),
)
.await
{
Ok(outcome) => {
if let Some(journal) = history.as_ref() {
journal.record_object_outcome(&outcome).await;
}
object_outcome = outcome;
}
Err(e) if e == "Cancelled" => {
if let Some(journal) = history.as_ref() {
journal.record_schema_objects_error(true).await;
}
emit_terminal_progress(
&app,
history.as_ref(),
TransferProgress {
transfer_id: transfer_id.clone(),
table: "schema objects".to_string(),
table_index: total_tables,
total_tables,
rows_transferred: 0,
total_rows: None,
status: TransferStatus::Cancelled,
error: None,
terminal: true,
},
)
.await;
dbx_core::transfer::clear_cancelled(&transfer_id).await;
return;
}
Err(e) => {
if let Some(journal) = history.as_ref() {
journal.record_schema_objects_error(false).await;
}
failed_tables.push("schema objects".to_string());
emit_progress(
&app,
TransferProgress {
transfer_id: transfer_id.clone(),
table: "schema objects".to_string(),
table_index: total_tables,
total_tables,
rows_transferred: 0,
total_rows: None,
status: TransferStatus::Error,
error: Some(e),
terminal: false,
},
);
}
}
if !object_outcome.failed.is_empty() {
failed_tables.push(format!("schema objects ({})", object_outcome.failed.len()));
}
// The rename pre-pass left one backup per rebuilt table. Drop them only now that
// every table, every deferred foreign key and every selected schema object has
// succeeded — a failure anywhere above keeps the originals recoverable under
// their backup names.
if let Some(backup_names) = backup_names.as_ref() {
if failed_tables.is_empty() {
if let Err(e) = dbx_core::transfer::drop_backup_tables(
&state,
&request,
target_db_type,
&target_pool_key,
backup_names,
&backup_drop_order,
)
.await
{
failed_tables.push(e);
}
} else {
log::warn!(
"[transfer] keeping {} backup table(s) because {} step(s) failed",
backup_names.len(),
failed_tables.len()
);
}
}
let skip_suffix = if !object_outcome.skipped.is_empty() && failed_tables.is_empty() {
format!(",跳过 {} 个已存在对象", object_outcome.skipped.len())
} else if !object_outcome.skipped.is_empty() {
format!(";跳过 {} 个已存在对象", object_outcome.skipped.len())
} else {
String::new()
};
emit_terminal_progress(
&app,
history.as_ref(),
TransferProgress {
transfer_id: transfer_id.clone(),
table: String::new(),
table_index: total_tables,
total_tables,
rows_transferred: last_rows_transferred,
total_rows: last_total_rows,
status: if failed_tables.is_empty() { TransferStatus::Done } else { TransferStatus::Error },
error: if failed_tables.is_empty() {
if skip_suffix.is_empty() {
None
} else {
Some(skip_suffix.clone())
}
} else {
Some(format!(
"{} table(s) failed: {}{}",
failed_tables.len(),
failed_tables.iter().take(5).cloned().collect::<Vec<_>>().join(", "),
skip_suffix
))
},
terminal: true,
},
)
.await;
dbx_core::transfer::clear_cancelled(&transfer_id).await;
emit_terminal_progress(&app, history.as_ref(), terminal).await;
});
Ok(())