diff --git a/tools/database/mysql2-schema-exporter.mjs b/tools/database/mysql2-schema-exporter.mjs index 7aa9b62..8593f64 100644 --- a/tools/database/mysql2-schema-exporter.mjs +++ b/tools/database/mysql2-schema-exporter.mjs @@ -2,6 +2,33 @@ import { createRequire } from 'node:module'; const require = createRequire(import.meta.url); const mysql2Version = require('mysql2/package.json').version; +const DEVELOPMENT_DATABASE = 'a_share_analysis_dev'; + +const CATALOG_QUERIES = Object.freeze({ + tables: `SELECT TABLE_NAME, TABLE_TYPE +FROM information_schema.TABLES +WHERE TABLE_SCHEMA = ? +ORDER BY TABLE_NAME`, + viewDependencies: `SELECT VIEW_NAME, TABLE_SCHEMA AS REFERENCED_SCHEMA_NAME, + TABLE_NAME AS REFERENCED_TABLE_NAME +FROM information_schema.VIEW_TABLE_USAGE +WHERE VIEW_SCHEMA = ? +ORDER BY VIEW_NAME, REFERENCED_SCHEMA_NAME, REFERENCED_TABLE_NAME`, + routines: `SELECT ROUTINE_NAME, ROUTINE_TYPE +FROM information_schema.ROUTINES +WHERE ROUTINE_SCHEMA = ? +ORDER BY ROUTINE_TYPE, ROUTINE_NAME`, + triggers: `SELECT TRIGGER_NAME, EVENT_OBJECT_TABLE, ACTION_TIMING, + EVENT_MANIPULATION, ACTION_ORDER +FROM information_schema.TRIGGERS +WHERE TRIGGER_SCHEMA = ? +ORDER BY EVENT_OBJECT_TABLE, ACTION_TIMING, EVENT_MANIPULATION, + ACTION_ORDER, TRIGGER_NAME`, + events: `SELECT EVENT_NAME +FROM information_schema.EVENTS +WHERE EVENT_SCHEMA = ? +ORDER BY EVENT_NAME`, +}); export const MYSQL2_EXPORTER_METADATA = Object.freeze({ name: 'mysql2-show-create', @@ -73,3 +100,159 @@ export function assertMysql57ServerMetadata(metadata) { throw new SchemaExportError('INVALID_DATABASE_COLLATION', '数据库排序规则必须为 utf8mb4_general_ci'); } } + +function requireCatalogText(value) { + if (typeof value !== 'string' || value.length === 0) { + throw new SchemaExportError('CATALOG_SHAPE_INVALID', 'Schema 对象目录字段无效'); + } + return value; +} + +async function queryCatalogRows(connection, sql, database, limits) { + const result = await connection.query({ + sql, + values: [database], + timeout: limits.queryTimeoutMs, + }); + if (!Array.isArray(result) || !Array.isArray(result[0])) { + throw new SchemaExportError('CATALOG_SHAPE_INVALID', 'Schema 对象目录查询结果无效'); + } + return result[0]; +} + +function assertObjectLimit(count, limits) { + if (!Number.isInteger(limits.maxObjects) || limits.maxObjects < 0 || count > limits.maxObjects) { + throw new SchemaExportError('OBJECT_LIMIT_EXCEEDED', 'Schema 对象数量超过安全上限'); + } +} + +/** + * 目录查询只从 information_schema 读取固定列,库名始终作为参数传入。每类对象 + * 在进入后续 SHOW CREATE 前先转换成稳定的小对象,并在累计数量越界时立即停止。 + */ +export async function readSchemaCatalog(connection, database, limits = DEFAULT_EXPORT_LIMITS) { + const tableRows = await queryCatalogRows(connection, CATALOG_QUERIES.tables, database, limits); + const tables = []; + const views = []; + for (const row of tableRows) { + const name = requireCatalogText(row.TABLE_NAME); + if (row.TABLE_TYPE === 'BASE TABLE') tables.push(name); + else if (row.TABLE_TYPE === 'VIEW') views.push(name); + else throw new SchemaExportError('UNKNOWN_TABLE_TYPE', 'Schema 目录包含未知表类型'); + } + tables.sort(); + views.sort(); + let objectCount = tables.length + views.length; + assertObjectLimit(objectCount, limits); + + const dependencyRows = await queryCatalogRows( + connection, + CATALOG_QUERIES.viewDependencies, + database, + limits, + ); + const viewDependencies = dependencyRows.map((row) => ({ + view: requireCatalogText(row.VIEW_NAME), + referencedSchema: requireCatalogText(row.REFERENCED_SCHEMA_NAME), + referencedObject: requireCatalogText(row.REFERENCED_TABLE_NAME), + })); + + const routineRows = await queryCatalogRows(connection, CATALOG_QUERIES.routines, database, limits); + const routines = routineRows.map((row) => { + const type = requireCatalogText(row.ROUTINE_TYPE); + if (!['FUNCTION', 'PROCEDURE'].includes(type)) { + throw new SchemaExportError('CATALOG_SHAPE_INVALID', 'Schema 目录包含未知例程类型'); + } + return { name: requireCatalogText(row.ROUTINE_NAME), type }; + }); + routines.sort((left, right) => left.type.localeCompare(right.type) || left.name.localeCompare(right.name)); + objectCount += routines.length; + assertObjectLimit(objectCount, limits); + + const triggerRows = await queryCatalogRows(connection, CATALOG_QUERIES.triggers, database, limits); + const triggers = triggerRows.map((row) => { + const order = Number(row.ACTION_ORDER); + if (!Number.isInteger(order) || order < 0) { + throw new SchemaExportError('CATALOG_SHAPE_INVALID', 'Schema 触发器执行顺序无效'); + } + return { + name: requireCatalogText(row.TRIGGER_NAME), + table: requireCatalogText(row.EVENT_OBJECT_TABLE), + timing: requireCatalogText(row.ACTION_TIMING), + event: requireCatalogText(row.EVENT_MANIPULATION), + order, + }; + }); + triggers.sort( + (left, right) => + left.table.localeCompare(right.table) || + left.timing.localeCompare(right.timing) || + left.event.localeCompare(right.event) || + left.order - right.order || + left.name.localeCompare(right.name), + ); + objectCount += triggers.length; + assertObjectLimit(objectCount, limits); + + const eventRows = await queryCatalogRows(connection, CATALOG_QUERIES.events, database, limits); + const events = eventRows.map((row) => requireCatalogText(row.EVENT_NAME)).sort(); + objectCount += events.length; + assertObjectLimit(objectCount, limits); + + return Object.freeze({ + tables: Object.freeze(tables), + views: Object.freeze(views), + viewDependencies: Object.freeze(viewDependencies.map((dependency) => Object.freeze(dependency))), + routines: Object.freeze(routines.map((routine) => Object.freeze(routine))), + triggers: Object.freeze(triggers.map((trigger) => Object.freeze(trigger))), + events: Object.freeze(events), + }); +} + +/** + * 视图恢复顺序使用稳定 Kahn 拓扑排序。跨库或未知依赖不能被默默忽略,否则恢复 + * 出来的 Schema 可能引用未捕获对象;同层节点始终按名称排序以保证逐字节确定性。 + */ +export function orderViewNames(catalog) { + const tables = new Set(catalog.tables ?? []); + const views = new Set(catalog.views ?? []); + const incoming = new Map([...views].map((view) => [view, 0])); + const dependents = new Map([...views].map((view) => [view, []])); + + for (const dependency of catalog.viewDependencies ?? []) { + if (dependency.referencedSchema !== DEVELOPMENT_DATABASE) { + throw new SchemaExportError( + 'CROSS_SCHEMA_VIEW_DEPENDENCY', + 'Schema 视图包含跨库依赖', + ); + } + if (!views.has(dependency.view)) { + throw new SchemaExportError('UNKNOWN_VIEW_DEPENDENCY', 'Schema 视图依赖来源未知'); + } + if (tables.has(dependency.referencedObject)) continue; + if (!views.has(dependency.referencedObject)) { + throw new SchemaExportError('UNKNOWN_VIEW_DEPENDENCY', 'Schema 视图依赖对象未知'); + } + incoming.set(dependency.view, incoming.get(dependency.view) + 1); + dependents.get(dependency.referencedObject).push(dependency.view); + } + + for (const names of dependents.values()) names.sort(); + const ready = [...views].filter((view) => incoming.get(view) === 0).sort(); + const ordered = []; + while (ready.length > 0) { + const view = ready.shift(); + ordered.push(view); + for (const dependent of dependents.get(view)) { + incoming.set(dependent, incoming.get(dependent) - 1); + if (incoming.get(dependent) === 0) { + ready.push(dependent); + ready.sort(); + } + } + } + if (ordered.length !== views.size) { + throw new SchemaExportError('VIEW_DEPENDENCY_CYCLE', 'Schema 视图依赖存在循环'); + } + return ordered; +} diff --git a/tools/database/mysql2-schema-exporter.test.mjs b/tools/database/mysql2-schema-exporter.test.mjs index 836eecd..98f26ac 100644 --- a/tools/database/mysql2-schema-exporter.test.mjs +++ b/tools/database/mysql2-schema-exporter.test.mjs @@ -7,7 +7,9 @@ import { SchemaExportError, assertMysql57ServerMetadata, buildConnectionOptions, + orderViewNames, quoteIdentifier, + readSchemaCatalog, } from './mysql2-schema-exporter.mjs'; const config = Object.freeze({ @@ -57,3 +59,170 @@ test('只接受 MySQL 5.7.x 与固定字符规则', () => { assert.throws(() => assertMysql57ServerMetadata({ ...valid, serverVersion })); } }); + +function createCatalogConnection(overrides = {}) { + const rowsBySource = { + 'information_schema.TABLES': [ + { TABLE_NAME: 'market_snapshot', TABLE_TYPE: 'BASE TABLE' }, + { TABLE_NAME: 'market_dashboard', TABLE_TYPE: 'VIEW' }, + { TABLE_NAME: 'market_summary', TABLE_TYPE: 'VIEW' }, + ], + 'information_schema.VIEW_TABLE_USAGE': [ + { + VIEW_NAME: 'market_dashboard', + REFERENCED_SCHEMA_NAME: 'a_share_analysis_dev', + REFERENCED_TABLE_NAME: 'market_summary', + }, + { + VIEW_NAME: 'market_summary', + REFERENCED_SCHEMA_NAME: 'a_share_analysis_dev', + REFERENCED_TABLE_NAME: 'market_snapshot', + }, + ], + 'information_schema.ROUTINES': [ + { ROUTINE_NAME: 'calculate_breadth', ROUTINE_TYPE: 'FUNCTION' }, + { ROUTINE_NAME: 'refresh_market_snapshot', ROUTINE_TYPE: 'PROCEDURE' }, + ], + 'information_schema.TRIGGERS': [ + { + TRIGGER_NAME: 'snapshot_before_insert', + EVENT_OBJECT_TABLE: 'market_snapshot', + ACTION_TIMING: 'BEFORE', + EVENT_MANIPULATION: 'INSERT', + ACTION_ORDER: 1, + }, + ], + 'information_schema.EVENTS': [{ EVENT_NAME: 'purge_market_snapshot' }], + ...overrides, + }; + const calls = []; + return { + calls, + async query(options) { + calls.push(options); + const source = Object.keys(rowsBySource).find((candidate) => options.sql.includes(candidate)); + assert.ok(source, `未识别的目录查询:${options.sql}`); + return [rowsBySource[source], []]; + }, + }; +} + +test('目录查询使用固定参数和超时并返回稳定对象', async () => { + const connection = createCatalogConnection(); + const catalog = await readSchemaCatalog(connection, 'a_share_analysis_dev', DEFAULT_EXPORT_LIMITS); + + assert.deepEqual(catalog, { + tables: ['market_snapshot'], + views: ['market_dashboard', 'market_summary'], + viewDependencies: [ + { + view: 'market_dashboard', + referencedSchema: 'a_share_analysis_dev', + referencedObject: 'market_summary', + }, + { + view: 'market_summary', + referencedSchema: 'a_share_analysis_dev', + referencedObject: 'market_snapshot', + }, + ], + routines: [ + { name: 'calculate_breadth', type: 'FUNCTION' }, + { name: 'refresh_market_snapshot', type: 'PROCEDURE' }, + ], + triggers: [ + { + name: 'snapshot_before_insert', + table: 'market_snapshot', + timing: 'BEFORE', + event: 'INSERT', + order: 1, + }, + ], + events: ['purge_market_snapshot'], + }); + assert.equal(connection.calls.length, 5); + for (const call of connection.calls) { + assert.deepEqual(call.values, ['a_share_analysis_dev']); + assert.equal(call.timeout, DEFAULT_EXPORT_LIMITS.queryTimeoutMs); + } +}); + +test('视图按依赖拓扑排序且同层按名称稳定排序', () => { + const catalog = { + tables: ['market_snapshot'], + views: ['z_dashboard', 'market_summary', 'a_dashboard'], + viewDependencies: [ + { + view: 'z_dashboard', + referencedSchema: 'a_share_analysis_dev', + referencedObject: 'market_summary', + }, + { + view: 'a_dashboard', + referencedSchema: 'a_share_analysis_dev', + referencedObject: 'market_summary', + }, + { + view: 'market_summary', + referencedSchema: 'a_share_analysis_dev', + referencedObject: 'market_snapshot', + }, + ], + }; + + assert.deepEqual(orderViewNames(catalog), ['market_summary', 'a_dashboard', 'z_dashboard']); +}); + +test('视图排序拒绝循环、跨库和未知依赖', () => { + const base = { tables: ['market_snapshot'], views: ['first_view', 'second_view'] }; + const invalidCases = [ + { + code: 'VIEW_DEPENDENCY_CYCLE', + dependencies: [ + { view: 'first_view', referencedSchema: 'a_share_analysis_dev', referencedObject: 'second_view' }, + { view: 'second_view', referencedSchema: 'a_share_analysis_dev', referencedObject: 'first_view' }, + ], + }, + { + code: 'CROSS_SCHEMA_VIEW_DEPENDENCY', + dependencies: [ + { view: 'first_view', referencedSchema: 'production', referencedObject: 'market_snapshot' }, + ], + }, + { + code: 'UNKNOWN_VIEW_DEPENDENCY', + dependencies: [ + { view: 'first_view', referencedSchema: 'a_share_analysis_dev', referencedObject: 'missing_object' }, + ], + }, + ]; + + for (const invalid of invalidCases) { + assert.throws( + () => orderViewNames({ ...base, viewDependencies: invalid.dependencies }), + (error) => error instanceof SchemaExportError && error.code === invalid.code, + ); + } +}); + +test('目录拒绝未知表类型和超过对象上限', async () => { + const unknownType = createCatalogConnection({ + 'information_schema.TABLES': [{ TABLE_NAME: 'temporary_data', TABLE_TYPE: 'SYSTEM VIEW' }], + }); + await assert.rejects( + () => readSchemaCatalog(unknownType, 'a_share_analysis_dev', DEFAULT_EXPORT_LIMITS), + (error) => error instanceof SchemaExportError && error.code === 'UNKNOWN_TABLE_TYPE', + ); + + const overLimit = createCatalogConnection({ + 'information_schema.TABLES': [ + { TABLE_NAME: 'first_table', TABLE_TYPE: 'BASE TABLE' }, + { TABLE_NAME: 'second_table', TABLE_TYPE: 'BASE TABLE' }, + ], + }); + await assert.rejects( + () => readSchemaCatalog(overLimit, 'a_share_analysis_dev', { ...DEFAULT_EXPORT_LIMITS, maxObjects: 1 }), + (error) => error instanceof SchemaExportError && error.code === 'OBJECT_LIMIT_EXCEEDED', + ); +});