fix(schema-diff): reduce MySQL metadata query pressure
Co-authored-by: zipg <4047349+zipg@users.noreply.github.com>
This commit is contained in:
zipg
GitHub
parent
d65f656704
commit
5a8b129edf
|
|
@ -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<TableSchemaDetail[]> {
|
||||
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";
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,70 @@
|
|||
import { describe, expect, it } from "vitest";
|
||||
import { createConcurrencyLimiter, mapWithConcurrency, schemaDiffMetadataConcurrency } from "@/lib/schemaDiffMetadataLoad";
|
||||
|
||||
function wait(ms: number) {
|
||||
return new Promise<void>((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);
|
||||
});
|
||||
});
|
||||
|
|
@ -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<T, R>(items: readonly T[], limit: number, worker: (item: T, index: number) => Promise<R>): Promise<R[]> {
|
||||
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<void>((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<T>(task: () => Promise<T>): Promise<T> {
|
||||
await acquire();
|
||||
try {
|
||||
return await task();
|
||||
} finally {
|
||||
release();
|
||||
}
|
||||
};
|
||||
}
|
||||
Loading…
Reference in New Issue