diff --git a/apps/desktop/src/components/diff/DataCompareDialog.vue b/apps/desktop/src/components/diff/DataCompareDialog.vue index 28718749c7..b9e021ffa3 100644 --- a/apps/desktop/src/components/diff/DataCompareDialog.vue +++ b/apps/desktop/src/components/diff/DataCompareDialog.vue @@ -8,13 +8,7 @@ import { Select, SelectContent, SelectItem, SelectTrigger, SelectValue } from "@ import { useConnectionStore } from "@/stores/connectionStore"; import { useToast } from "@/composables/useToast"; import { isSchemaAware } from "@/lib/databaseCapabilities"; -import { buildTableSelectSql, qualifiedTableName } from "@/lib/tableSelectSql"; -import { - compareDataRows, - generateDataSyncSql, - generateDataSyncStatements, - type DataCompareResult, -} from "@/lib/dataCompare"; +import type { DataCompareResult } from "@/lib/dataCompare"; import * as api from "@/lib/api"; import DatabaseIcon from "@/components/icons/DatabaseIcon.vue"; import { Copy, GitCompareArrows, Loader2, Play, ArrowLeftRight } from "lucide-vue-next"; @@ -221,13 +215,6 @@ function clearResult() { targetRowCount.value = null; } -async function countTableRows(connectionId: string, database: string, schema: string, tableName: string) { - const config = store.getConfig(connectionId); - const table = qualifiedTableName({ databaseType: config?.db_type, schema, tableName }); - const result = await api.executeQuery(connectionId, database, `SELECT COUNT(*) AS row_count FROM ${table}`, schema); - return Number(result.rows[0]?.[0] ?? 0); -} - async function inferKeyColumns() { if (!sourceConnectionId.value || !sourceDatabase.value || !sourceTable.value) return; const columns = await api.getColumns( @@ -254,8 +241,6 @@ async function startCompare() { if (keyColumns.value.length === 0) await inferKeyColumns(); if (keyColumns.value.length === 0) throw new Error(t("dataCompare.noKeyColumns")); - const sourceConfig = store.getConfig(sourceConnectionId.value); - const targetConfig = store.getConfig(targetConnectionId.value); const sourceColumns = await api.getColumns( sourceConnectionId.value, sourceDatabase.value, @@ -276,48 +261,25 @@ async function startCompare() { throw new Error(t("dataCompare.missingKeyColumns", { columns: missingKeys.join(", ") })); } if (columns.length === 0) throw new Error(t("dataCompare.noCommonColumns")); - const [srcCount, tgtCount] = await Promise.all([ - countTableRows(sourceConnectionId.value, sourceDatabase.value, sourceSchema.value, sourceTable.value), - countTableRows(targetConnectionId.value, targetDatabase.value, targetSchema.value, targetTable.value), - ]); - sourceRowCount.value = srcCount; - targetRowCount.value = tgtCount; - const sourceSql = buildTableSelectSql({ - databaseType: sourceConfig?.db_type, - schema: sourceSchema.value, - tableName: sourceTable.value, - primaryKeys: keyColumns.value, - limit: rowLimitNumber.value, - }); - const targetSql = buildTableSelectSql({ - databaseType: targetConfig?.db_type, - schema: targetSchema.value, - tableName: targetTable.value, - primaryKeys: keyColumns.value, - limit: rowLimitNumber.value, - }); - const [sourceResult, targetResult] = await Promise.all([ - api.executeQuery(sourceConnectionId.value, sourceDatabase.value, sourceSql, sourceSchema.value), - api.executeQuery(targetConnectionId.value, targetDatabase.value, targetSql, targetSchema.value), - ]); - const diff = compareDataRows({ + const preparation = await api.prepareDataCompareFromTables({ + sourceConnectionId: sourceConnectionId.value, + sourceDatabase: sourceDatabase.value, + sourceSchema: sourceSchema.value, + sourceTable: sourceTable.value, + targetConnectionId: targetConnectionId.value, + targetDatabase: targetDatabase.value, + targetSchema: targetSchema.value, + targetTable: targetTable.value, columns, keyColumns: keyColumns.value, - sourceRows: sourceResult.rows, - targetRows: targetResult.rows, + rowLimit: rowLimitNumber.value, }); - result.value = diff; - const syncOptions = { - tableName: targetTable.value, - schema: targetSchema.value, - columns, - keyColumns: keyColumns.value, - diff, - databaseType: targetConfig?.db_type, - }; - syncStatements.value = generateDataSyncStatements(syncOptions); - syncSql.value = generateDataSyncSql(syncOptions); + sourceRowCount.value = preparation.sourceRowCount; + targetRowCount.value = preparation.targetRowCount; + result.value = preparation.result; + syncStatements.value = preparation.syncStatements; + syncSql.value = preparation.syncSql; } catch (e: any) { toast(e?.message || String(e), 5000); } finally { diff --git a/apps/desktop/src/components/diff/SchemaDiffDialog.vue b/apps/desktop/src/components/diff/SchemaDiffDialog.vue index 40bb5a8827..00d919358f 100644 --- a/apps/desktop/src/components/diff/SchemaDiffDialog.vue +++ b/apps/desktop/src/components/diff/SchemaDiffDialog.vue @@ -10,15 +10,8 @@ import { useConnectionStore } from "@/stores/connectionStore"; import DatabaseIcon from "@/components/icons/DatabaseIcon.vue"; import * as api from "@/lib/api"; import { isSchemaAware } from "@/lib/databaseCapabilities"; -import { - diffColumns, - diffForeignKeys, - diffIndexes, - diffTables, - diffTriggers, - generateSyncSql, - type TableDiff, -} from "@/lib/schemaDiff"; +import type { TableDiff, TableSchemaDetail } from "@/lib/schemaDiff"; +import type { TableInfo } from "@/types/database"; import { sqlMetadataRefreshTarget } from "@/lib/sqlMetadataRefresh"; import { useToast } from "@/composables/useToast"; import { Loader2, Copy, Play, GitCompareArrows, ArrowLeftRight } from "lucide-vue-next"; @@ -57,6 +50,7 @@ const executing = ref(false); const executedCount = ref(0); const executeTotal = ref(0); const syncErrors = ref<{ sql: string; error: string }[]>([]); +const syncSql = ref(""); const allSelected = computed(() => diffs.value.length > 0 && diffs.value.every((d) => d.selected)); const someSelected = computed(() => diffs.value.some((d) => d.selected) && !allSelected.value); @@ -64,15 +58,9 @@ const someSelected = computed(() => diffs.value.some((d) => d.selected) && !allS function toggleAll() { const next = !allSelected.value; diffs.value.forEach((d) => (d.selected = next)); + refreshSelectedSyncSql().catch((e) => toast(e?.message || String(e), 5000)); } -const syncSql = computed(() => { - const selected = diffs.value.filter((d) => d.selected); - if (selected.length === 0) return ""; - const srcConfig = store.getConfig(targetConnectionId.value); - return generateSyncSql(selected, srcConfig?.db_type || "mysql", targetSchema.value); -}); - const sqlConnections = computed(() => store.connections.filter((c) => !["redis", "mongodb", "elasticsearch"].includes(c.db_type)), ); @@ -173,87 +161,30 @@ async function startCompare() { if (!canCompare.value) return; step.value = "comparing"; diffs.value = []; + syncSql.value = ""; try { await store.ensureConnected(sourceConnectionId.value); await store.ensureConnected(targetConnectionId.value); + const targetConfig = store.getConfig(targetConnectionId.value); const [srcTables, tgtTables] = await Promise.all([ api.listTables(sourceConnectionId.value, sourceDatabase.value, sourceSchema.value), api.listTables(targetConnectionId.value, targetDatabase.value, targetSchema.value), ]); - const srcTableNames = srcTables.filter((t) => t.table_type !== "VIEW").map((t) => t.name); - const tgtTableNames = tgtTables.filter((t) => t.table_type !== "VIEW").map((t) => t.name); - const srcTableComments = new Map(srcTables.map((t) => [t.name, t.comment ?? null])); - const tgtTableComments = new Map(tgtTables.map((t) => [t.name, t.comment ?? null])); - const srcViewNames = srcTables.filter((t) => t.table_type === "VIEW").map((t) => t.name); - const tgtViewNames = tgtTables.filter((t) => t.table_type === "VIEW").map((t) => t.name); - const { added, removed, common } = diffTables(srcTableNames, tgtTableNames); - const { added: addedViews, removed: removedViews } = diffTables(srcViewNames, tgtViewNames); + const { sourceDetails, targetDetails } = await loadSchemaDiffDetails(srcTables, tgtTables); + const result = await api.prepareSchemaDiff({ + sourceTables: srcTables, + targetTables: tgtTables, + sourceDetails, + targetDetails, + databaseType: targetConfig?.db_type || "mysql", + targetSchema: targetSchema.value, + }); - const result: SelectableTableDiff[] = []; - - for (const name of added) { - const ddl = await api.getTableDdl(sourceConnectionId.value, sourceDatabase.value, sourceSchema.value, name); - result.push({ type: "added", objectType: "table", name, ddl, selected: true }); - } - - for (const name of removed) { - result.push({ type: "removed", objectType: "table", name, selected: true }); - } - - for (const name of addedViews) { - result.push({ type: "added", objectType: "view", name, selected: true }); - } - - for (const name of removedViews) { - result.push({ type: "removed", objectType: "view", name, selected: true }); - } - - for (const name of common) { - const [srcCols, tgtCols, srcIdx, tgtIdx, srcFks, tgtFks, srcTriggers, tgtTriggers] = await Promise.all([ - api.getColumns(sourceConnectionId.value, sourceDatabase.value, sourceSchema.value, name), - api.getColumns(targetConnectionId.value, targetDatabase.value, targetSchema.value, name), - api.listIndexes(sourceConnectionId.value, sourceDatabase.value, sourceSchema.value, name), - api.listIndexes(targetConnectionId.value, targetDatabase.value, targetSchema.value, name), - api.listForeignKeys(sourceConnectionId.value, sourceDatabase.value, sourceSchema.value, name), - api.listForeignKeys(targetConnectionId.value, targetDatabase.value, targetSchema.value, name), - api.listTriggers(sourceConnectionId.value, sourceDatabase.value, sourceSchema.value, name), - api.listTriggers(targetConnectionId.value, targetDatabase.value, targetSchema.value, name), - ]); - - const colDiffs = diffColumns(srcCols, tgtCols); - const idxDiffs = diffIndexes(srcIdx, tgtIdx); - const fkDiffs = diffForeignKeys(srcFks, tgtFks); - const triggerDiffs = diffTriggers(srcTriggers, tgtTriggers); - const srcComment = srcTableComments.get(name) ?? null; - const tgtComment = tgtTableComments.get(name) ?? null; - const commentChanged = (srcComment ?? "") !== (tgtComment ?? ""); - - if ( - colDiffs.length > 0 || - idxDiffs.length > 0 || - fkDiffs.length > 0 || - triggerDiffs.length > 0 || - commentChanged - ) { - result.push({ - type: "modified", - objectType: "table", - name, - columns: colDiffs.length > 0 ? colDiffs : undefined, - indexes: idxDiffs.length > 0 ? idxDiffs : undefined, - foreignKeys: fkDiffs.length > 0 ? fkDiffs : undefined, - triggers: triggerDiffs.length > 0 ? triggerDiffs : undefined, - sourceTableComment: commentChanged ? srcComment : undefined, - targetTableComment: commentChanged ? tgtComment : undefined, - selected: true, - }); - } - } - - diffs.value = result; + diffs.value = result.diffs.map((diff) => ({ ...diff, selected: true })); + syncSql.value = result.syncSql; step.value = "result"; } catch (e: any) { toast(e?.message || String(e), 5000); @@ -261,6 +192,57 @@ async function startCompare() { } } +async function loadSchemaDiffDetails(sourceTables: TableInfo[], targetTables: TableInfo[]) { + const sourceTableNames = sourceTables.filter((table) => table.table_type !== "VIEW").map((table) => table.name); + const targetTableNames = targetTables.filter((table) => table.table_type !== "VIEW").map((table) => table.name); + const targetTableSet = new Set(targetTableNames); + const sourceTableSet = new Set(sourceTableNames); + const sourceDetails: TableSchemaDetail[] = []; + const targetDetails: TableSchemaDetail[] = []; + + for (const name of sourceTableNames) { + if (targetTableSet.has(name)) { + const [columns, indexes, foreignKeys, triggers] = await Promise.all([ + api.getColumns(sourceConnectionId.value, sourceDatabase.value, sourceSchema.value, name), + api.listIndexes(sourceConnectionId.value, sourceDatabase.value, sourceSchema.value, name), + api.listForeignKeys(sourceConnectionId.value, sourceDatabase.value, sourceSchema.value, name), + api.listTriggers(sourceConnectionId.value, sourceDatabase.value, sourceSchema.value, name), + ]); + sourceDetails.push({ name, columns, indexes, foreignKeys, triggers }); + } else { + const ddl = await api.getTableDdl(sourceConnectionId.value, sourceDatabase.value, sourceSchema.value, name); + sourceDetails.push({ name, ddl }); + } + } + + for (const name of targetTableNames) { + if (!sourceTableSet.has(name)) continue; + const [columns, indexes, foreignKeys, triggers] = await Promise.all([ + api.getColumns(targetConnectionId.value, targetDatabase.value, targetSchema.value, name), + api.listIndexes(targetConnectionId.value, targetDatabase.value, targetSchema.value, name), + api.listForeignKeys(targetConnectionId.value, targetDatabase.value, targetSchema.value, name), + api.listTriggers(targetConnectionId.value, targetDatabase.value, targetSchema.value, name), + ]); + targetDetails.push({ name, columns, indexes, foreignKeys, triggers }); + } + + return { sourceDetails, targetDetails }; +} + +async function refreshSelectedSyncSql() { + const selected = diffs.value.filter((diff) => diff.selected); + if (selected.length === 0) { + syncSql.value = ""; + return; + } + const targetConfig = store.getConfig(targetConnectionId.value); + syncSql.value = await api.generateSchemaSyncSql(selected, targetConfig?.db_type || "mysql", targetSchema.value); +} + +function onDiffSelectionChange() { + refreshSelectedSyncSql().catch((e) => toast(e?.message || String(e), 5000)); +} + async function executeSql() { const sql = syncSql.value.trim(); if (!sql || executing.value) return; @@ -324,6 +306,7 @@ function diffLabel(type: string) { function resetResult() { step.value = "select"; diffs.value = []; + syncSql.value = ""; syncErrors.value = []; executedCount.value = 0; executeTotal.value = 0; @@ -543,7 +526,12 @@ watch( - + {{ d.name }} diff --git a/apps/desktop/src/lib/api.ts b/apps/desktop/src/lib/api.ts index 2843fa9eb8..72d033a308 100644 --- a/apps/desktop/src/lib/api.ts +++ b/apps/desktop/src/lib/api.ts @@ -77,6 +77,8 @@ export const listIndexes = forward("listIndexes"); export const listForeignKeys = forward("listForeignKeys"); export const listTriggers = forward("listTriggers"); export const getTableDdl = forward("getTableDdl"); +export const prepareSchemaDiff = forward("prepareSchemaDiff"); +export const generateSchemaSyncSql = forward("generateSchemaSyncSql"); // Query export const executeQuery = forward("executeQuery"); @@ -89,6 +91,8 @@ export const closeQuerySession = forward("closeQuerySession"); export const analyzeSqlReferences = forward("analyzeSqlReferences"); export const analyzeEditableQueryEditability = forward("analyzeEditableQueryEditability"); export const prepareDataGridSave = forward("prepareDataGridSave"); +export const prepareDataCompare = forward("prepareDataCompare"); +export const prepareDataCompareFromTables = forward("prepareDataCompareFromTables"); // AI export const aiComplete = forward("aiComplete"); diff --git a/apps/desktop/src/lib/dataCompare.ts b/apps/desktop/src/lib/dataCompare.ts index e5be7ac37e..0c943f19cb 100644 --- a/apps/desktop/src/lib/dataCompare.ts +++ b/apps/desktop/src/lib/dataCompare.ts @@ -1,6 +1,4 @@ import type { DatabaseType, QueryResult } from "@/types/database"; -import { quoteTableIdentifier } from "./tableSelectSql"; -import { formatGridSqlLiteral } from "./dataGridSql"; export type DataCompareCellValue = QueryResult["rows"][number][number]; @@ -30,152 +28,39 @@ export interface DataCompareResult { modified: DataCompareModifiedRow[]; } -export interface CompareDataRowsOptions { - columns: readonly string[]; - keyColumns: readonly string[]; - sourceRows: readonly (readonly DataCompareCellValue[])[]; - targetRows: readonly (readonly DataCompareCellValue[])[]; -} - -export interface GenerateDataSyncSqlOptions { +export interface DataComparePreparationOptions { tableName: string; schema?: string; - columns: readonly string[]; - keyColumns: readonly string[]; - diff: DataCompareResult; + columns: string[]; + keyColumns: string[]; + sourceRows: DataCompareCellValue[][]; + targetRows: DataCompareCellValue[][]; databaseType?: DatabaseType; } -function rowObject( - columns: readonly string[], - row: readonly DataCompareCellValue[], -): Record { - const item: Record = {}; - columns.forEach((column, index) => { - item[column] = row[index] ?? null; - }); - return item; +export interface DataComparePreparation { + result: DataCompareResult; + syncStatements: string[]; + syncSql: string; } -function keyFor(row: Record, keyColumns: readonly string[]): string { - return keyColumns.map((column) => JSON.stringify(row[column] ?? null)).join("\u001f"); +export interface DataCompareFromTablesOptions { + sourceConnectionId: string; + sourceDatabase: string; + sourceSchema: string; + sourceTable: string; + targetConnectionId: string; + targetDatabase: string; + targetSchema: string; + targetTable: string; + columns: string[]; + keyColumns: string[]; + rowLimit: number; } -function keyValues(row: Record, keyColumns: readonly string[]) { - const values: Record = {}; - keyColumns.forEach((column) => { - values[column] = row[column] ?? null; - }); - return values; -} - -export function compareDataRows(options: CompareDataRowsOptions): DataCompareResult { - if (options.keyColumns.length === 0) { - throw new Error("At least one key column is required for data comparison"); - } - - const source = new Map>(); - const target = new Map>(); - options.sourceRows.forEach((row) => { - const item = rowObject(options.columns, row); - const key = keyFor(item, options.keyColumns); - if (source.has(key)) throw new Error(`Duplicate source key: ${key}`); - source.set(key, item); - }); - options.targetRows.forEach((row) => { - const item = rowObject(options.columns, row); - const key = keyFor(item, options.keyColumns); - if (target.has(key)) throw new Error(`Duplicate target key: ${key}`); - target.set(key, item); - }); - - const added: DataCompareRow[] = []; - const removed: DataCompareRow[] = []; - const modified: DataCompareModifiedRow[] = []; - - for (const [key, sourceValues] of source) { - const targetValues = target.get(key); - if (!targetValues) { - added.push({ key, keyValues: keyValues(sourceValues, options.keyColumns), values: sourceValues }); - continue; - } - - const changes = options.columns - .filter((column) => !options.keyColumns.includes(column)) - .filter((column) => sourceValues[column] !== targetValues[column]) - .map((column) => ({ column, source: sourceValues[column] ?? null, target: targetValues[column] ?? null })); - - if (changes.length > 0) { - modified.push({ - key, - keyValues: keyValues(sourceValues, options.keyColumns), - sourceValues, - targetValues, - changes, - }); - } - } - - for (const [key, targetValues] of target) { - if (!source.has(key)) { - removed.push({ key, keyValues: keyValues(targetValues, options.keyColumns), values: targetValues }); - } - } - - return { added, removed, modified }; -} - -function qualifiedTableName(schema: string | undefined, tableName: string, databaseType?: DatabaseType): string { - const table = quoteTableIdentifier(databaseType, tableName); - return schema ? `${quoteTableIdentifier(databaseType, schema)}.${table}` : table; -} - -function whereByKey( - keyValues: Record, - keyColumns: readonly string[], - databaseType?: DatabaseType, -): string { - return keyColumns - .map( - (column) => - `${quoteTableIdentifier(databaseType, column)} = ${formatGridSqlLiteral(keyValues[column], databaseType)}`, - ) - .join(" AND "); -} - -export function generateDataSyncStatements(options: GenerateDataSyncSqlOptions): string[] { - const table = qualifiedTableName(options.schema, options.tableName, options.databaseType); - const statements: string[] = []; - - for (const row of options.diff.added) { - const columns = options.columns.map((column) => quoteTableIdentifier(options.databaseType, column)).join(", "); - const values = options.columns - .map((column) => formatGridSqlLiteral(row.values[column], options.databaseType)) - .join(", "); - statements.push(`INSERT INTO ${table} (${columns}) VALUES (${values});`); - } - - for (const row of options.diff.modified) { - const assignments = row.changes - .map( - (change) => - `${quoteTableIdentifier(options.databaseType, change.column)} = ${formatGridSqlLiteral(change.source, options.databaseType)}`, - ) - .join(", "); - statements.push( - `UPDATE ${table} SET ${assignments} WHERE ${whereByKey(row.keyValues, options.keyColumns, options.databaseType)};`, - ); - } - - for (const row of options.diff.removed) { - statements.push( - `DELETE FROM ${table} WHERE ${whereByKey(row.keyValues, options.keyColumns, options.databaseType)};`, - ); - } - - return statements; -} - -export function generateDataSyncSql(options: GenerateDataSyncSqlOptions): string { - return generateDataSyncStatements(options).join("\n"); +export interface DataCompareFromTablesPreparation extends DataComparePreparation { + sourceRowCount: number; + targetRowCount: number; + sourceTruncated: boolean; + targetTruncated: boolean; } diff --git a/apps/desktop/src/lib/http.ts b/apps/desktop/src/lib/http.ts index 9217ccb085..f514a7c757 100644 --- a/apps/desktop/src/lib/http.ts +++ b/apps/desktop/src/lib/http.ts @@ -11,6 +11,7 @@ import type { TriggerInfo, QueryResult, SqlReferenceAnalysis, + DatabaseType, InstalledPlugin, JdbcDriverInfo, JdbcPluginStatus, @@ -51,6 +52,13 @@ import type { } from "./tauri"; import type { QueryEditability } from "@/lib/sqlAnalysis"; import type { DataGridSaveStatementOptions } from "@/lib/dataGridSql"; +import type { + DataCompareFromTablesOptions, + DataCompareFromTablesPreparation, + DataComparePreparation, + DataComparePreparationOptions, +} from "@/lib/dataCompare"; +import type { SchemaDiffPreparation, SchemaDiffPreparationOptions, TableDiff } from "@/lib/schemaDiff"; import type { DataGridSavePreparation } from "./tauri"; // --------------------------------------------------------------------------- @@ -342,6 +350,18 @@ export async function getTableDdl( return get(`/api/schema/ddl?${qs({ connection_id: connectionId, database, schema, table })}`); } +export async function prepareSchemaDiff(options: SchemaDiffPreparationOptions): Promise { + return post("/api/schema-diff/prepare", options); +} + +export async function generateSchemaSyncSql( + diffs: TableDiff[], + databaseType: DatabaseType, + targetSchema?: string, +): Promise { + return post("/api/schema-diff/generate-sync-sql", { diffs, databaseType, targetSchema }); +} + // --------------------------------------------------------------------------- // Query // --------------------------------------------------------------------------- @@ -415,6 +435,16 @@ export async function prepareDataGridSave(options: DataGridSaveStatementOptions) return post("/api/query/prepare-data-grid-save", { options }); } +export async function prepareDataCompare(options: DataComparePreparationOptions): Promise { + return post("/api/data-compare/prepare", options); +} + +export async function prepareDataCompareFromTables( + options: DataCompareFromTablesOptions, +): Promise { + return post("/api/data-compare/prepare-from-tables", options); +} + // --------------------------------------------------------------------------- // AI // --------------------------------------------------------------------------- diff --git a/apps/desktop/src/lib/schemaDiff.ts b/apps/desktop/src/lib/schemaDiff.ts index bcd8528cea..a2ad576dd5 100644 --- a/apps/desktop/src/lib/schemaDiff.ts +++ b/apps/desktop/src/lib/schemaDiff.ts @@ -1,4 +1,4 @@ -import type { ColumnInfo, IndexInfo, ForeignKeyInfo, TriggerInfo, DatabaseType } from "@/types/database"; +import type { ColumnInfo, IndexInfo, ForeignKeyInfo, TriggerInfo, DatabaseType, TableInfo } from "@/types/database"; export interface ColumnDiff { type: "added" | "removed" | "modified"; @@ -45,407 +45,25 @@ export interface TableDiff { targetTableComment?: string | null; } -export function diffColumns(source: ColumnInfo[], target: ColumnInfo[]): ColumnDiff[] { - const diffs: ColumnDiff[] = []; - const targetMap = new Map(target.map((c) => [c.name, c])); - const sourceMap = new Map(source.map((c) => [c.name, c])); - - for (const sc of source) { - const tc = targetMap.get(sc.name); - if (!tc) { - diffs.push({ type: "added", name: sc.name, source: sc }); - } else { - const changes: string[] = []; - if (sc.data_type.toLowerCase() !== tc.data_type.toLowerCase()) { - changes.push(`type: ${tc.data_type} → ${sc.data_type}`); - } - if (sc.is_nullable !== tc.is_nullable) { - changes.push(`nullable: ${tc.is_nullable ? "YES" : "NO"} → ${sc.is_nullable ? "YES" : "NO"}`); - } - if ((sc.column_default ?? "") !== (tc.column_default ?? "")) { - changes.push(`default: ${tc.column_default ?? "NULL"} → ${sc.column_default ?? "NULL"}`); - } - if ((sc.comment ?? "") !== (tc.comment ?? "")) { - changes.push(`comment: ${tc.comment ?? ""} → ${sc.comment ?? ""}`); - } - if (changes.length > 0) { - diffs.push({ type: "modified", name: sc.name, source: sc, target: tc, changes }); - } - } - } - - for (const tc of target) { - if (!sourceMap.has(tc.name)) { - diffs.push({ type: "removed", name: tc.name, target: tc }); - } - } - - return diffs; +export interface TableSchemaDetail { + name: string; + columns?: ColumnInfo[]; + indexes?: IndexInfo[]; + foreignKeys?: ForeignKeyInfo[]; + triggers?: TriggerInfo[]; + ddl?: string; } -export function diffIndexes(source: IndexInfo[], target: IndexInfo[]): IndexDiff[] { - const diffs: IndexDiff[] = []; - const targetMap = new Map(target.map((i) => [i.name, i])); - const sourceMap = new Map(source.map((i) => [i.name, i])); - - for (const si of source) { - if (si.is_primary) continue; - const ti = targetMap.get(si.name); - if (!ti) { - diffs.push({ type: "added", name: si.name, source: si }); - continue; - } - - const changes: string[] = []; - if (si.is_unique !== ti.is_unique) { - changes.push(`unique: ${ti.is_unique ? "YES" : "NO"} → ${si.is_unique ? "YES" : "NO"}`); - } - if (si.columns.join(",") !== ti.columns.join(",")) { - changes.push(`columns: ${ti.columns.join(", ")} → ${si.columns.join(", ")}`); - } - if ((si.index_type ?? "") !== (ti.index_type ?? "")) { - changes.push(`type: ${ti.index_type ?? "default"} → ${si.index_type ?? "default"}`); - } - if ((si.filter ?? "") !== (ti.filter ?? "")) { - changes.push(`filter: ${ti.filter ?? "none"} → ${si.filter ?? "none"}`); - } - if ((si.included_columns ?? []).join(",") !== (ti.included_columns ?? []).join(",")) { - changes.push( - `include: ${(ti.included_columns ?? []).join(", ") || "none"} → ${(si.included_columns ?? []).join(", ") || "none"}`, - ); - } - - if (changes.length > 0) { - diffs.push({ type: "modified", name: si.name, source: si, target: ti, changes }); - } - } - - for (const ti of target) { - if (ti.is_primary) continue; - if (!sourceMap.has(ti.name)) { - diffs.push({ type: "removed", name: ti.name, target: ti }); - } - } - - return diffs; +export interface SchemaDiffPreparationOptions { + sourceTables: TableInfo[]; + targetTables: TableInfo[]; + sourceDetails: TableSchemaDetail[]; + targetDetails: TableSchemaDetail[]; + databaseType: DatabaseType; + targetSchema?: string; } -export function diffForeignKeys(source: ForeignKeyInfo[], target: ForeignKeyInfo[]): ForeignKeyDiff[] { - const diffs: ForeignKeyDiff[] = []; - const targetMap = new Map(target.map((fk) => [fk.name, fk])); - const sourceMap = new Map(source.map((fk) => [fk.name, fk])); - - for (const sfk of source) { - const tfk = targetMap.get(sfk.name); - if (!tfk) { - diffs.push({ type: "added", name: sfk.name, source: sfk }); - continue; - } - - const changes: string[] = []; - if (sfk.column !== tfk.column) changes.push(`column: ${tfk.column} → ${sfk.column}`); - if (sfk.ref_table !== tfk.ref_table) changes.push(`ref table: ${tfk.ref_table} → ${sfk.ref_table}`); - if (sfk.ref_column !== tfk.ref_column) changes.push(`ref column: ${tfk.ref_column} → ${sfk.ref_column}`); - - if (changes.length > 0) { - diffs.push({ type: "modified", name: sfk.name, source: sfk, target: tfk, changes }); - } - } - - for (const tfk of target) { - if (!sourceMap.has(tfk.name)) { - diffs.push({ type: "removed", name: tfk.name, target: tfk }); - } - } - - return diffs; -} - -export function diffTriggers(source: TriggerInfo[], target: TriggerInfo[]): TriggerDiff[] { - const diffs: TriggerDiff[] = []; - const targetMap = new Map(target.map((trigger) => [trigger.name, trigger])); - const sourceMap = new Map(source.map((trigger) => [trigger.name, trigger])); - - for (const sourceTrigger of source) { - const targetTrigger = targetMap.get(sourceTrigger.name); - if (!targetTrigger) { - diffs.push({ type: "added", name: sourceTrigger.name, source: sourceTrigger }); - continue; - } - - const changes: string[] = []; - if (sourceTrigger.event !== targetTrigger.event) { - changes.push(`event: ${targetTrigger.event} → ${sourceTrigger.event}`); - } - if (sourceTrigger.timing !== targetTrigger.timing) { - changes.push(`timing: ${targetTrigger.timing} → ${sourceTrigger.timing}`); - } - - if (changes.length > 0) { - diffs.push({ type: "modified", name: sourceTrigger.name, source: sourceTrigger, target: targetTrigger, changes }); - } - } - - for (const targetTrigger of target) { - if (!sourceMap.has(targetTrigger.name)) { - diffs.push({ type: "removed", name: targetTrigger.name, target: targetTrigger }); - } - } - - return diffs; -} - -export function diffTables( - sourceTables: string[], - targetTables: string[], -): { added: string[]; removed: string[]; common: string[] } { - const targetSet = new Set(targetTables); - const sourceSet = new Set(sourceTables); - return { - added: sourceTables.filter((t) => !targetSet.has(t)), - removed: targetTables.filter((t) => !sourceSet.has(t)), - common: sourceTables.filter((t) => targetSet.has(t)), - }; -} - -function quoteId(name: string, dbType: DatabaseType): string { - if (dbType === "mysql" || dbType === "doris" || dbType === "starrocks") { - return `\`${name.replace(/`/g, "``")}\``; - } - return `"${name.replace(/"/g, '""')}"`; -} - -function columnDef(col: ColumnInfo, dbType: DatabaseType): string { - let def = `${quoteId(col.name, dbType)} ${col.data_type}`; - if (!col.is_nullable) def += " NOT NULL"; - if (col.column_default !== null && col.column_default !== undefined) { - def += ` DEFAULT ${col.column_default}`; - } - return def; -} - -function qualifiedName(name: string, dbType: DatabaseType, schema?: string): string { - return schema ? `${quoteId(schema, dbType)}.${quoteId(name, dbType)}` : quoteId(name, dbType); -} - -function dropIndexSql(tableName: string, indexName: string, dbType: DatabaseType, schema?: string): string { - const qt = qualifiedName(tableName, dbType, schema); - const qi = qualifiedName(indexName, dbType, schema); - if (dbType === "mysql" || dbType === "doris" || dbType === "starrocks") { - return `DROP INDEX ${quoteId(indexName, dbType)} ON ${qt};`; - } - return `DROP INDEX IF EXISTS ${qi};`; -} - -function createIndexSql(tableName: string, idx: IndexInfo, dbType: DatabaseType, schema?: string): string { - const qt = qualifiedName(tableName, dbType, schema); - const cols = idx.columns.map((c) => quoteId(c, dbType)).join(", "); - const unique = idx.is_unique ? "UNIQUE " : ""; - const idxType = idx.index_type ?? ""; - const usingClause = idxType && dbType === "postgres" ? ` USING ${idxType}` : ""; - const typePrefix = idxType && dbType === "sqlserver" ? `${idxType} ` : ""; - const incCols = idx.included_columns ?? []; - const includeClause = - incCols.length > 0 && (dbType === "postgres" || dbType === "sqlserver") - ? ` INCLUDE (${incCols.map((c) => quoteId(c, dbType)).join(", ")})` - : ""; - const supportsWhere = dbType === "postgres" || dbType === "sqlserver" || dbType === "sqlite"; - const filter = idx.filter && supportsWhere ? ` WHERE ${idx.filter}` : ""; - return `CREATE ${unique}${typePrefix}INDEX ${quoteId(idx.name, dbType)} ON ${qt}${usingClause} (${cols})${includeClause}${filter};`; -} - -function dropForeignKeySql(tableName: string, fkName: string, dbType: DatabaseType, schema?: string): string { - const qt = qualifiedName(tableName, dbType, schema); - const qf = quoteId(fkName, dbType); - if (dbType === "mysql" || dbType === "doris" || dbType === "starrocks") { - return `ALTER TABLE ${qt} DROP FOREIGN KEY ${qf};`; - } - return `ALTER TABLE ${qt} DROP CONSTRAINT ${qf};`; -} - -function addForeignKeySql(tableName: string, fk: ForeignKeyInfo, dbType: DatabaseType, schema?: string): string { - const qt = qualifiedName(tableName, dbType, schema); - return `ALTER TABLE ${qt} ADD CONSTRAINT ${quoteId(fk.name, dbType)} FOREIGN KEY (${quoteId(fk.column, dbType)}) REFERENCES ${quoteId(fk.ref_table, dbType)} (${quoteId(fk.ref_column, dbType)});`; -} - -function dropObjectSql(diff: TableDiff, dbType: DatabaseType, schema?: string): string { - const objectType = diff.objectType === "view" ? "VIEW" : "TABLE"; - return `DROP ${objectType} IF EXISTS ${qualifiedName(diff.name, dbType, schema)};`; -} - -function commentLiteral(comment: string): string { - return `'${comment.replace(/'/g, "''")}'`; -} - -function columnCommentSql( - tableName: string, - colName: string, - comment: string, - dbType: DatabaseType, - schema?: string, -): string { - const isMySQL = dbType === "mysql" || dbType === "doris" || dbType === "starrocks"; - if (isMySQL) { - return `-- Column comment for ${colName}: use ALTER TABLE ... MODIFY COLUMN to set comment in MySQL`; - } - const qt = qualifiedName(tableName, dbType, schema); - return `COMMENT ON COLUMN ${qt}.${quoteId(colName, dbType)} IS ${commentLiteral(comment)};`; -} - -function tableCommentSql(tableName: string, comment: string, dbType: DatabaseType, schema?: string): string { - const isMySQL = dbType === "mysql" || dbType === "doris" || dbType === "starrocks"; - const qt = qualifiedName(tableName, dbType, schema); - if (isMySQL) { - return `ALTER TABLE ${qt} COMMENT = ${commentLiteral(comment)};`; - } - return `COMMENT ON TABLE ${qt} IS ${commentLiteral(comment)};`; -} - -export function generateSyncSql(diffs: TableDiff[], dbType: DatabaseType, schema?: string): string { - const lines: string[] = []; - const isMySQL = dbType === "mysql" || dbType === "doris" || dbType === "starrocks"; - - for (const diff of diffs) { - const qt = qualifiedName(diff.name, dbType, schema); - - if (diff.type === "added" && diff.ddl) { - lines.push(`-- Create ${diff.objectType ?? "table"}: ${diff.name}`); - lines.push(diff.ddl + ";"); - lines.push(""); - continue; - } - - if (diff.type === "added" && diff.objectType === "view") { - lines.push(`-- View exists only in source: ${diff.name}`); - lines.push("-- Source view definition is not available from this driver yet."); - lines.push(""); - continue; - } - - if (diff.type === "removed") { - lines.push(`-- Drop ${diff.objectType ?? "table"}: ${diff.name}`); - lines.push(dropObjectSql(diff, dbType, schema)); - lines.push(""); - continue; - } - - if (diff.type === "modified") { - const parts: string[] = []; - - if (diff.foreignKeys) { - for (const fk of diff.foreignKeys) { - if (fk.type === "removed" || fk.type === "modified") { - lines.push(dropForeignKeySql(diff.name, fk.name, dbType, schema)); - } - } - } - - if (diff.columns) { - for (const col of diff.columns) { - if (col.type === "added" && col.source) { - parts.push(` ADD COLUMN ${columnDef(col.source, dbType)}`); - } else if (col.type === "removed") { - parts.push(` DROP COLUMN ${quoteId(col.name, dbType)}`); - } else if (col.type === "modified" && col.source) { - if (isMySQL) { - parts.push(` MODIFY COLUMN ${columnDef(col.source, dbType)}`); - } else { - const qn = quoteId(col.name, dbType); - if (col.changes?.some((c) => c.startsWith("type:"))) { - parts.push(` ALTER COLUMN ${qn} TYPE ${col.source.data_type}`); - } - if (col.changes?.some((c) => c.startsWith("nullable:"))) { - parts.push( - col.source.is_nullable ? ` ALTER COLUMN ${qn} DROP NOT NULL` : ` ALTER COLUMN ${qn} SET NOT NULL`, - ); - } - if (col.changes?.some((c) => c.startsWith("default:"))) { - if (col.source.column_default !== null && col.source.column_default !== undefined) { - parts.push(` ALTER COLUMN ${qn} SET DEFAULT ${col.source.column_default}`); - } else { - parts.push(` ALTER COLUMN ${qn} DROP DEFAULT`); - } - } - } - } - } - } - - if (parts.length > 0) { - lines.push(`-- Alter table: ${diff.name}`); - if (isMySQL) { - lines.push(`ALTER TABLE ${qt}`); - lines.push(parts.join(",\n") + ";"); - } else { - for (const part of parts) { - lines.push(`ALTER TABLE ${qt}${part};`); - } - } - lines.push(""); - } - - if (diff.columns) { - for (const col of diff.columns) { - if (col.source && col.changes?.some((c) => c.startsWith("comment:"))) { - lines.push(columnCommentSql(diff.name, col.name, col.source.comment ?? "", dbType, schema)); - } - if (col.type === "added" && col.source?.comment) { - lines.push(columnCommentSql(diff.name, col.name, col.source.comment, dbType, schema)); - } - } - } - - if (diff.sourceTableComment !== undefined && diff.sourceTableComment !== diff.targetTableComment) { - lines.push(tableCommentSql(diff.name, diff.sourceTableComment ?? "", dbType, schema)); - } - - if (diff.indexes) { - for (const idx of diff.indexes) { - if (idx.type === "added" && idx.source) { - lines.push(createIndexSql(diff.name, idx.source, dbType, schema)); - } else if (idx.type === "removed") { - lines.push(dropIndexSql(diff.name, idx.name, dbType, schema)); - } else if (idx.type === "modified" && idx.source) { - lines.push(dropIndexSql(diff.name, idx.name, dbType, schema)); - lines.push(createIndexSql(diff.name, idx.source, dbType, schema)); - } - } - } - - if (diff.foreignKeys) { - for (const fk of diff.foreignKeys) { - if (fk.type === "added" && fk.source) { - lines.push(addForeignKeySql(diff.name, fk.source, dbType, schema)); - } else if (fk.type === "modified" && fk.source) { - lines.push(addForeignKeySql(diff.name, fk.source, dbType, schema)); - } - } - } - - if (diff.triggers) { - for (const trigger of diff.triggers) { - lines.push( - `-- Trigger ${trigger.type}: ${trigger.name} on ${diff.name}; review trigger definition manually.`, - ); - } - } - - if (diff.indexes || diff.foreignKeys || diff.triggers) { - if ( - (diff.indexes?.length ?? 0) > 0 || - (diff.foreignKeys?.length ?? 0) > 0 || - (diff.triggers?.length ?? 0) > 0 - ) { - lines.push(""); - } - } - - if (dbType === "sqlite" && diff.foreignKeys?.length) { - lines.push(`-- SQLite foreign key synchronization may require table rebuild for: ${diff.name}`); - lines.push(""); - } - } - } - - return lines.join("\n").trim(); +export interface SchemaDiffPreparation { + diffs: TableDiff[]; + syncSql: string; } diff --git a/apps/desktop/src/lib/tauri.ts b/apps/desktop/src/lib/tauri.ts index 0a8f7d26f1..28c02a02ff 100644 --- a/apps/desktop/src/lib/tauri.ts +++ b/apps/desktop/src/lib/tauri.ts @@ -13,6 +13,7 @@ import type { TriggerInfo, QueryResult, SqlReferenceAnalysis, + DatabaseType, InstalledPlugin, JdbcDriverInfo, JdbcPluginStatus, @@ -23,6 +24,13 @@ import type { import type { AiConfig } from "@/stores/settingsStore"; import type { QueryEditability } from "@/lib/sqlAnalysis"; import type { DataGridSaveStatementOptions } from "@/lib/dataGridSql"; +import type { + DataCompareFromTablesOptions, + DataCompareFromTablesPreparation, + DataComparePreparation, + DataComparePreparationOptions, +} from "@/lib/dataCompare"; +import type { SchemaDiffPreparation, SchemaDiffPreparationOptions, TableDiff } from "@/lib/schemaDiff"; export interface AgentDriverInfo { db_type: string; @@ -329,6 +337,16 @@ export async function prepareDataGridSave(options: DataGridSaveStatementOptions) return invoke("prepare_data_grid_save", { options }); } +export async function prepareDataCompare(options: DataComparePreparationOptions): Promise { + return invoke("prepare_data_compare", { options }); +} + +export async function prepareDataCompareFromTables( + options: DataCompareFromTablesOptions, +): Promise { + return invoke("prepare_data_compare_from_tables", { options }); +} + export async function listIndexes( connectionId: string, database: string, @@ -365,6 +383,18 @@ export async function getTableDdl( return invoke("get_table_ddl", { connectionId, database, schema, table }); } +export async function prepareSchemaDiff(options: SchemaDiffPreparationOptions): Promise { + return invoke("prepare_schema_diff", { options }); +} + +export async function generateSchemaSyncSql( + diffs: TableDiff[], + databaseType: DatabaseType, + targetSchema?: string, +): Promise { + return invoke("generate_schema_sync_sql", { diffs, databaseType, targetSchema }); +} + export async function saveConnections(configs: ConnectionConfig[]): Promise { return invoke("save_connections", { configs }); } diff --git a/crates/dbx-core/src/data_compare.rs b/crates/dbx-core/src/data_compare.rs new file mode 100644 index 0000000000..4b118ba2cd --- /dev/null +++ b/crates/dbx-core/src/data_compare.rs @@ -0,0 +1,639 @@ +use std::collections::HashMap; + +use serde::{Deserialize, Serialize}; +use serde_json::Value; + +use crate::connection::AppState; +use crate::models::connection::DatabaseType; +use crate::query::{execute_sql_statement_with_options, QueryExecutionOptions}; +use crate::sql_dialect::{ + build_count_table_sql, build_table_select_sql, qualified_table_name, quote_table_identifier, TableSelectSqlOptions, +}; + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CompareDataRowsOptions { + pub columns: Vec, + pub key_columns: Vec, + #[serde(default)] + pub source_rows: Vec>, + #[serde(default)] + pub target_rows: Vec>, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct DataComparePreparationOptions { + pub table_name: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub schema: Option, + pub columns: Vec, + pub key_columns: Vec, + #[serde(default)] + pub source_rows: Vec>, + #[serde(default)] + pub target_rows: Vec>, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub database_type: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct DataCompareFromTablesOptions { + pub source_connection_id: String, + pub source_database: String, + pub source_schema: String, + pub source_table: String, + pub target_connection_id: String, + pub target_database: String, + pub target_schema: String, + pub target_table: String, + pub columns: Vec, + pub key_columns: Vec, + pub row_limit: usize, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct DataCompareChangedCell { + pub column: String, + pub source: Value, + pub target: Value, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct DataCompareRow { + pub key: String, + pub key_values: HashMap, + pub values: HashMap, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct DataCompareModifiedRow { + pub key: String, + pub key_values: HashMap, + pub source_values: HashMap, + pub target_values: HashMap, + pub changes: Vec, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct DataCompareResult { + pub added: Vec, + pub removed: Vec, + pub modified: Vec, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct DataComparePreparation { + pub result: DataCompareResult, + pub sync_statements: Vec, + pub sync_sql: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct DataCompareFromTablesPreparation { + pub result: DataCompareResult, + pub sync_statements: Vec, + pub sync_sql: String, + pub source_row_count: u64, + pub target_row_count: u64, + pub source_truncated: bool, + pub target_truncated: bool, +} + +pub fn prepare_data_compare(options: DataComparePreparationOptions) -> Result { + let result = compare_data_rows(CompareDataRowsOptions { + columns: options.columns.clone(), + key_columns: options.key_columns.clone(), + source_rows: options.source_rows, + target_rows: options.target_rows, + })?; + let sync_statements = generate_data_sync_statements(&GenerateDataSyncSqlOptions { + table_name: options.table_name, + schema: options.schema, + columns: options.columns, + key_columns: options.key_columns, + diff: result.clone(), + database_type: options.database_type, + }); + let sync_sql = sync_statements.join("\n"); + Ok(DataComparePreparation { result, sync_statements, sync_sql }) +} + +pub async fn prepare_data_compare_from_tables( + state: &AppState, + options: DataCompareFromTablesOptions, +) -> Result { + 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?; + let row_limit = options.row_limit.max(1); + + let source_count_sql = + build_count_table_sql(Some(source_database_type), Some(&options.source_schema), &options.source_table); + let target_count_sql = + build_count_table_sql(Some(target_database_type), Some(&options.target_schema), &options.target_table); + + let source_count_result = execute_sql_statement_with_options( + state, + &options.source_connection_id, + &options.source_database, + &source_count_sql, + Some(&options.source_schema), + None, + QueryExecutionOptions { max_rows: Some(1), ..Default::default() }, + ) + .await?; + let target_count_result = execute_sql_statement_with_options( + state, + &options.target_connection_id, + &options.target_database, + &target_count_sql, + Some(&options.target_schema), + None, + QueryExecutionOptions { max_rows: Some(1), ..Default::default() }, + ) + .await?; + let source_row_count = first_count(&source_count_result.rows)?; + let target_row_count = first_count(&target_count_result.rows)?; + + let source_sql = build_data_compare_select_sql( + source_database_type, + &options.source_schema, + &options.source_table, + &options.columns, + &options.key_columns, + row_limit, + ); + let target_sql = build_data_compare_select_sql( + target_database_type, + &options.target_schema, + &options.target_table, + &options.columns, + &options.key_columns, + row_limit, + ); + + let source_result = execute_sql_statement_with_options( + state, + &options.source_connection_id, + &options.source_database, + &source_sql, + Some(&options.source_schema), + None, + QueryExecutionOptions { max_rows: Some(row_limit), ..Default::default() }, + ) + .await?; + let target_result = execute_sql_statement_with_options( + state, + &options.target_connection_id, + &options.target_database, + &target_sql, + Some(&options.target_schema), + None, + QueryExecutionOptions { max_rows: Some(row_limit), ..Default::default() }, + ) + .await?; + + let preparation = prepare_data_compare(DataComparePreparationOptions { + table_name: options.target_table, + schema: Some(options.target_schema), + columns: options.columns, + key_columns: options.key_columns, + source_rows: source_result.rows, + target_rows: target_result.rows, + database_type: Some(target_database_type), + })?; + + Ok(DataCompareFromTablesPreparation { + result: preparation.result, + sync_statements: preparation.sync_statements, + sync_sql: preparation.sync_sql, + source_row_count, + target_row_count, + source_truncated: source_row_count > row_limit as u64, + target_truncated: target_row_count > row_limit as u64, + }) +} + +pub fn compare_data_rows(options: CompareDataRowsOptions) -> Result { + if options.key_columns.is_empty() { + return Err("At least one key column is required for data comparison".to_string()); + } + + let mut source: HashMap> = HashMap::new(); + let mut target: HashMap> = HashMap::new(); + let mut source_order = Vec::new(); + let mut target_order = Vec::new(); + + for row in &options.source_rows { + let item = row_object(&options.columns, row); + let key = key_for(&item, &options.key_columns); + if source.contains_key(&key) { + return Err(format!("Duplicate source key: {key}")); + } + source_order.push(key.clone()); + source.insert(key, item); + } + + for row in &options.target_rows { + let item = row_object(&options.columns, row); + let key = key_for(&item, &options.key_columns); + if target.contains_key(&key) { + return Err(format!("Duplicate target key: {key}")); + } + target_order.push(key.clone()); + target.insert(key, item); + } + + let mut added = Vec::new(); + let mut removed = Vec::new(); + let mut modified = Vec::new(); + + for key in &source_order { + let source_values = source.get(key).expect("source key should exist"); + let Some(target_values) = target.get(key) else { + added.push(DataCompareRow { + key: key.clone(), + key_values: key_values(source_values, &options.key_columns), + values: source_values.clone(), + }); + continue; + }; + + let changes: Vec = options + .columns + .iter() + .filter(|column| !options.key_columns.contains(column)) + .filter(|column| value_for(source_values, column) != value_for(target_values, column)) + .map(|column| DataCompareChangedCell { + column: column.clone(), + source: value_for(source_values, column), + target: value_for(target_values, column), + }) + .collect(); + + if !changes.is_empty() { + modified.push(DataCompareModifiedRow { + key: key.clone(), + key_values: key_values(source_values, &options.key_columns), + source_values: source_values.clone(), + target_values: target_values.clone(), + changes, + }); + } + } + + for key in &target_order { + if let Some(target_values) = target.get(key).filter(|_| !source.contains_key(key)) { + removed.push(DataCompareRow { + key: key.clone(), + key_values: key_values(target_values, &options.key_columns), + values: target_values.clone(), + }); + } + } + + Ok(DataCompareResult { added, removed, modified }) +} + +#[derive(Debug, Clone)] +struct GenerateDataSyncSqlOptions { + table_name: String, + schema: Option, + columns: Vec, + key_columns: Vec, + diff: DataCompareResult, + database_type: Option, +} + +fn row_object(columns: &[String], row: &[Value]) -> HashMap { + columns + .iter() + .enumerate() + .map(|(index, column)| (column.clone(), row.get(index).cloned().unwrap_or(Value::Null))) + .collect() +} + +fn key_for(row: &HashMap, key_columns: &[String]) -> String { + key_columns.iter().map(|column| json_stringify(&value_for(row, column))).collect::>().join("\u{001f}") +} + +fn key_values(row: &HashMap, key_columns: &[String]) -> HashMap { + key_columns.iter().map(|column| (column.clone(), value_for(row, column))).collect() +} + +fn value_for(row: &HashMap, column: &str) -> Value { + row.get(column).cloned().unwrap_or(Value::Null) +} + +fn json_stringify(value: &Value) -> String { + serde_json::to_string(value).unwrap_or_else(|_| "null".to_string()) +} + +fn generate_data_sync_statements(options: &GenerateDataSyncSqlOptions) -> Vec { + let table = qualified_table_name(options.database_type, options.schema.as_deref(), &options.table_name); + let mut statements = Vec::new(); + + for row in &options.diff.added { + let columns = options + .columns + .iter() + .map(|column| quote_table_identifier(options.database_type, column)) + .collect::>() + .join(", "); + let values = options + .columns + .iter() + .map(|column| { + format_grid_sql_literal(row.values.get(column).unwrap_or(&Value::Null), options.database_type) + }) + .collect::>() + .join(", "); + statements.push(format!("INSERT INTO {table} ({columns}) VALUES ({values});")); + } + + for row in &options.diff.modified { + let assignments = row + .changes + .iter() + .map(|change| { + format!( + "{} = {}", + quote_table_identifier(options.database_type, &change.column), + format_grid_sql_literal(&change.source, options.database_type) + ) + }) + .collect::>() + .join(", "); + statements.push(format!( + "UPDATE {table} SET {assignments} WHERE {};", + where_by_key(&row.key_values, &options.key_columns, options.database_type) + )); + } + + for row in &options.diff.removed { + statements.push(format!( + "DELETE FROM {table} WHERE {};", + where_by_key(&row.key_values, &options.key_columns, options.database_type) + )); + } + + statements +} + +async fn connection_database_type(state: &AppState, connection_id: &str) -> Result { + state + .configs + .read() + .await + .get(connection_id) + .map(|config| config.db_type) + .ok_or_else(|| format!("Connection config not found: {connection_id}")) +} + +fn first_count(rows: &[Vec]) -> Result { + let value = rows.first().and_then(|row| row.first()).ok_or_else(|| "COUNT query returned no rows".to_string())?; + match value { + Value::Number(number) => { + number.as_u64().or_else(|| number.as_i64().and_then(|value| u64::try_from(value).ok())) + } + Value::String(text) => text.parse::().ok(), + _ => None, + } + .ok_or_else(|| format!("COUNT query returned non-numeric value: {value}")) +} + +fn build_data_compare_select_sql( + database_type: DatabaseType, + schema: &str, + table_name: &str, + columns: &[String], + key_columns: &[String], + row_limit: usize, +) -> String { + build_table_select_sql(TableSelectSqlOptions { + database_type: Some(database_type), + schema: Some(schema), + table_name, + columns, + order_columns: key_columns, + limit: row_limit, + }) +} + +fn where_by_key( + key_values: &HashMap, + key_columns: &[String], + database_type: Option, +) -> String { + key_columns + .iter() + .map(|column| { + format!( + "{} = {}", + quote_table_identifier(database_type, column), + format_grid_sql_literal(key_values.get(column).unwrap_or(&Value::Null), database_type) + ) + }) + .collect::>() + .join(" AND ") +} + +fn format_grid_sql_literal(value: &Value, database_type: Option) -> String { + match value { + Value::Null => "NULL".to_string(), + Value::Bool(value) => { + if *value { + "TRUE".to_string() + } else { + "FALSE".to_string() + } + } + Value::Number(number) => number.to_string(), + Value::String(text) if text.is_empty() && database_type == Some(DatabaseType::SqlServer) => "N''".to_string(), + Value::String(text) if text.is_empty() => "''".to_string(), + Value::String(text) => { + let escaped = format!("'{}'", literal_text(text, database_type).replace('\\', "\\\\").replace('\'', "''")); + if database_type == Some(DatabaseType::SqlServer) { + format!("N{escaped}") + } else { + escaped + } + } + other => { + let text = other.to_string(); + let escaped = format!("'{}'", text.replace('\\', "\\\\").replace('\'', "''")); + if database_type == Some(DatabaseType::SqlServer) { + format!("N{escaped}") + } else { + escaped + } + } + } +} + +fn literal_text(text: &str, database_type: Option) -> String { + if database_type == Some(DatabaseType::Tdengine) { + return format_tdengine_timestamp_literal_text(text); + } + text.to_string() +} + +fn format_tdengine_timestamp_literal_text(text: &str) -> String { + // Keep non-timestamp text unchanged; TDengine timestamp normalization is UI-parity best effort in Rust. + if text.len() < 19 || !text.as_bytes().get(10).is_some_and(|ch| *ch == b' ') { + return text.to_string(); + } + text.replacen(' ', "T", 1) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::models::connection::DatabaseType; + use serde_json::json; + + #[test] + fn compares_rows_by_primary_key_and_reports_added_removed_and_modified_rows() { + let diff = compare_data_rows(CompareDataRowsOptions { + columns: vec!["id".to_string(), "name".to_string(), "active".to_string()], + key_columns: vec!["id".to_string()], + source_rows: vec![ + vec![json!(1), json!("Ada"), json!(true)], + vec![json!(2), json!("Bob"), json!(false)], + vec![json!(4), json!("Dora"), json!(true)], + ], + target_rows: vec![ + vec![json!(1), json!("Ada"), json!(true)], + vec![json!(2), json!("Bobby"), json!(false)], + vec![json!(3), json!("Cara"), json!(true)], + ], + }) + .expect("data comparison should succeed"); + + assert_eq!( + diff.added.iter().map(|row| row.key_values.get("id").cloned()).collect::>(), + vec![Some(json!(4))] + ); + assert_eq!( + diff.removed.iter().map(|row| row.key_values.get("id").cloned()).collect::>(), + vec![Some(json!(3))] + ); + assert_eq!(diff.modified[0].changes[0].column, "name"); + assert_eq!(diff.modified[0].changes[0].source, json!("Bob")); + assert_eq!(diff.modified[0].changes[0].target, json!("Bobby")); + } + + #[test] + fn generates_data_synchronization_sql() { + let preparation = prepare_data_compare(DataComparePreparationOptions { + table_name: "users".to_string(), + schema: Some("public".to_string()), + columns: vec!["id".to_string(), "name".to_string(), "active".to_string()], + key_columns: vec!["id".to_string()], + source_rows: vec![vec![json!(1), json!("Ada"), json!(true)], vec![json!(2), json!("Bob"), json!(false)]], + target_rows: vec![ + vec![json!(1), json!("Ada Lovelace"), json!(true)], + vec![json!(3), json!("Cara"), json!(true)], + ], + database_type: Some(DatabaseType::Postgres), + }) + .expect("data compare preparation should succeed"); + + assert_eq!( + preparation.sync_sql, + [ + "INSERT INTO \"public\".\"users\" (\"id\", \"name\", \"active\") VALUES (2, 'Bob', FALSE);", + "UPDATE \"public\".\"users\" SET \"name\" = 'Ada' WHERE \"id\" = 1;", + "DELETE FROM \"public\".\"users\" WHERE \"id\" = 3;", + ] + .join("\n") + ); + assert_eq!(preparation.sync_statements.len(), 3); + } + + #[test] + fn requires_at_least_one_key_column() { + let err = compare_data_rows(CompareDataRowsOptions { + columns: vec!["id".to_string()], + key_columns: Vec::new(), + source_rows: vec![vec![json!(1)]], + target_rows: vec![vec![json!(1)]], + }) + .expect_err("missing key columns should fail"); + + assert!(err.contains("At least one key column")); + } + + #[test] + fn rejects_duplicate_row_keys() { + let err = compare_data_rows(CompareDataRowsOptions { + columns: vec!["id".to_string(), "name".to_string()], + key_columns: vec!["id".to_string()], + source_rows: vec![vec![json!(1), json!("Ada")], vec![json!(1), json!("Ada Clone")]], + target_rows: vec![vec![json!(1), json!("Ada")]], + }) + .expect_err("duplicate source keys should fail"); + + assert!(err.contains("Duplicate source key")); + } + + #[test] + fn builds_backend_table_select_sql_with_explicit_columns_and_key_order() { + assert_eq!( + build_data_compare_select_sql( + DatabaseType::Postgres, + "public", + "users", + &["id".to_string(), "name".to_string()], + &["id".to_string()], + 1000, + ), + "SELECT \"id\", \"name\" FROM \"public\".\"users\" ORDER BY \"id\" ASC LIMIT 1000;" + ); + } + + #[test] + fn builds_backend_table_select_sql_for_sqlserver_limit_syntax() { + assert_eq!( + build_data_compare_select_sql( + DatabaseType::SqlServer, + "dbo", + "users", + &["id".to_string(), "name".to_string()], + &["id".to_string()], + 50, + ), + "SELECT TOP (50) [id], [name] FROM [dbo].[users] ORDER BY [id] ASC" + ); + } + + #[test] + fn shared_sql_dialect_helpers_build_data_compare_table_sql() { + use crate::sql_dialect::{ + build_count_table_sql as build_shared_count_table_sql, build_table_select_sql, TableSelectSqlOptions, + }; + + assert_eq!( + build_shared_count_table_sql(Some(DatabaseType::Postgres), Some("public"), "users"), + "SELECT COUNT(*) AS row_count FROM \"public\".\"users\"" + ); + assert_eq!( + build_table_select_sql(TableSelectSqlOptions { + database_type: Some(DatabaseType::Oracle), + schema: Some("APP"), + table_name: "EVENTS", + columns: &["ID".to_string(), "NAME".to_string()], + order_columns: &["ID".to_string()], + limit: 25, + }), + "SELECT \"ID\", \"NAME\" FROM \"APP\".\"EVENTS\" ORDER BY \"ID\" ASC FETCH FIRST 25 ROWS ONLY" + ); + } +} diff --git a/crates/dbx-core/src/data_grid_sql.rs b/crates/dbx-core/src/data_grid_sql.rs index a2237a8e11..d29625b050 100644 --- a/crates/dbx-core/src/data_grid_sql.rs +++ b/crates/dbx-core/src/data_grid_sql.rs @@ -3,6 +3,7 @@ use serde::{Deserialize, Serialize}; use serde_json::Value; use crate::models::connection::DatabaseType; +use crate::sql_dialect::quote_table_identifier; const DBX_ROWID_COLUMN: &str = "__DBX_ROWID"; const DBX_NEO4J_ELEMENT_ID_COLUMN: &str = "__DBX_ELEMENT_ID"; @@ -976,55 +977,11 @@ fn predicate_ident(database_type: Option, name: &str) -> String { } fn quote_ident(database_type: Option, name: &str) -> String { - match database_type { - Some(DatabaseType::Mysql | DatabaseType::Hive | DatabaseType::Tdengine | DatabaseType::Access) => { - format!("`{}`", name.replace('`', "``")) - } - Some(DatabaseType::Informix) if is_plain_identifier(name) => name.to_string(), - Some(DatabaseType::Neo4j) => format!("`{}`", name.replace('`', "``")), - Some(DatabaseType::SqlServer) => format!("[{}]", name.replace(']', "]]")), - _ => format!("\"{}\"", name.replace('"', "\"\"")), - } + quote_table_identifier(database_type, name) } fn qualified_table_name(database_type: Option, schema: Option<&str>, table_name: &str) -> String { - if is_schema_aware(database_type) && schema.is_some_and(|schema| !schema.is_empty()) { - format!("{}.{}", quote_ident(database_type, schema.unwrap()), quote_ident(database_type, table_name)) - } else { - quote_ident(database_type, table_name) - } -} - -fn is_schema_aware(database_type: Option) -> bool { - matches!( - database_type, - Some( - DatabaseType::Postgres - | DatabaseType::SqlServer - | DatabaseType::Oracle - | DatabaseType::Redshift - | DatabaseType::Dameng - | DatabaseType::Gaussdb - | DatabaseType::Kingbase - | DatabaseType::Highgo - | DatabaseType::Vastbase - | DatabaseType::Yashandb - | DatabaseType::Databricks - | DatabaseType::SapHana - | DatabaseType::Teradata - | DatabaseType::Vertica - | DatabaseType::Exasol - | DatabaseType::OpenGauss - | DatabaseType::OceanbaseOracle - | DatabaseType::Gbase - | DatabaseType::Jdbc - | DatabaseType::H2 - | DatabaseType::Snowflake - | DatabaseType::Trino - | DatabaseType::Db2 - | DatabaseType::Tdengine - ) - ) + crate::sql_dialect::qualified_table_name(database_type, schema, table_name) } fn uses_keyless_row_predicate(database_type: Option) -> bool { @@ -1068,15 +1025,6 @@ fn uses_keyless_row_predicate(database_type: Option) -> bool { ) } -fn is_plain_identifier(name: &str) -> bool { - let mut chars = name.chars(); - let Some(first) = chars.next() else { - return false; - }; - (first.is_ascii_alphabetic() || first == '_') - && chars.all(|ch| ch.is_ascii_alphanumeric() || ch == '_' || ch == '$') -} - fn column_info_for<'a>(columns: &'a [DataGridColumnInfo], name: &str) -> Option<&'a DataGridColumnInfo> { let normalized = normalize_column_name(name); columns.iter().find(|column| normalize_column_name(&column.name) == normalized) diff --git a/crates/dbx-core/src/lib.rs b/crates/dbx-core/src/lib.rs index fd3c9a5b85..932705a93d 100644 --- a/crates/dbx-core/src/lib.rs +++ b/crates/dbx-core/src/lib.rs @@ -4,6 +4,7 @@ pub mod ai; pub mod connection; pub mod connection_secrets; pub mod csv_export; +pub mod data_compare; pub mod data_grid_sql; pub mod database_capabilities; pub mod database_export; @@ -18,8 +19,10 @@ pub mod query_cancel; pub mod redis_ops; pub mod saved_sql; pub mod schema; +pub mod schema_diff; pub mod sql; pub mod sql_analysis; +pub mod sql_dialect; pub mod sql_editability; pub mod storage; pub mod table_import; diff --git a/crates/dbx-core/src/schema_diff.rs b/crates/dbx-core/src/schema_diff.rs new file mode 100644 index 0000000000..4f9e4686c3 --- /dev/null +++ b/crates/dbx-core/src/schema_diff.rs @@ -0,0 +1,1050 @@ +use std::collections::{HashMap, HashSet}; + +use serde::{Deserialize, Serialize}; + +use crate::models::connection::DatabaseType; +use crate::types::{ColumnInfo, ForeignKeyInfo, IndexInfo, TableInfo, TriggerInfo}; + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct ColumnDiff { + #[serde(rename = "type")] + pub diff_type: String, + pub name: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub source: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub target: Option, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub changes: Vec, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct IndexDiff { + #[serde(rename = "type")] + pub diff_type: String, + pub name: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub source: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub target: Option, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub changes: Vec, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct ForeignKeyDiff { + #[serde(rename = "type")] + pub diff_type: String, + pub name: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub source: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub target: Option, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub changes: Vec, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct TriggerDiff { + #[serde(rename = "type")] + pub diff_type: String, + pub name: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub source: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub target: Option, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub changes: Vec, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct TableDiff { + #[serde(rename = "type")] + pub diff_type: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub object_type: Option, + pub name: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub columns: Option>, + #[serde(skip_serializing_if = "Option::is_none")] + pub indexes: Option>, + #[serde(skip_serializing_if = "Option::is_none")] + pub foreign_keys: Option>, + #[serde(skip_serializing_if = "Option::is_none")] + pub triggers: Option>, + #[serde(skip_serializing_if = "Option::is_none")] + pub ddl: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub source_table_comment: Option>, + #[serde(skip_serializing_if = "Option::is_none")] + pub target_table_comment: Option>, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct TableSchemaDetail { + pub name: String, + #[serde(default)] + pub columns: Vec, + #[serde(default)] + pub indexes: Vec, + #[serde(default)] + pub foreign_keys: Vec, + #[serde(default)] + pub triggers: Vec, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub ddl: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct SchemaDiffPreparationOptions { + #[serde(default)] + pub source_tables: Vec, + #[serde(default)] + pub target_tables: Vec, + #[serde(default)] + pub source_details: Vec, + #[serde(default)] + pub target_details: Vec, + pub database_type: DatabaseType, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub target_schema: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct SchemaDiffPreparation { + pub diffs: Vec, + pub sync_sql: String, +} + +pub fn prepare_schema_diff(options: SchemaDiffPreparationOptions) -> SchemaDiffPreparation { + let diffs = diff_schema(&options); + let sync_sql = generate_schema_sync_sql(&diffs, options.database_type, options.target_schema.as_deref()); + SchemaDiffPreparation { diffs, sync_sql } +} + +fn diff_schema(options: &SchemaDiffPreparationOptions) -> Vec { + let source_details: HashMap<&str, &TableSchemaDetail> = + options.source_details.iter().map(|detail| (detail.name.as_str(), detail)).collect(); + let target_details: HashMap<&str, &TableSchemaDetail> = + options.target_details.iter().map(|detail| (detail.name.as_str(), detail)).collect(); + let source_table_comments: HashMap<&str, Option> = + options.source_tables.iter().map(|table| (table.name.as_str(), table.comment.clone())).collect(); + let target_table_comments: HashMap<&str, Option> = + options.target_tables.iter().map(|table| (table.name.as_str(), table.comment.clone())).collect(); + + let source_table_names: Vec = options + .source_tables + .iter() + .filter(|table| table.table_type != "VIEW") + .map(|table| table.name.clone()) + .collect(); + let target_table_names: Vec = options + .target_tables + .iter() + .filter(|table| table.table_type != "VIEW") + .map(|table| table.name.clone()) + .collect(); + let source_view_names: Vec = options + .source_tables + .iter() + .filter(|table| table.table_type == "VIEW") + .map(|table| table.name.clone()) + .collect(); + let target_view_names: Vec = options + .target_tables + .iter() + .filter(|table| table.table_type == "VIEW") + .map(|table| table.name.clone()) + .collect(); + + let (added, removed, common) = diff_names(&source_table_names, &target_table_names); + let (added_views, removed_views, _) = diff_names(&source_view_names, &target_view_names); + let mut result = Vec::new(); + + for name in added { + result.push(TableDiff { + diff_type: "added".to_string(), + object_type: Some("table".to_string()), + ddl: source_details.get(name.as_str()).and_then(|detail| detail.ddl.clone()), + name, + columns: None, + indexes: None, + foreign_keys: None, + triggers: None, + source_table_comment: None, + target_table_comment: None, + }); + } + + for name in removed { + result.push(TableDiff { + diff_type: "removed".to_string(), + object_type: Some("table".to_string()), + name, + columns: None, + indexes: None, + foreign_keys: None, + triggers: None, + ddl: None, + source_table_comment: None, + target_table_comment: None, + }); + } + + for name in added_views { + result.push(TableDiff { + diff_type: "added".to_string(), + object_type: Some("view".to_string()), + name, + columns: None, + indexes: None, + foreign_keys: None, + triggers: None, + ddl: None, + source_table_comment: None, + target_table_comment: None, + }); + } + + for name in removed_views { + result.push(TableDiff { + diff_type: "removed".to_string(), + object_type: Some("view".to_string()), + name, + columns: None, + indexes: None, + foreign_keys: None, + triggers: None, + ddl: None, + source_table_comment: None, + target_table_comment: None, + }); + } + + for name in common { + let Some(source) = source_details.get(name.as_str()) else { continue }; + let Some(target) = target_details.get(name.as_str()) else { continue }; + let column_diffs = diff_columns(&source.columns, &target.columns); + let index_diffs = diff_indexes(&source.indexes, &target.indexes); + let foreign_key_diffs = diff_foreign_keys(&source.foreign_keys, &target.foreign_keys); + let trigger_diffs = diff_triggers(&source.triggers, &target.triggers); + let source_comment = source_table_comments.get(name.as_str()).cloned().unwrap_or(None); + let target_comment = target_table_comments.get(name.as_str()).cloned().unwrap_or(None); + let comment_changed = source_comment.clone().unwrap_or_default() != target_comment.clone().unwrap_or_default(); + + if !column_diffs.is_empty() + || !index_diffs.is_empty() + || !foreign_key_diffs.is_empty() + || !trigger_diffs.is_empty() + || comment_changed + { + result.push(TableDiff { + diff_type: "modified".to_string(), + object_type: Some("table".to_string()), + name, + columns: (!column_diffs.is_empty()).then_some(column_diffs), + indexes: (!index_diffs.is_empty()).then_some(index_diffs), + foreign_keys: (!foreign_key_diffs.is_empty()).then_some(foreign_key_diffs), + triggers: (!trigger_diffs.is_empty()).then_some(trigger_diffs), + ddl: None, + source_table_comment: comment_changed.then_some(source_comment), + target_table_comment: comment_changed.then_some(target_comment), + }); + } + } + + result +} + +fn diff_names(source: &[String], target: &[String]) -> (Vec, Vec, Vec) { + let source_set: HashSet<&str> = source.iter().map(String::as_str).collect(); + let target_set: HashSet<&str> = target.iter().map(String::as_str).collect(); + ( + source.iter().filter(|name| !target_set.contains(name.as_str())).cloned().collect(), + target.iter().filter(|name| !source_set.contains(name.as_str())).cloned().collect(), + source.iter().filter(|name| target_set.contains(name.as_str())).cloned().collect(), + ) +} + +pub fn diff_columns(source: &[ColumnInfo], target: &[ColumnInfo]) -> Vec { + let mut diffs = Vec::new(); + let target_map: HashMap<&str, &ColumnInfo> = target.iter().map(|column| (column.name.as_str(), column)).collect(); + let source_map: HashMap<&str, &ColumnInfo> = source.iter().map(|column| (column.name.as_str(), column)).collect(); + + for source_column in source { + if let Some(target_column) = target_map.get(source_column.name.as_str()) { + let mut changes = Vec::new(); + if source_column.data_type.to_lowercase() != target_column.data_type.to_lowercase() { + changes.push(format!("type: {} → {}", target_column.data_type, source_column.data_type)); + } + if source_column.is_nullable != target_column.is_nullable { + changes.push(format!( + "nullable: {} → {}", + if target_column.is_nullable { "YES" } else { "NO" }, + if source_column.is_nullable { "YES" } else { "NO" } + )); + } + if source_column.column_default.as_deref().unwrap_or_default() + != target_column.column_default.as_deref().unwrap_or_default() + { + changes.push(format!( + "default: {} → {}", + target_column.column_default.as_deref().unwrap_or("NULL"), + source_column.column_default.as_deref().unwrap_or("NULL") + )); + } + if source_column.comment.as_deref().unwrap_or_default() + != target_column.comment.as_deref().unwrap_or_default() + { + changes.push(format!( + "comment: {} → {}", + target_column.comment.as_deref().unwrap_or_default(), + source_column.comment.as_deref().unwrap_or_default() + )); + } + if !changes.is_empty() { + diffs.push(ColumnDiff { + diff_type: "modified".to_string(), + name: source_column.name.clone(), + source: Some(source_column.clone()), + target: Some((*target_column).clone()), + changes, + }); + } + } else { + diffs.push(ColumnDiff { + diff_type: "added".to_string(), + name: source_column.name.clone(), + source: Some(source_column.clone()), + target: None, + changes: Vec::new(), + }); + } + } + + for target_column in target { + if !source_map.contains_key(target_column.name.as_str()) { + diffs.push(ColumnDiff { + diff_type: "removed".to_string(), + name: target_column.name.clone(), + source: None, + target: Some(target_column.clone()), + changes: Vec::new(), + }); + } + } + + diffs +} + +pub fn diff_indexes(source: &[IndexInfo], target: &[IndexInfo]) -> Vec { + let mut diffs = Vec::new(); + let target_map: HashMap<&str, &IndexInfo> = target.iter().map(|index| (index.name.as_str(), index)).collect(); + let source_map: HashMap<&str, &IndexInfo> = source.iter().map(|index| (index.name.as_str(), index)).collect(); + + for source_index in source { + if source_index.is_primary { + continue; + } + let Some(target_index) = target_map.get(source_index.name.as_str()) else { + diffs.push(IndexDiff { + diff_type: "added".to_string(), + name: source_index.name.clone(), + source: Some(source_index.clone()), + target: None, + changes: Vec::new(), + }); + continue; + }; + + let mut changes = Vec::new(); + if source_index.is_unique != target_index.is_unique { + changes.push(format!( + "unique: {} → {}", + if target_index.is_unique { "YES" } else { "NO" }, + if source_index.is_unique { "YES" } else { "NO" } + )); + } + if source_index.columns.join(",") != target_index.columns.join(",") { + changes.push(format!("columns: {} → {}", target_index.columns.join(", "), source_index.columns.join(", "))); + } + if source_index.index_type.as_deref().unwrap_or_default() + != target_index.index_type.as_deref().unwrap_or_default() + { + changes.push(format!( + "type: {} → {}", + target_index.index_type.as_deref().unwrap_or("default"), + source_index.index_type.as_deref().unwrap_or("default") + )); + } + if source_index.filter.as_deref().unwrap_or_default() != target_index.filter.as_deref().unwrap_or_default() { + changes.push(format!( + "filter: {} → {}", + target_index.filter.as_deref().unwrap_or("none"), + source_index.filter.as_deref().unwrap_or("none") + )); + } + let source_included = source_index.included_columns.clone().unwrap_or_default(); + let target_included = target_index.included_columns.clone().unwrap_or_default(); + if source_included.join(",") != target_included.join(",") { + changes.push(format!( + "include: {} → {}", + if target_included.is_empty() { "none".to_string() } else { target_included.join(", ") }, + if source_included.is_empty() { "none".to_string() } else { source_included.join(", ") } + )); + } + if !changes.is_empty() { + diffs.push(IndexDiff { + diff_type: "modified".to_string(), + name: source_index.name.clone(), + source: Some(source_index.clone()), + target: Some((*target_index).clone()), + changes, + }); + } + } + + for target_index in target { + if target_index.is_primary { + continue; + } + if !source_map.contains_key(target_index.name.as_str()) { + diffs.push(IndexDiff { + diff_type: "removed".to_string(), + name: target_index.name.clone(), + source: None, + target: Some(target_index.clone()), + changes: Vec::new(), + }); + } + } + + diffs +} + +pub fn diff_foreign_keys(source: &[ForeignKeyInfo], target: &[ForeignKeyInfo]) -> Vec { + let mut diffs = Vec::new(); + let target_map: HashMap<&str, &ForeignKeyInfo> = target.iter().map(|fk| (fk.name.as_str(), fk)).collect(); + let source_map: HashMap<&str, &ForeignKeyInfo> = source.iter().map(|fk| (fk.name.as_str(), fk)).collect(); + + for source_fk in source { + let Some(target_fk) = target_map.get(source_fk.name.as_str()) else { + diffs.push(ForeignKeyDiff { + diff_type: "added".to_string(), + name: source_fk.name.clone(), + source: Some(source_fk.clone()), + target: None, + changes: Vec::new(), + }); + continue; + }; + + let mut changes = Vec::new(); + if source_fk.column != target_fk.column { + changes.push(format!("column: {} → {}", target_fk.column, source_fk.column)); + } + if source_fk.ref_table != target_fk.ref_table { + changes.push(format!("ref table: {} → {}", target_fk.ref_table, source_fk.ref_table)); + } + if source_fk.ref_column != target_fk.ref_column { + changes.push(format!("ref column: {} → {}", target_fk.ref_column, source_fk.ref_column)); + } + if !changes.is_empty() { + diffs.push(ForeignKeyDiff { + diff_type: "modified".to_string(), + name: source_fk.name.clone(), + source: Some(source_fk.clone()), + target: Some((*target_fk).clone()), + changes, + }); + } + } + + for target_fk in target { + if !source_map.contains_key(target_fk.name.as_str()) { + diffs.push(ForeignKeyDiff { + diff_type: "removed".to_string(), + name: target_fk.name.clone(), + source: None, + target: Some(target_fk.clone()), + changes: Vec::new(), + }); + } + } + + diffs +} + +pub fn diff_triggers(source: &[TriggerInfo], target: &[TriggerInfo]) -> Vec { + let mut diffs = Vec::new(); + let target_map: HashMap<&str, &TriggerInfo> = + target.iter().map(|trigger| (trigger.name.as_str(), trigger)).collect(); + let source_map: HashMap<&str, &TriggerInfo> = + source.iter().map(|trigger| (trigger.name.as_str(), trigger)).collect(); + + for source_trigger in source { + let Some(target_trigger) = target_map.get(source_trigger.name.as_str()) else { + diffs.push(TriggerDiff { + diff_type: "added".to_string(), + name: source_trigger.name.clone(), + source: Some(source_trigger.clone()), + target: None, + changes: Vec::new(), + }); + continue; + }; + + let mut changes = Vec::new(); + if source_trigger.event != target_trigger.event { + changes.push(format!("event: {} → {}", target_trigger.event, source_trigger.event)); + } + if source_trigger.timing != target_trigger.timing { + changes.push(format!("timing: {} → {}", target_trigger.timing, source_trigger.timing)); + } + if !changes.is_empty() { + diffs.push(TriggerDiff { + diff_type: "modified".to_string(), + name: source_trigger.name.clone(), + source: Some(source_trigger.clone()), + target: Some((*target_trigger).clone()), + changes, + }); + } + } + + for target_trigger in target { + if !source_map.contains_key(target_trigger.name.as_str()) { + diffs.push(TriggerDiff { + diff_type: "removed".to_string(), + name: target_trigger.name.clone(), + source: None, + target: Some(target_trigger.clone()), + changes: Vec::new(), + }); + } + } + + diffs +} + +fn quote_id(name: &str, db_type: DatabaseType) -> String { + if matches!(db_type, DatabaseType::Mysql | DatabaseType::Doris | DatabaseType::StarRocks) { + format!("`{}`", name.replace('`', "``")) + } else { + format!("\"{}\"", name.replace('"', "\"\"")) + } +} + +fn column_def(col: &ColumnInfo, db_type: DatabaseType) -> String { + let mut definition = format!("{} {}", quote_id(&col.name, db_type), col.data_type); + if !col.is_nullable { + definition.push_str(" NOT NULL"); + } + if let Some(default) = &col.column_default { + definition.push_str(&format!(" DEFAULT {default}")); + } + definition +} + +fn qualified_name(name: &str, db_type: DatabaseType, schema: Option<&str>) -> String { + schema + .map(|schema| format!("{}.{}", quote_id(schema, db_type), quote_id(name, db_type))) + .unwrap_or_else(|| quote_id(name, db_type)) +} + +fn drop_index_sql(table_name: &str, index_name: &str, db_type: DatabaseType, schema: Option<&str>) -> String { + let table = qualified_name(table_name, db_type, schema); + let index = qualified_name(index_name, db_type, schema); + if matches!(db_type, DatabaseType::Mysql | DatabaseType::Doris | DatabaseType::StarRocks) { + format!("DROP INDEX {} ON {table};", quote_id(index_name, db_type)) + } else { + format!("DROP INDEX IF EXISTS {index};") + } +} + +fn create_index_sql(table_name: &str, index: &IndexInfo, db_type: DatabaseType, schema: Option<&str>) -> String { + let table = qualified_name(table_name, db_type, schema); + let columns = index.columns.iter().map(|column| quote_id(column, db_type)).collect::>().join(", "); + let unique = if index.is_unique { "UNIQUE " } else { "" }; + let index_type = index.index_type.as_deref().unwrap_or_default(); + let using_clause = if !index_type.is_empty() && db_type == DatabaseType::Postgres { + format!(" USING {index_type}") + } else { + String::new() + }; + let type_prefix = if !index_type.is_empty() && db_type == DatabaseType::SqlServer { + format!("{index_type} ") + } else { + String::new() + }; + let included_columns = index.included_columns.clone().unwrap_or_default(); + let include_clause = + if !included_columns.is_empty() && matches!(db_type, DatabaseType::Postgres | DatabaseType::SqlServer) { + format!( + " INCLUDE ({})", + included_columns.iter().map(|column| quote_id(column, db_type)).collect::>().join(", ") + ) + } else { + String::new() + }; + let supports_where = matches!(db_type, DatabaseType::Postgres | DatabaseType::SqlServer | DatabaseType::Sqlite); + let filter = if supports_where { index.filter.as_deref().unwrap_or_default() } else { "" }; + let filter_clause = if filter.is_empty() { String::new() } else { format!(" WHERE {filter}") }; + format!( + "CREATE {unique}{type_prefix}INDEX {} ON {table}{using_clause} ({columns}){include_clause}{filter_clause};", + quote_id(&index.name, db_type) + ) +} + +fn drop_foreign_key_sql(table_name: &str, fk_name: &str, db_type: DatabaseType, schema: Option<&str>) -> String { + let table = qualified_name(table_name, db_type, schema); + let fk = quote_id(fk_name, db_type); + if matches!(db_type, DatabaseType::Mysql | DatabaseType::Doris | DatabaseType::StarRocks) { + format!("ALTER TABLE {table} DROP FOREIGN KEY {fk};") + } else { + format!("ALTER TABLE {table} DROP CONSTRAINT {fk};") + } +} + +fn add_foreign_key_sql(table_name: &str, fk: &ForeignKeyInfo, db_type: DatabaseType, schema: Option<&str>) -> String { + let table = qualified_name(table_name, db_type, schema); + format!( + "ALTER TABLE {table} ADD CONSTRAINT {} FOREIGN KEY ({}) REFERENCES {} ({});", + quote_id(&fk.name, db_type), + quote_id(&fk.column, db_type), + quote_id(&fk.ref_table, db_type), + quote_id(&fk.ref_column, db_type) + ) +} + +fn drop_object_sql(diff: &TableDiff, db_type: DatabaseType, schema: Option<&str>) -> String { + let object_type = if diff.object_type.as_deref() == Some("view") { "VIEW" } else { "TABLE" }; + format!("DROP {object_type} IF EXISTS {};", qualified_name(&diff.name, db_type, schema)) +} + +fn comment_literal(comment: &str) -> String { + format!("'{}'", comment.replace('\'', "''")) +} + +fn column_comment_sql( + table_name: &str, + column_name: &str, + comment: &str, + db_type: DatabaseType, + schema: Option<&str>, +) -> String { + if matches!(db_type, DatabaseType::Mysql | DatabaseType::Doris | DatabaseType::StarRocks) { + return format!( + "-- Column comment for {column_name}: use ALTER TABLE ... MODIFY COLUMN to set comment in MySQL" + ); + } + let table = qualified_name(table_name, db_type, schema); + format!("COMMENT ON COLUMN {table}.{} IS {};", quote_id(column_name, db_type), comment_literal(comment)) +} + +fn table_comment_sql(table_name: &str, comment: &str, db_type: DatabaseType, schema: Option<&str>) -> String { + let table = qualified_name(table_name, db_type, schema); + if matches!(db_type, DatabaseType::Mysql | DatabaseType::Doris | DatabaseType::StarRocks) { + format!("ALTER TABLE {table} COMMENT = {};", comment_literal(comment)) + } else { + format!("COMMENT ON TABLE {table} IS {};", comment_literal(comment)) + } +} + +pub fn generate_schema_sync_sql(diffs: &[TableDiff], db_type: DatabaseType, schema: Option<&str>) -> String { + let mut lines = Vec::new(); + let is_mysql = matches!(db_type, DatabaseType::Mysql | DatabaseType::Doris | DatabaseType::StarRocks); + + for diff in diffs { + let table = qualified_name(&diff.name, db_type, schema); + + if diff.diff_type == "added" && diff.ddl.is_some() { + lines.push(format!("-- Create {}: {}", diff.object_type.as_deref().unwrap_or("table"), diff.name)); + lines.push(format!("{};", diff.ddl.as_deref().unwrap_or_default())); + lines.push(String::new()); + continue; + } + + if diff.diff_type == "added" && diff.object_type.as_deref() == Some("view") { + lines.push(format!("-- View exists only in source: {}", diff.name)); + lines.push("-- Source view definition is not available from this driver yet.".to_string()); + lines.push(String::new()); + continue; + } + + if diff.diff_type == "removed" { + lines.push(format!("-- Drop {}: {}", diff.object_type.as_deref().unwrap_or("table"), diff.name)); + lines.push(drop_object_sql(diff, db_type, schema)); + lines.push(String::new()); + continue; + } + + if diff.diff_type != "modified" { + continue; + } + + let mut parts = Vec::new(); + if let Some(foreign_keys) = &diff.foreign_keys { + for fk in foreign_keys { + if fk.diff_type == "removed" || fk.diff_type == "modified" { + lines.push(drop_foreign_key_sql(&diff.name, &fk.name, db_type, schema)); + } + } + } + + if let Some(columns) = &diff.columns { + for column in columns { + match column.diff_type.as_str() { + "added" => { + if let Some(source) = &column.source { + parts.push(format!(" ADD COLUMN {}", column_def(source, db_type))); + } + } + "removed" => { + parts.push(format!(" DROP COLUMN {}", quote_id(&column.name, db_type))); + } + "modified" => { + if let Some(source) = &column.source { + if is_mysql { + parts.push(format!(" MODIFY COLUMN {}", column_def(source, db_type))); + } else { + let name = quote_id(&column.name, db_type); + if column.changes.iter().any(|change| change.starts_with("type:")) { + parts.push(format!(" ALTER COLUMN {name} TYPE {}", source.data_type)); + } + if column.changes.iter().any(|change| change.starts_with("nullable:")) { + parts.push(if source.is_nullable { + format!(" ALTER COLUMN {name} DROP NOT NULL") + } else { + format!(" ALTER COLUMN {name} SET NOT NULL") + }); + } + if column.changes.iter().any(|change| change.starts_with("default:")) { + parts.push(if let Some(default) = &source.column_default { + format!(" ALTER COLUMN {name} SET DEFAULT {default}") + } else { + format!(" ALTER COLUMN {name} DROP DEFAULT") + }); + } + } + } + } + _ => {} + } + } + } + + if !parts.is_empty() { + lines.push(format!("-- Alter table: {}", diff.name)); + if is_mysql { + lines.push(format!("ALTER TABLE {table}")); + lines.push(format!("{};", parts.join(",\n"))); + } else { + for part in parts { + lines.push(format!("ALTER TABLE {table}{part};")); + } + } + lines.push(String::new()); + } + + if let Some(columns) = &diff.columns { + for column in columns { + if let Some(source) = &column.source { + if column.changes.iter().any(|change| change.starts_with("comment:")) { + lines.push(column_comment_sql( + &diff.name, + &column.name, + source.comment.as_deref().unwrap_or_default(), + db_type, + schema, + )); + } + if column.diff_type == "added" { + if let Some(comment) = &source.comment { + lines.push(column_comment_sql(&diff.name, &column.name, comment, db_type, schema)); + } + } + } + } + } + + if diff.source_table_comment.is_some() && diff.source_table_comment != diff.target_table_comment { + let comment = diff.source_table_comment.as_ref().and_then(|comment| comment.as_deref()).unwrap_or_default(); + lines.push(table_comment_sql(&diff.name, comment, db_type, schema)); + } + + if let Some(indexes) = &diff.indexes { + for index in indexes { + match index.diff_type.as_str() { + "added" => { + if let Some(source) = &index.source { + lines.push(create_index_sql(&diff.name, source, db_type, schema)); + } + } + "removed" => lines.push(drop_index_sql(&diff.name, &index.name, db_type, schema)), + "modified" => { + if let Some(source) = &index.source { + lines.push(drop_index_sql(&diff.name, &index.name, db_type, schema)); + lines.push(create_index_sql(&diff.name, source, db_type, schema)); + } + } + _ => {} + } + } + } + + if let Some(foreign_keys) = &diff.foreign_keys { + for fk in foreign_keys { + if (fk.diff_type == "added" || fk.diff_type == "modified") && fk.source.is_some() { + lines.push(add_foreign_key_sql(&diff.name, fk.source.as_ref().unwrap(), db_type, schema)); + } + } + } + + if let Some(triggers) = &diff.triggers { + for trigger in triggers { + lines.push(format!( + "-- Trigger {}: {} on {}; review trigger definition manually.", + trigger.diff_type, trigger.name, diff.name + )); + } + } + + if diff.indexes.as_ref().is_some_and(|indexes| !indexes.is_empty()) + || diff.foreign_keys.as_ref().is_some_and(|foreign_keys| !foreign_keys.is_empty()) + || diff.triggers.as_ref().is_some_and(|triggers| !triggers.is_empty()) + { + lines.push(String::new()); + } + + if db_type == DatabaseType::Sqlite + && diff.foreign_keys.as_ref().is_some_and(|foreign_keys| !foreign_keys.is_empty()) + { + lines.push(format!("-- SQLite foreign key synchronization may require table rebuild for: {}", diff.name)); + lines.push(String::new()); + } + } + + lines.join("\n").trim().to_string() +} + +#[cfg(test)] +mod tests { + use super::*; + + fn index(overrides: IndexInfo) -> IndexInfo { + IndexInfo { + name: if overrides.name.is_empty() { "idx_users_email".to_string() } else { overrides.name }, + columns: if overrides.columns.is_empty() { vec!["email".to_string()] } else { overrides.columns }, + is_unique: overrides.is_unique, + is_primary: overrides.is_primary, + filter: overrides.filter, + index_type: overrides.index_type, + included_columns: overrides.included_columns, + comment: overrides.comment, + } + } + + fn foreign_key(overrides: ForeignKeyInfo) -> ForeignKeyInfo { + ForeignKeyInfo { + name: if overrides.name.is_empty() { "orders_user_id_fk".to_string() } else { overrides.name }, + column: if overrides.column.is_empty() { "user_id".to_string() } else { overrides.column }, + ref_table: if overrides.ref_table.is_empty() { "users".to_string() } else { overrides.ref_table }, + ref_column: if overrides.ref_column.is_empty() { "id".to_string() } else { overrides.ref_column }, + } + } + + #[test] + fn detects_modified_indexes_not_only_added_or_removed_indexes() { + let diffs = diff_indexes( + &[index(IndexInfo { + name: "idx_orders_status".to_string(), + columns: vec!["status".to_string(), "created_at".to_string()], + is_unique: false, + is_primary: false, + filter: None, + index_type: None, + included_columns: None, + comment: None, + })], + &[index(IndexInfo { + name: "idx_orders_status".to_string(), + columns: vec!["status".to_string()], + is_unique: true, + is_primary: false, + filter: None, + index_type: None, + included_columns: None, + comment: None, + })], + ); + + assert_eq!(diffs.len(), 1); + assert_eq!(diffs[0].diff_type, "modified"); + assert_eq!(diffs[0].changes, vec!["unique: YES → NO", "columns: status → status, created_at"]); + } + + #[test] + fn detects_foreign_key_additions_removals_and_target_changes() { + let diffs = diff_foreign_keys( + &[ + foreign_key(ForeignKeyInfo { + name: "orders_user_id_fk".to_string(), + column: String::new(), + ref_table: String::new(), + ref_column: String::new(), + }), + foreign_key(ForeignKeyInfo { + name: "orders_account_id_fk".to_string(), + column: "account_id".to_string(), + ref_table: "accounts".to_string(), + ref_column: String::new(), + }), + ], + &[ + foreign_key(ForeignKeyInfo { + name: "orders_user_id_fk".to_string(), + column: String::new(), + ref_table: "members".to_string(), + ref_column: String::new(), + }), + foreign_key(ForeignKeyInfo { + name: "orders_region_id_fk".to_string(), + column: "region_id".to_string(), + ref_table: "regions".to_string(), + ref_column: String::new(), + }), + ], + ); + + let summary: Vec<_> = diffs.iter().map(|diff| (diff.diff_type.as_str(), diff.name.as_str())).collect(); + assert_eq!( + summary, + vec![ + ("modified", "orders_user_id_fk"), + ("added", "orders_account_id_fk"), + ("removed", "orders_region_id_fk"), + ] + ); + } + + #[test] + fn generates_sync_sql_for_index_and_foreign_key_changes() { + let diffs = vec![TableDiff { + diff_type: "modified".to_string(), + object_type: None, + name: "orders".to_string(), + columns: None, + indexes: Some(vec![IndexDiff { + diff_type: "modified".to_string(), + name: "idx_orders_status".to_string(), + source: Some(index(IndexInfo { + name: "idx_orders_status".to_string(), + columns: vec!["status".to_string(), "created_at".to_string()], + is_unique: true, + is_primary: false, + filter: None, + index_type: None, + included_columns: None, + comment: None, + })), + target: None, + changes: Vec::new(), + }]), + foreign_keys: Some(vec![ForeignKeyDiff { + diff_type: "modified".to_string(), + name: "orders_user_id_fk".to_string(), + source: Some(foreign_key(ForeignKeyInfo { + name: "orders_user_id_fk".to_string(), + column: String::new(), + ref_table: "users".to_string(), + ref_column: String::new(), + })), + target: None, + changes: Vec::new(), + }]), + triggers: None, + ddl: None, + source_table_comment: None, + target_table_comment: None, + }]; + + assert_eq!( + generate_schema_sync_sql(&diffs, DatabaseType::Postgres, None), + [ + "ALTER TABLE \"orders\" DROP CONSTRAINT \"orders_user_id_fk\";", + "DROP INDEX IF EXISTS \"idx_orders_status\";", + "CREATE UNIQUE INDEX \"idx_orders_status\" ON \"orders\" (\"status\", \"created_at\");", + "ALTER TABLE \"orders\" ADD CONSTRAINT \"orders_user_id_fk\" FOREIGN KEY (\"user_id\") REFERENCES \"users\" (\"id\");", + ] + .join("\n") + ); + } + + #[test] + fn qualifies_generated_schema_sync_sql_with_target_schema() { + let diffs = vec![TableDiff { + diff_type: "modified".to_string(), + object_type: None, + name: "orders".to_string(), + columns: Some(vec![ColumnDiff { + diff_type: "added".to_string(), + name: "status".to_string(), + source: Some(ColumnInfo { + name: "status".to_string(), + data_type: "text".to_string(), + is_nullable: true, + column_default: None, + is_primary_key: false, + extra: None, + comment: None, + numeric_precision: None, + numeric_scale: None, + character_maximum_length: None, + }), + target: None, + changes: Vec::new(), + }]), + indexes: Some(vec![IndexDiff { + diff_type: "added".to_string(), + name: "idx_orders_status".to_string(), + source: Some(index(IndexInfo { + name: "idx_orders_status".to_string(), + columns: vec!["status".to_string()], + is_unique: false, + is_primary: false, + filter: None, + index_type: None, + included_columns: None, + comment: None, + })), + target: None, + changes: Vec::new(), + }]), + foreign_keys: None, + triggers: None, + ddl: None, + source_table_comment: None, + target_table_comment: None, + }]; + + assert_eq!( + generate_schema_sync_sql(&diffs, DatabaseType::Postgres, Some("sales")), + [ + "-- Alter table: orders", + "ALTER TABLE \"sales\".\"orders\" ADD COLUMN \"status\" text;", + "", + "CREATE INDEX \"idx_orders_status\" ON \"sales\".\"orders\" (\"status\");", + ] + .join("\n") + ); + } +} diff --git a/crates/dbx-core/src/sql_dialect.rs b/crates/dbx-core/src/sql_dialect.rs new file mode 100644 index 0000000000..f93505d5fd --- /dev/null +++ b/crates/dbx-core/src/sql_dialect.rs @@ -0,0 +1,168 @@ +use crate::models::connection::DatabaseType; + +#[derive(Debug, Clone, Copy)] +pub struct TableSelectSqlOptions<'a> { + pub database_type: Option, + pub schema: Option<&'a str>, + pub table_name: &'a str, + pub columns: &'a [String], + pub order_columns: &'a [String], + pub limit: usize, +} + +pub fn build_count_table_sql(database_type: Option, schema: Option<&str>, table_name: &str) -> String { + format!("SELECT COUNT(*) AS row_count FROM {}", qualified_table_name(database_type, schema, table_name)) +} + +pub fn build_table_select_sql(options: TableSelectSqlOptions<'_>) -> String { + let database_type = options.database_type; + let table = qualified_table_name(database_type, options.schema, options.table_name); + let select_columns = if options.columns.is_empty() { + "*".to_string() + } else { + options + .columns + .iter() + .map(|column| quote_table_identifier(database_type, column)) + .collect::>() + .join(", ") + }; + let order_by = if options.order_columns.is_empty() { + String::new() + } else { + format!( + " ORDER BY {}", + options + .order_columns + .iter() + .map(|column| format!("{} ASC", quote_table_identifier(database_type, column))) + .collect::>() + .join(", ") + ) + }; + let limit = options.limit; + + if database_type.is_some_and(uses_fetch_first) { + return format!("SELECT {select_columns} FROM {table}{order_by} FETCH FIRST {limit} ROWS ONLY"); + } + + if database_type == Some(DatabaseType::SqlServer) { + return format!("SELECT TOP ({limit}) {select_columns} FROM {table}{order_by}"); + } + + format!("SELECT {select_columns} FROM {table}{order_by} LIMIT {limit};") +} + +pub fn qualified_table_name(database_type: Option, schema: Option<&str>, table_name: &str) -> String { + if database_type.is_some_and(is_schema_aware) && schema.is_some_and(|schema| !schema.trim().is_empty()) { + return format!( + "{}.{}", + quote_table_identifier(database_type, schema.unwrap()), + quote_table_identifier(database_type, table_name) + ); + } + quote_table_identifier(database_type, table_name) +} + +pub fn quote_table_identifier(database_type: Option, name: &str) -> String { + match database_type { + Some(DatabaseType::Mysql | DatabaseType::Hive | DatabaseType::Tdengine | DatabaseType::Access) => { + format!("`{}`", name.replace('`', "``")) + } + Some(DatabaseType::Informix) if is_simple_informix_identifier(name) => name.to_string(), + Some(DatabaseType::Neo4j) => format!("`{}`", name.replace('`', "``")), + Some(DatabaseType::SqlServer) => format!("[{}]", name.replace(']', "]]")), + _ => format!("\"{}\"", name.replace('"', "\"\"")), + } +} + +pub fn is_schema_aware(database_type: DatabaseType) -> bool { + matches!( + database_type, + DatabaseType::Postgres + | DatabaseType::SqlServer + | DatabaseType::Oracle + | DatabaseType::Redshift + | DatabaseType::Dameng + | DatabaseType::Gaussdb + | DatabaseType::Kingbase + | DatabaseType::Highgo + | DatabaseType::Vastbase + | DatabaseType::Yashandb + | DatabaseType::Databricks + | DatabaseType::SapHana + | DatabaseType::Teradata + | DatabaseType::Vertica + | DatabaseType::Exasol + | DatabaseType::OpenGauss + | DatabaseType::OceanbaseOracle + | DatabaseType::Gbase + | DatabaseType::Jdbc + | DatabaseType::H2 + | DatabaseType::Snowflake + | DatabaseType::Trino + | DatabaseType::Db2 + | DatabaseType::Tdengine + ) +} + +pub fn uses_fetch_first(database_type: DatabaseType) -> bool { + matches!(database_type, DatabaseType::Oracle | DatabaseType::Dameng) +} + +fn is_simple_informix_identifier(name: &str) -> bool { + let mut chars = name.chars(); + let Some(first) = chars.next() else { + return false; + }; + (first.is_ascii_alphabetic() || first == '_') + && chars.all(|ch| ch.is_ascii_alphanumeric() || ch == '_' || ch == '$') +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn quotes_identifiers_by_database_type() { + assert_eq!(quote_table_identifier(Some(DatabaseType::Mysql), "user`name"), "`user``name`"); + assert_eq!(quote_table_identifier(Some(DatabaseType::SqlServer), "user]name"), "[user]]name]"); + assert_eq!(quote_table_identifier(Some(DatabaseType::Postgres), "user\"name"), "\"user\"\"name\""); + assert_eq!(quote_table_identifier(Some(DatabaseType::Informix), "users_1"), "users_1"); + } + + #[test] + fn qualifies_schema_only_for_schema_aware_databases() { + assert_eq!(qualified_table_name(Some(DatabaseType::Postgres), Some("public"), "users"), "\"public\".\"users\""); + assert_eq!(qualified_table_name(Some(DatabaseType::Mysql), Some("public"), "users"), "`users`"); + } + + #[test] + fn builds_select_sql_with_limit_syntax_for_database_type() { + let columns = vec!["id".to_string(), "name".to_string()]; + let keys = vec!["id".to_string()]; + + assert_eq!( + build_table_select_sql(TableSelectSqlOptions { + database_type: Some(DatabaseType::Postgres), + schema: Some("public"), + table_name: "users", + columns: &columns, + order_columns: &keys, + limit: 100, + }), + "SELECT \"id\", \"name\" FROM \"public\".\"users\" ORDER BY \"id\" ASC LIMIT 100;" + ); + assert_eq!( + build_table_select_sql(TableSelectSqlOptions { + database_type: Some(DatabaseType::SqlServer), + schema: Some("dbo"), + table_name: "users", + columns: &columns, + order_columns: &keys, + limit: 100, + }), + "SELECT TOP (100) [id], [name] FROM [dbo].[users] ORDER BY [id] ASC" + ); + } +} diff --git a/crates/dbx-web/src/main.rs b/crates/dbx-web/src/main.rs index 2264b1d026..54bf6af838 100644 --- a/crates/dbx-web/src/main.rs +++ b/crates/dbx-web/src/main.rs @@ -112,6 +112,8 @@ async fn main() { .route("/schema/foreign-keys", get(routes::schema::list_foreign_keys)) .route("/schema/triggers", get(routes::schema::list_triggers)) .route("/schema/ddl", get(routes::schema::get_ddl)) + .route("/schema-diff/prepare", post(routes::schema_diff::prepare_schema_diff)) + .route("/schema-diff/generate-sync-sql", post(routes::schema_diff::generate_schema_sync_sql)) .route( "/schema/cache", post(routes::schema_cache::save_schema_cache).get(routes::schema_cache::load_schema_cache), @@ -126,6 +128,8 @@ async fn main() { .route("/query/analyze-sql-references", post(routes::query::analyze_sql_references)) .route("/query/analyze-editability", post(routes::query::analyze_editable_query_editability)) .route("/query/prepare-data-grid-save", post(routes::query::prepare_data_grid_save)) + .route("/data-compare/prepare", post(routes::data_compare::prepare_data_compare)) + .route("/data-compare/prepare-from-tables", post(routes::data_compare::prepare_data_compare_from_tables)) .route("/query/cancel", post(routes::query::cancel_query)) .route("/query/close-session", post(routes::query::close_query_session)) .route("/export/query-result-json", post(routes::text_export::export_query_result_json)) diff --git a/crates/dbx-web/src/routes/data_compare.rs b/crates/dbx-web/src/routes/data_compare.rs new file mode 100644 index 0000000000..ccade33e04 --- /dev/null +++ b/crates/dbx-web/src/routes/data_compare.rs @@ -0,0 +1,19 @@ +use axum::extract::State; +use axum::Json; +use std::sync::Arc; + +use crate::error::AppError; +use crate::state::WebState; + +pub async fn prepare_data_compare( + Json(options): Json, +) -> Result, AppError> { + dbx_core::data_compare::prepare_data_compare(options).map(Json).map_err(AppError) +} + +pub async fn prepare_data_compare_from_tables( + State(state): State>, + Json(options): Json, +) -> Result, AppError> { + dbx_core::data_compare::prepare_data_compare_from_tables(&state.app, options).await.map(Json).map_err(AppError) +} diff --git a/crates/dbx-web/src/routes/mod.rs b/crates/dbx-web/src/routes/mod.rs index e8e284454b..e0d98208d7 100644 --- a/crates/dbx-web/src/routes/mod.rs +++ b/crates/dbx-web/src/routes/mod.rs @@ -2,6 +2,7 @@ pub mod agents; pub mod ai; pub mod app_settings; pub mod connection; +pub mod data_compare; pub mod database_export; pub mod history; pub mod layout; @@ -12,6 +13,7 @@ pub mod redis; pub mod saved_sql; pub mod schema; pub mod schema_cache; +pub mod schema_diff; pub mod sql_file; pub mod table_import; pub mod text_export; diff --git a/crates/dbx-web/src/routes/schema_diff.rs b/crates/dbx-web/src/routes/schema_diff.rs new file mode 100644 index 0000000000..f8282d415a --- /dev/null +++ b/crates/dbx-web/src/routes/schema_diff.rs @@ -0,0 +1,20 @@ +use axum::Json; +use serde::Deserialize; + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct GenerateSchemaSyncSqlRequest { + pub diffs: Vec, + pub database_type: dbx_core::models::connection::DatabaseType, + pub target_schema: Option, +} + +pub async fn prepare_schema_diff( + Json(options): Json, +) -> Json { + Json(dbx_core::schema_diff::prepare_schema_diff(options)) +} + +pub async fn generate_schema_sync_sql(Json(req): Json) -> Json { + Json(dbx_core::schema_diff::generate_schema_sync_sql(&req.diffs, req.database_type, req.target_schema.as_deref())) +} diff --git a/packages/app-tests/dataCompare.test.ts b/packages/app-tests/dataCompare.test.ts deleted file mode 100644 index 90d9ca77fc..0000000000 --- a/packages/app-tests/dataCompare.test.ts +++ /dev/null @@ -1,87 +0,0 @@ -import { strict as assert } from "node:assert"; -import test from "node:test"; -import { compareDataRows, generateDataSyncSql } from "../../apps/desktop/src/lib/dataCompare.ts"; - -test("compares rows by primary key and reports added, removed, and modified rows", () => { - const diff = compareDataRows({ - columns: ["id", "name", "active"], - keyColumns: ["id"], - sourceRows: [ - [1, "Ada", true], - [2, "Bob", false], - [4, "Dora", true], - ], - targetRows: [ - [1, "Ada", true], - [2, "Bobby", false], - [3, "Cara", true], - ], - }); - - assert.deepEqual( - diff.added.map((row) => row.keyValues), - [{ id: 4 }], - ); - assert.deepEqual( - diff.removed.map((row) => row.keyValues), - [{ id: 3 }], - ); - assert.deepEqual( - diff.modified.map((row) => row.changes), - [[{ column: "name", source: "Bob", target: "Bobby" }]], - ); -}); - -test("generates data synchronization SQL", () => { - const diff = compareDataRows({ - columns: ["id", "name", "active"], - keyColumns: ["id"], - sourceRows: [ - [1, "Ada", true], - [2, "Bob", false], - ], - targetRows: [ - [1, "Ada Lovelace", true], - [3, "Cara", true], - ], - }); - - assert.equal( - generateDataSyncSql({ - tableName: "users", - schema: "public", - columns: ["id", "name", "active"], - keyColumns: ["id"], - diff, - databaseType: "postgres", - }), - [ - `INSERT INTO "public"."users" ("id", "name", "active") VALUES (2, 'Bob', FALSE);`, - `UPDATE "public"."users" SET "name" = 'Ada' WHERE "id" = 1;`, - `DELETE FROM "public"."users" WHERE "id" = 3;`, - ].join("\n"), - ); -}); - -test("requires at least one key column", () => { - assert.throws( - () => compareDataRows({ columns: ["id"], keyColumns: [], sourceRows: [[1]], targetRows: [[1]] }), - /At least one key column/, - ); -}); - -test("rejects duplicate row keys", () => { - assert.throws( - () => - compareDataRows({ - columns: ["id", "name"], - keyColumns: ["id"], - sourceRows: [ - [1, "Ada"], - [1, "Ada Clone"], - ], - targetRows: [[1, "Ada"]], - }), - /Duplicate source key/, - ); -}); diff --git a/packages/app-tests/dataCompareBackend.test.ts b/packages/app-tests/dataCompareBackend.test.ts new file mode 100644 index 0000000000..06c30adf62 --- /dev/null +++ b/packages/app-tests/dataCompareBackend.test.ts @@ -0,0 +1,52 @@ +import { readFileSync } from "node:fs"; +import { strict as assert } from "node:assert"; +import test from "node:test"; + +const apiSource = readFileSync("apps/desktop/src/lib/api.ts", "utf8"); +const tauriSource = readFileSync("apps/desktop/src/lib/tauri.ts", "utf8"); +const httpSource = readFileSync("apps/desktop/src/lib/http.ts", "utf8"); +const dataCompareDialogSource = readFileSync("apps/desktop/src/components/diff/DataCompareDialog.vue", "utf8"); +const dataCompareSource = readFileSync("apps/desktop/src/lib/dataCompare.ts", "utf8"); +const tauriCommandsSource = readFileSync("src-tauri/src/commands/mod.rs", "utf8"); +const tauriLibSource = readFileSync("src-tauri/src/lib.rs", "utf8"); +const webRoutesSource = readFileSync("crates/dbx-web/src/routes/mod.rs", "utf8"); +const webMainSource = readFileSync("crates/dbx-web/src/main.rs", "utf8"); +const rustCoreLibSource = readFileSync("crates/dbx-core/src/lib.rs", "utf8"); + +test("shared API exposes backend data compare preparation", () => { + assert.match(apiSource, /export const prepareDataCompare = forward\("prepareDataCompare"\)/); + assert.match(apiSource, /export const prepareDataCompareFromTables = forward\("prepareDataCompareFromTables"\)/); + assert.match(tauriSource, /export async function prepareDataCompare\(/); + assert.match(tauriSource, /invoke\("prepare_data_compare"/); + assert.match(tauriSource, /export async function prepareDataCompareFromTables\(/); + assert.match(tauriSource, /invoke\("prepare_data_compare_from_tables"/); + assert.match(httpSource, /export async function prepareDataCompare\(/); + assert.match(httpSource, /\/api\/data-compare\/prepare/); + assert.match(httpSource, /export async function prepareDataCompareFromTables\(/); + assert.match(httpSource, /\/api\/data-compare\/prepare-from-tables/); +}); + +test("data compare dialog delegates comparison and sync SQL generation to backend API", () => { + assert.match(dataCompareDialogSource, /await api\.prepareDataCompareFromTables\(/); + assert.doesNotMatch(dataCompareDialogSource, /await api\.prepareDataCompare\(/); + assert.doesNotMatch(dataCompareDialogSource, /buildTableSelectSql/); + assert.doesNotMatch(dataCompareDialogSource, /compareDataRows\(/); + assert.doesNotMatch(dataCompareDialogSource, /generateDataSyncStatements\(/); + assert.doesNotMatch(dataCompareDialogSource, /generateDataSyncSql\(/); +}); + +test("frontend data compare module no longer owns executable compare or SQL logic", () => { + assert.doesNotMatch(dataCompareSource, /export function compareDataRows/); + assert.doesNotMatch(dataCompareSource, /export function generateDataSyncStatements/); + assert.doesNotMatch(dataCompareSource, /export function generateDataSyncSql/); +}); + +test("Rust backends register data compare APIs", () => { + assert.match(rustCoreLibSource, /pub mod data_compare/); + assert.match(tauriCommandsSource, /pub mod data_compare/); + assert.match(tauriLibSource, /commands::data_compare::prepare_data_compare/); + assert.match(tauriLibSource, /commands::data_compare::prepare_data_compare_from_tables/); + assert.match(webRoutesSource, /pub mod data_compare/); + assert.match(webMainSource, /\/data-compare\/prepare/); + assert.match(webMainSource, /\/data-compare\/prepare-from-tables/); +}); diff --git a/packages/app-tests/schemaDiff.test.ts b/packages/app-tests/schemaDiff.test.ts deleted file mode 100644 index 88d9cf7cd6..0000000000 --- a/packages/app-tests/schemaDiff.test.ts +++ /dev/null @@ -1,126 +0,0 @@ -import { strict as assert } from "node:assert"; -import test from "node:test"; -import { diffForeignKeys, diffIndexes, generateSyncSql, type TableDiff } from "../../apps/desktop/src/lib/schemaDiff.ts"; -import type { ForeignKeyInfo, IndexInfo } from "../../apps/desktop/src/types/database.ts"; - -function index(overrides: Partial): IndexInfo { - return { - name: "idx_users_email", - columns: ["email"], - is_unique: false, - is_primary: false, - ...overrides, - }; -} - -function foreignKey(overrides: Partial): ForeignKeyInfo { - return { - name: "orders_user_id_fk", - column: "user_id", - ref_table: "users", - ref_column: "id", - ...overrides, - }; -} - -test("detects modified indexes, not only added or removed indexes", () => { - const diffs = diffIndexes( - [index({ name: "idx_orders_status", columns: ["status", "created_at"], is_unique: false })], - [index({ name: "idx_orders_status", columns: ["status"], is_unique: true })], - ); - - assert.equal(diffs.length, 1); - assert.equal(diffs[0].type, "modified"); - assert.deepEqual(diffs[0].changes, ["unique: YES → NO", "columns: status → status, created_at"]); -}); - -test("detects foreign key additions, removals, and target changes", () => { - const diffs = diffForeignKeys( - [ - foreignKey({ name: "orders_user_id_fk" }), - foreignKey({ name: "orders_account_id_fk", column: "account_id", ref_table: "accounts" }), - ], - [ - foreignKey({ name: "orders_user_id_fk", ref_table: "members" }), - foreignKey({ name: "orders_region_id_fk", column: "region_id", ref_table: "regions" }), - ], - ); - - assert.deepEqual( - diffs.map((diff) => [diff.type, diff.name]), - [ - ["modified", "orders_user_id_fk"], - ["added", "orders_account_id_fk"], - ["removed", "orders_region_id_fk"], - ], - ); -}); - -test("generates sync SQL for index and foreign key changes", () => { - const diffs: TableDiff[] = [ - { - type: "modified", - name: "orders", - indexes: [ - { - type: "modified", - name: "idx_orders_status", - source: index({ name: "idx_orders_status", columns: ["status", "created_at"], is_unique: true }), - }, - ], - foreignKeys: [ - { - type: "modified", - name: "orders_user_id_fk", - source: foreignKey({ name: "orders_user_id_fk", ref_table: "users" }), - }, - ], - }, - ]; - - assert.equal( - generateSyncSql(diffs, "postgres"), - [ - 'ALTER TABLE "orders" DROP CONSTRAINT "orders_user_id_fk";', - 'DROP INDEX IF EXISTS "idx_orders_status";', - 'CREATE UNIQUE INDEX "idx_orders_status" ON "orders" ("status", "created_at");', - 'ALTER TABLE "orders" ADD CONSTRAINT "orders_user_id_fk" FOREIGN KEY ("user_id") REFERENCES "users" ("id");', - ].join("\n"), - ); -}); - -test("qualifies generated schema sync SQL with target schema", () => { - const diffs: TableDiff[] = [ - { - type: "modified", - name: "orders", - columns: [ - { - type: "added", - name: "status", - source: { - name: "status", - data_type: "text", - is_nullable: true, - column_default: null, - is_primary_key: false, - extra: null, - }, - }, - ], - indexes: [ - { type: "added", name: "idx_orders_status", source: index({ name: "idx_orders_status", columns: ["status"] }) }, - ], - }, - ]; - - assert.equal( - generateSyncSql(diffs, "postgres", "sales"), - [ - "-- Alter table: orders", - 'ALTER TABLE "sales"."orders" ADD COLUMN "status" text;', - "", - 'CREATE INDEX "idx_orders_status" ON "sales"."orders" ("status");', - ].join("\n"), - ); -}); diff --git a/packages/app-tests/schemaDiffBackend.test.ts b/packages/app-tests/schemaDiffBackend.test.ts new file mode 100644 index 0000000000..9e23f34abd --- /dev/null +++ b/packages/app-tests/schemaDiffBackend.test.ts @@ -0,0 +1,52 @@ +import { readFileSync } from "node:fs"; +import { strict as assert } from "node:assert"; +import test from "node:test"; + +const apiSource = readFileSync("apps/desktop/src/lib/api.ts", "utf8"); +const tauriSource = readFileSync("apps/desktop/src/lib/tauri.ts", "utf8"); +const httpSource = readFileSync("apps/desktop/src/lib/http.ts", "utf8"); +const schemaDiffDialogSource = readFileSync("apps/desktop/src/components/diff/SchemaDiffDialog.vue", "utf8"); +const schemaDiffSource = readFileSync("apps/desktop/src/lib/schemaDiff.ts", "utf8"); +const tauriCommandsSource = readFileSync("src-tauri/src/commands/mod.rs", "utf8"); +const tauriLibSource = readFileSync("src-tauri/src/lib.rs", "utf8"); +const webRoutesSource = readFileSync("crates/dbx-web/src/routes/mod.rs", "utf8"); +const webMainSource = readFileSync("crates/dbx-web/src/main.rs", "utf8"); +const rustCoreLibSource = readFileSync("crates/dbx-core/src/lib.rs", "utf8"); + +test("shared API exposes backend schema diff preparation and SQL generation", () => { + assert.match(apiSource, /export const prepareSchemaDiff = forward\("prepareSchemaDiff"\)/); + assert.match(apiSource, /export const generateSchemaSyncSql = forward\("generateSchemaSyncSql"\)/); + assert.match(tauriSource, /export async function prepareSchemaDiff\(/); + assert.match(tauriSource, /invoke\("prepare_schema_diff"/); + assert.match(tauriSource, /export async function generateSchemaSyncSql\(/); + assert.match(tauriSource, /invoke\("generate_schema_sync_sql"/); + assert.match(httpSource, /export async function prepareSchemaDiff\(/); + assert.match(httpSource, /\/api\/schema-diff\/prepare/); + assert.match(httpSource, /export async function generateSchemaSyncSql\(/); + assert.match(httpSource, /\/api\/schema-diff\/generate-sync-sql/); +}); + +test("schema diff dialog delegates diff and SQL generation to backend APIs", () => { + assert.match(schemaDiffDialogSource, /await api\.prepareSchemaDiff\(/); + assert.match(schemaDiffDialogSource, /await api\.generateSchemaSyncSql\(/); + assert.doesNotMatch(schemaDiffDialogSource, /diffColumns\(/); + assert.doesNotMatch(schemaDiffDialogSource, /diffIndexes\(/); + assert.doesNotMatch(schemaDiffDialogSource, /diffForeignKeys\(/); + assert.doesNotMatch(schemaDiffDialogSource, /diffTriggers\(/); + assert.doesNotMatch(schemaDiffDialogSource, /generateSyncSql\(/); +}); + +test("frontend schema diff module no longer owns executable diff logic", () => { + assert.doesNotMatch(schemaDiffSource, /export function diffColumns/); + assert.doesNotMatch(schemaDiffSource, /export function generateSyncSql/); +}); + +test("Rust backends register schema diff APIs", () => { + assert.match(rustCoreLibSource, /pub mod schema_diff/); + assert.match(tauriCommandsSource, /pub mod schema_diff/); + assert.match(tauriLibSource, /commands::schema_diff::prepare_schema_diff/); + assert.match(tauriLibSource, /commands::schema_diff::generate_schema_sync_sql/); + assert.match(webRoutesSource, /pub mod schema_diff/); + assert.match(webMainSource, /\/schema-diff\/prepare/); + assert.match(webMainSource, /\/schema-diff\/generate-sync-sql/); +}); diff --git a/src-tauri/src/commands/data_compare.rs b/src-tauri/src/commands/data_compare.rs new file mode 100644 index 0000000000..be79696edc --- /dev/null +++ b/src-tauri/src/commands/data_compare.rs @@ -0,0 +1,19 @@ +use std::sync::Arc; +use tauri::State; + +use crate::commands::connection::AppState; + +#[tauri::command] +pub fn prepare_data_compare( + options: dbx_core::data_compare::DataComparePreparationOptions, +) -> Result { + dbx_core::data_compare::prepare_data_compare(options) +} + +#[tauri::command] +pub async fn prepare_data_compare_from_tables( + state: State<'_, Arc>, + options: dbx_core::data_compare::DataCompareFromTablesOptions, +) -> Result { + dbx_core::data_compare::prepare_data_compare_from_tables(&state, options).await +} diff --git a/src-tauri/src/commands/mod.rs b/src-tauri/src/commands/mod.rs index e654eb8899..86a3647182 100644 --- a/src-tauri/src/commands/mod.rs +++ b/src-tauri/src/commands/mod.rs @@ -5,6 +5,7 @@ pub mod connection; #[allow(dead_code, unused_imports)] mod connection_secrets; pub mod csv_export; +pub mod data_compare; pub mod database_export; pub mod deep_link; pub mod external_sql; @@ -18,6 +19,7 @@ pub mod redis_cmd; pub mod saved_sql; pub mod schema; pub mod schema_cache; +pub mod schema_diff; pub mod sql_file; pub mod system_fonts; pub mod table_import; diff --git a/src-tauri/src/commands/schema_diff.rs b/src-tauri/src/commands/schema_diff.rs new file mode 100644 index 0000000000..5d8adba9b1 --- /dev/null +++ b/src-tauri/src/commands/schema_diff.rs @@ -0,0 +1,15 @@ +#[tauri::command] +pub fn prepare_schema_diff( + options: dbx_core::schema_diff::SchemaDiffPreparationOptions, +) -> Result { + Ok(dbx_core::schema_diff::prepare_schema_diff(options)) +} + +#[tauri::command] +pub fn generate_schema_sync_sql( + diffs: Vec, + database_type: dbx_core::models::connection::DatabaseType, + target_schema: Option, +) -> Result { + Ok(dbx_core::schema_diff::generate_schema_sync_sql(&diffs, database_type, target_schema.as_deref())) +} diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 21bc3b0f4b..4c3118a5f7 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -277,6 +277,8 @@ pub fn run() { commands::schema::list_foreign_keys, commands::schema::list_triggers, commands::schema::get_table_ddl, + commands::schema_diff::prepare_schema_diff, + commands::schema_diff::generate_schema_sync_sql, commands::schema_cache::save_schema_cache, commands::schema_cache::load_schema_cache, commands::schema_cache::delete_schema_cache_prefix, @@ -290,6 +292,8 @@ pub fn run() { commands::query::analyze_sql_references, commands::query::analyze_editable_query_editability, commands::query::prepare_data_grid_save, + commands::data_compare::prepare_data_compare, + commands::data_compare::prepare_data_compare_from_tables, commands::sql_file::preview_sql_file, commands::sql_file::execute_sql_file, commands::sql_file::cancel_sql_file_execution,