diff --git a/apps/desktop/src/components/diff/SchemaDiffDialog.vue b/apps/desktop/src/components/diff/SchemaDiffDialog.vue index 6282f7d19..f8e517e6e 100644 --- a/apps/desktop/src/components/diff/SchemaDiffDialog.vue +++ b/apps/desktop/src/components/diff/SchemaDiffDialog.vue @@ -15,10 +15,11 @@ import SchemaDiffDeployStep from "@/components/diff/SchemaDiffDeployStep.vue"; import SchemaDiffOptionsPanel from "@/components/diff/SchemaDiffOptionsPanel.vue"; import { getSchemaDiffOptionsForDbType } from "@/lib/schemaDiffOptions"; +import { createConcurrencyLimiter, mapWithConcurrency, schemaDiffMetadataConcurrency } from "@/lib/schemaDiffMetadataLoad"; import { normalizeSchemaDiffCompareOptions } from "@/types/schemaDiff"; import type { SchemaDiffCompareOptions, SchemaDiffConfig } from "@/types/schemaDiff"; -import type { ObjectSourceKind } from "@/types/database"; -import { buildDeploySqlForObjects, convertToSchemaDiffObjects, groupDiffObjects, type OperationGroup, type SchemaDiffObject, type DiffOperationType, type DiffObjectKind, type SchemaDiffPreparation } from "@/lib/schemaDiff"; +import type { ObjectSourceKind, TableInfo } from "@/types/database"; +import { buildDeploySqlForObjects, convertToSchemaDiffObjects, groupDiffObjects, type OperationGroup, type SchemaDiffObject, type DiffOperationType, type DiffObjectKind, type SchemaDiffPreparation, type TableSchemaDetail } from "@/lib/schemaDiff"; import { compileSchemaDiffTableFilter, filterSchemaDiffTables } from "@/lib/schemaDiffTableFilter"; import { Splitpanes, Pane } from "splitpanes"; import "splitpanes/dist/splitpanes.css"; @@ -279,13 +280,45 @@ function isViewOrMaterializedView(tableType: string): ObjectSourceKind | undefin } } +interface SchemaDetailLoadContext { + connectionId: string; + database: string; + schema: string; + dbType: string; + options: SchemaDiffCompareOptions; +} + +function shouldLoadIndexes(options: SchemaDiffCompareOptions): boolean { + return options.indexes || options.primaryKeys || options.uniqueKeys; +} + +async function loadSchemaDetails(tables: TableInfo[], context: SchemaDetailLoadContext): Promise { + const concurrency = schemaDiffMetadataConcurrency(context.dbType, tables.length); + const runMetadataQuery = createConcurrencyLimiter(concurrency); + + return mapWithConcurrency(tables, concurrency, async (table) => { + const objectType = isViewOrMaterializedView(table.table_type); + const [columns, indexes, foreignKeys, triggers, ddl] = await Promise.all([ + runMetadataQuery(() => api.getColumns(context.connectionId, context.database, context.schema, table.name)), + shouldLoadIndexes(context.options) ? runMetadataQuery(() => api.listIndexes(context.connectionId, context.database, context.schema, table.name)) : Promise.resolve([]), + context.options.foreignKeys ? runMetadataQuery(() => api.listForeignKeys(context.connectionId, context.database, context.schema, table.name)) : Promise.resolve([]), + context.options.triggers ? runMetadataQuery(() => api.listTriggers(context.connectionId, context.database, context.schema, table.name)) : Promise.resolve([]), + runMetadataQuery(() => api.getTableDdl(context.connectionId, context.database, context.schema, table.name, objectType)), + ]); + + return { name: table.name, columns, indexes, foreignKeys, triggers, ddl }; + }); +} + async function handleCompare() { loading.value = true; step.value = "compare"; try { + const sourceConfig = store.getConfig(sourceConnectionId.value); const targetConfig = store.getConfig(targetConnectionId.value); const dbType = targetConfig?.db_type || "mysql"; + const sourceDbType = sourceConfig?.db_type || dbType; const opts = normalizeSchemaDiffCompareOptions(activeConfig.value?.options, dbType); const tableFilter = compileSchemaDiffTableFilter(opts); @@ -295,34 +328,21 @@ async function handleCompare() { const [srcTables, tgtTables] = await Promise.all([api.listTables(sourceConnectionId.value, sourceDatabase.value, sourceSchema.value), api.listTables(targetConnectionId.value, targetDatabase.value, targetSchema.value)]); const { sourceTables, targetTables } = filterSchemaDiffTables(srcTables, tgtTables, tableFilter); - // Load schema details in parallel - const sourceDetails = await Promise.all( - sourceTables.map(async (table) => { - const objectType = isViewOrMaterializedView(table.table_type); - const [columns, indexes, foreignKeys, triggers, ddl] = await Promise.all([ - api.getColumns(sourceConnectionId.value, sourceDatabase.value, sourceSchema.value, table.name), - api.listIndexes(sourceConnectionId.value, sourceDatabase.value, sourceSchema.value, table.name), - api.listForeignKeys(sourceConnectionId.value, sourceDatabase.value, sourceSchema.value, table.name), - api.listTriggers(sourceConnectionId.value, sourceDatabase.value, sourceSchema.value, table.name), - api.getTableDdl(sourceConnectionId.value, sourceDatabase.value, sourceSchema.value, table.name, objectType), - ]); - return { name: table.name, columns, indexes, foreignKeys, triggers, ddl }; - }), - ); + const sourceDetails = await loadSchemaDetails(sourceTables, { + connectionId: sourceConnectionId.value, + database: sourceDatabase.value, + schema: sourceSchema.value, + dbType: sourceDbType, + options: opts, + }); - const targetDetails = await Promise.all( - targetTables.map(async (table) => { - const objectType = isViewOrMaterializedView(table.table_type); - const [columns, indexes, foreignKeys, triggers, ddl] = await Promise.all([ - api.getColumns(targetConnectionId.value, targetDatabase.value, targetSchema.value, table.name), - api.listIndexes(targetConnectionId.value, targetDatabase.value, targetSchema.value, table.name), - api.listForeignKeys(targetConnectionId.value, targetDatabase.value, targetSchema.value, table.name), - api.listTriggers(targetConnectionId.value, targetDatabase.value, targetSchema.value, table.name), - api.getTableDdl(targetConnectionId.value, targetDatabase.value, targetSchema.value, table.name, objectType), - ]); - return { name: table.name, columns, indexes, foreignKeys, triggers, ddl }; - }), - ); + const targetDetails = await loadSchemaDetails(targetTables, { + connectionId: targetConnectionId.value, + database: targetDatabase.value, + schema: targetSchema.value, + dbType, + options: opts, + }); const isPostgresLike = dbType === "postgres" || dbType === "opengauss"; diff --git a/apps/desktop/src/lib/__tests__/schemaDiffMetadataLoad.spec.ts b/apps/desktop/src/lib/__tests__/schemaDiffMetadataLoad.spec.ts new file mode 100644 index 000000000..7dcde6ae8 --- /dev/null +++ b/apps/desktop/src/lib/__tests__/schemaDiffMetadataLoad.spec.ts @@ -0,0 +1,70 @@ +import { describe, expect, it } from "vitest"; +import { createConcurrencyLimiter, mapWithConcurrency, schemaDiffMetadataConcurrency } from "@/lib/schemaDiffMetadataLoad"; + +function wait(ms: number) { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +describe("schemaDiffMetadataLoad", () => { + it("uses adaptive metadata concurrency for MySQL-compatible databases", () => { + expect(schemaDiffMetadataConcurrency("mysql")).toBe(2); + expect(schemaDiffMetadataConcurrency("MariaDB")).toBe(2); + expect(schemaDiffMetadataConcurrency("mysql", 30)).toBe(4); + expect(schemaDiffMetadataConcurrency("mysql", 31)).toBe(2); + expect(schemaDiffMetadataConcurrency("MariaDB", 12)).toBe(4); + expect(schemaDiffMetadataConcurrency("postgres")).toBe(6); + expect(schemaDiffMetadataConcurrency("postgres", 100)).toBe(6); + expect(schemaDiffMetadataConcurrency(undefined)).toBe(6); + }); + + it("maps items with limited concurrency and preserves output order", async () => { + let active = 0; + let maxActive = 0; + + const result = await mapWithConcurrency([30, 5, 10, 1], 2, async (delay, index) => { + active += 1; + maxActive = Math.max(maxActive, active); + await wait(delay); + active -= 1; + return `${index}:${delay}`; + }); + + expect(result).toEqual(["0:30", "1:5", "2:10", "3:1"]); + expect(maxActive).toBeLessThanOrEqual(2); + }); + + it("propagates the first worker error and stops scheduling new work", async () => { + const started: number[] = []; + + await expect( + mapWithConcurrency([1, 2, 3], 1, async (item) => { + started.push(item); + if (item === 2) throw new Error("boom"); + return item; + }), + ).rejects.toThrow("boom"); + + expect(started).toEqual([1, 2]); + }); + + it("limits arbitrary async tasks", async () => { + const runLimited = createConcurrencyLimiter(2); + let active = 0; + let maxActive = 0; + + const result = await Promise.all( + [8, 6, 4, 2].map((delay, index) => + runLimited(async () => { + active += 1; + maxActive = Math.max(maxActive, active); + await wait(delay); + active -= 1; + return index; + }), + ), + ); + + expect(result).toEqual([0, 1, 2, 3]); + expect(maxActive).toBeLessThanOrEqual(2); + }); +}); diff --git a/apps/desktop/src/lib/schemaDiffMetadataLoad.ts b/apps/desktop/src/lib/schemaDiffMetadataLoad.ts new file mode 100644 index 000000000..8973ffcc0 --- /dev/null +++ b/apps/desktop/src/lib/schemaDiffMetadataLoad.ts @@ -0,0 +1,83 @@ +const MYSQL_LARGE_SCHEMA_DIFF_METADATA_CONCURRENCY = 2; +const MYSQL_SMALL_SCHEMA_DIFF_METADATA_CONCURRENCY = 4; +const MYSQL_SMALL_SCHEMA_TABLE_LIMIT = 30; +const DEFAULT_SCHEMA_DIFF_METADATA_CONCURRENCY = 6; + +function normalizeConcurrencyLimit(limit: number): number { + return Number.isFinite(limit) ? Math.max(1, Math.floor(limit)) : 1; +} + +export function schemaDiffMetadataConcurrency(dbType: string | null | undefined, tableCount?: number): number { + const normalizedDbType = (dbType || "").toLowerCase(); + if (normalizedDbType === "mysql" || normalizedDbType === "mariadb") { + if (typeof tableCount === "number" && tableCount <= MYSQL_SMALL_SCHEMA_TABLE_LIMIT) { + return MYSQL_SMALL_SCHEMA_DIFF_METADATA_CONCURRENCY; + } + return MYSQL_LARGE_SCHEMA_DIFF_METADATA_CONCURRENCY; + } + return DEFAULT_SCHEMA_DIFF_METADATA_CONCURRENCY; +} + +export async function mapWithConcurrency(items: readonly T[], limit: number, worker: (item: T, index: number) => Promise): Promise { + const workerCount = Math.min(normalizeConcurrencyLimit(limit), items.length); + if (workerCount === 0) return []; + + const results: R[] = []; + let nextIndex = 0; + let hasError = false; + let firstError: unknown; + + async function runWorker() { + while (!hasError) { + const index = nextIndex++; + if (index >= items.length) return; + + try { + results[index] = await worker(items[index], index); + } catch (error) { + hasError = true; + firstError = error; + return; + } + } + } + + await Promise.all(Array.from({ length: workerCount }, runWorker)); + if (hasError) throw firstError; + return results; +} + +export function createConcurrencyLimiter(limit: number) { + const maxActive = normalizeConcurrencyLimit(limit); + let active = 0; + const queue: Array<() => void> = []; + + async function acquire() { + if (active < maxActive) { + active += 1; + return; + } + + await new Promise((resolve) => { + queue.push(() => { + active += 1; + resolve(); + }); + }); + } + + function release() { + active = Math.max(0, active - 1); + const next = queue.shift(); + if (next) next(); + } + + return async function runLimited(task: () => Promise): Promise { + await acquire(); + try { + return await task(); + } finally { + release(); + } + }; +}