diff --git a/frontend/src/components/data-sync/DataSyncPreflightPanel.test.tsx b/frontend/src/components/data-sync/DataSyncPreflightPanel.test.tsx index e3718416..5a920de0 100644 --- a/frontend/src/components/data-sync/DataSyncPreflightPanel.test.tsx +++ b/frontend/src/components/data-sync/DataSyncPreflightPanel.test.tsx @@ -2,14 +2,64 @@ import React from 'react'; import TestRenderer, { act } from 'react-test-renderer'; import { afterEach, describe, expect, it, vi } from 'vitest'; +const runtimeApi = vi.hoisted(() => ({ + ClipboardSetText: vi.fn(() => Promise.resolve(false)), +})); + +vi.mock('../../../wailsjs/runtime/runtime', () => runtimeApi); + import { DataSyncPreflightPanel } from './DataSyncPreflightPanel'; + +Object.assign(globalThis, { + navigator: { + clipboard: { + writeText: vi.fn(() => Promise.resolve()), + }, + }, +}); import { createDataSyncWorkbenchTranslate, dataSyncValidationIssueText, } from './text'; +const unmigratedIndexSnapshot = () => ({ + taskId: 'task-1', + taskRevision: 3, + status: 'warning' as const, + issues: [ + { + id: 'unmigrated_index:map-1:0', + code: 'unmigrated_index' as const, + severity: 'warning' as const, + stage: 'mappings' as const, + mappingId: 'map-1', + message: 'review remediation', + detail: { + unmigratedIndex: { + name: 'idx_name_prefix', + columns: [{ name: 'name', prefixLength: 12 }], + unique: false, + indexType: 'BTREE', + reason: 'review remediation', + remediationStatements: [ + 'CREATE INDEX idx_name_prefix ON public.users (left(name, 12))', + ], + }, + }, + }, + ], + definitionHash: 'hash-1', + approvalRequired: false, + approvalSatisfied: false, + checkedAt: '2030-08-08T00:00:00.000Z', +}); + afterEach(() => { vi.useRealTimers(); + runtimeApi.ClipboardSetText.mockReset(); + runtimeApi.ClipboardSetText.mockResolvedValue(false); + (navigator.clipboard.writeText as ReturnType).mockReset(); + (navigator.clipboard.writeText as ReturnType).mockResolvedValue(undefined); }); describe('DataSyncPreflightPanel production approval', () => { @@ -73,6 +123,64 @@ describe('DataSyncPreflightPanel production approval', () => { ); }); + it('renders unmigrated indexes and copies remediation DDL', async () => { + const writeText = navigator.clipboard.writeText as ReturnType; + const renderer = TestRenderer.create( + undefined} + />, + ); + + expect(JSON.stringify(renderer.toJSON())).toContain('Unmigrated indexes (1)'); + expect(JSON.stringify(renderer.toJSON())).toContain('Columns: name(12)'); + + const copyButton = renderer.root + .findAllByType('button') + .find((button) => button.children.includes('Copy remediation DDL'))!; + await act(async () => { + await copyButton.props.onClick(); + }); + expect(runtimeApi.ClipboardSetText).toHaveBeenCalledWith( + '-- idx_name_prefix: review remediation\n\nCREATE INDEX idx_name_prefix ON public.users (left(name, 12));', + ); + expect(writeText).toHaveBeenCalledWith( + '-- idx_name_prefix: review remediation\n\nCREATE INDEX idx_name_prefix ON public.users (left(name, 12));', + ); + }); + + it('shows a visible error when both clipboard paths fail', async () => { + runtimeApi.ClipboardSetText.mockRejectedValueOnce(new Error('runtime unavailable')); + (navigator.clipboard.writeText as ReturnType).mockRejectedValueOnce( + new Error('clipboard denied'), + ); + const renderer = TestRenderer.create( + undefined} + />, + ); + + const copyButton = renderer.root + .findAllByType('button') + .find((button) => button.children.includes('Copy remediation DDL'))!; + await act(async () => { + await copyButton.props.onClick(); + }); + + expect(JSON.stringify(renderer.toJSON())).toContain( + 'Copy failed. Copy the DDL manually from the list below.', + ); + }); + it('uses the backend notBefore window and never creates a frontend-only approval', () => { vi.useFakeTimers(); const now = Date.parse('2030-08-08T00:00:00.000Z'); diff --git a/frontend/src/components/data-sync/DataSyncPreflightPanel.tsx b/frontend/src/components/data-sync/DataSyncPreflightPanel.tsx index 5d164220..646f343f 100644 --- a/frontend/src/components/data-sync/DataSyncPreflightPanel.tsx +++ b/frontend/src/components/data-sync/DataSyncPreflightPanel.tsx @@ -1,5 +1,7 @@ import React, { useEffect, useMemo, useState } from 'react'; +import { ClipboardSetText } from '../../../wailsjs/runtime/runtime'; + import type { DataSyncPreflightSnapshot, DataSyncApprovalChallenge, @@ -7,6 +9,11 @@ import type { DataSyncTaskStage, DataSyncValidationIssue, } from './model'; +import { + buildDataSyncPreflightRemediationSQL, + collectDataSyncPreflightIndexes, + formatDataSyncPreflightIndexColumns, +} from './dataSyncPreflightIndexes'; import { dataSyncValidationIssueText, type DataSyncWorkbenchTranslate, @@ -69,6 +76,7 @@ export const DataSyncPreflightPanel: React.FC<{ onApprove, }) => { const [clock, setClock] = useState(Date.now()); + const [copyError, setCopyError] = useState(false); const effectiveIssues = stale ? [] : snapshot?.issues || []; const approvalRequired = Boolean(snapshot && snapshot.approvalRequired !== false); const approvalCurrent = Boolean( @@ -93,6 +101,29 @@ export const DataSyncPreflightPanel: React.FC<{ : stale ? 'stale' : snapshot?.status || 'stale'; + const unmigratedIndexes = useMemo( + () => collectDataSyncPreflightIndexes(effectiveIssues), + [effectiveIssues], + ); + const remediationSQL = useMemo( + () => buildDataSyncPreflightRemediationSQL(unmigratedIndexes), + [unmigratedIndexes], + ); + + const copyRemediationSQL = async () => { + if (!remediationSQL) return; + setCopyError(false); + try { + if (await ClipboardSetText(remediationSQL)) return; + } catch { + // Fall back to the browser clipboard when the Wails runtime is unavailable. + } + try { + await navigator.clipboard.writeText(remediationSQL); + } catch { + setCopyError(true); + } + }; useEffect(() => { setClock(Date.now()); @@ -183,25 +214,69 @@ export const DataSyncPreflightPanel: React.FC<{

{running ? t('preflight.running') : t('preflight.empty')}

) : ( -
    - {effectiveIssues.map((issue) => ( -
  1. -
    - - {t(`preflight.severity.${issue.severity}`)} - -

    {issueText(issue, t)}

    + <> +
      + {effectiveIssues.map((issue) => ( +
    1. +
      + + {t(`preflight.severity.${issue.severity}`)} + +

      {issueText(issue, t)}

      +
      + +
      2. + ))} +
    + {unmigratedIndexes.length > 0 ? ( +
    +
    + + {t('preflight.index_remediation.title', { + count: unmigratedIndexes.length, + })} + +
    - -
    2. - ))} -
+ {copyError ? ( +

+ {t('preflight.index_remediation.copy_failed')} +

+ ) : null} +
    + {unmigratedIndexes.map((index, itemIndex) => { + const remediationStatements = index.remediationStatements || []; + return ( +
  1. + {index.name} + {index.indexType || 'BTREE'} +

    {t('preflight.index_remediation.columns', { + columns: formatDataSyncPreflightIndexColumns(index.columns), + })}

    +

    {index.reason}

    + {remediationStatements.length > 0 ? ( + 
    {remediationStatements.join('\n')}
    + ) : null} +
    2. + ); + })} +
+ + ) : null} + )} ); diff --git a/frontend/src/components/data-sync/DataSyncWorkbench.css b/frontend/src/components/data-sync/DataSyncWorkbench.css index 4057048e..ff3ecfd7 100644 --- a/frontend/src/components/data-sync/DataSyncWorkbench.css +++ b/frontend/src/components/data-sync/DataSyncWorkbench.css @@ -1673,6 +1673,44 @@ textarea.gn-data-sync-control { .gn-data-sync-preflight-checklist p { margin: 0; } +.gn-data-sync-index-remediation { + margin-top: 14px; + padding-top: 12px; + border-top: 0.5px solid var(--gn-br-2, rgba(15, 23, 42, 0.12)); +} + +.gn-data-sync-index-remediation__header { + display: flex; + align-items: center; + justify-content: space-between; + gap: 12px; + margin-bottom: 10px; +} + +.gn-data-sync-index-remediation__error { + margin: 0 0 8px; + color: var(--gn-danger, #dc2626); + font-size: var(--gn-font-size-xs, 11px); +} + +.gn-data-sync-index-remediation .gn-data-sync-summary-list li { + padding: 10px 0; + border-bottom: 0.5px solid var(--gn-br-1, rgba(15, 23, 42, 0.08)); +} + +.gn-data-sync-index-remediation .gn-data-sync-summary-list p { + margin: 4px 0 0; +} + +.gn-data-sync-index-remediation pre { + margin: 8px 0 0; + padding: 8px; + overflow: auto; + white-space: pre-wrap; + border-radius: 10px; + background: var(--gn-bg-soft, rgba(15, 23, 42, 0.04)); +} + .gn-data-sync-action-bar { display: flex; min-height: 52px; diff --git a/frontend/src/components/data-sync/dataSyncPreflightIndexes.test.ts b/frontend/src/components/data-sync/dataSyncPreflightIndexes.test.ts new file mode 100644 index 00000000..e31eb0fc --- /dev/null +++ b/frontend/src/components/data-sync/dataSyncPreflightIndexes.test.ts @@ -0,0 +1,68 @@ +import { describe, expect, it } from 'vitest'; + +import { + buildDataSyncPreflightRemediationSQL, + collectDataSyncPreflightIndexes, + formatDataSyncPreflightIndexColumns, +} from './dataSyncPreflightIndexes'; + +describe('data sync preflight indexes', () => { + it('collects only structured unmigrated index warnings and preserves prefixes', () => { + const indexes = collectDataSyncPreflightIndexes([ + { + id: 'unmigrated_index:map-1:0', + code: 'unmigrated_index', + severity: 'warning', + stage: 'mappings', + mappingId: 'map-1', + message: 'review', + detail: { + unmigratedIndex: { + name: 'idx_lookup', + columns: [{ name: 'name', prefixLength: 12 }, { name: 'email' }], + unique: false, + indexType: 'BTREE', + reason: 'review', + remediationStatements: [], + }, + }, + }, + ]); + + expect(indexes).toHaveLength(1); + expect(formatDataSyncPreflightIndexColumns(indexes[0].columns)).toBe('name(12), email'); + }); + + it('keeps MongoDB remediation commands as valid JSON', () => { + const command = '{"createIndexes":"articles","indexes":[{"name":"idx_body","key":{"body":"text"},"unique":false}]}'; + const text = buildDataSyncPreflightRemediationSQL([ + { + name: 'idx_body', + columns: [{ name: 'body' }], + unique: false, + indexType: 'FULLTEXT', + reason: 'review text index semantics', + remediationStatements: [command], + }, + ]); + + expect(text).toContain(`\n\n${command}`); + expect(text).not.toContain(`${command};`); + }); + + it('sanitizes multiline metadata before building copyable SQL', () => { + const sql = buildDataSyncPreflightRemediationSQL([ + { + name: 'idx\nunsafe', + columns: [], + unique: false, + indexType: 'FULLTEXT', + reason: 'line one\nline two', + remediationStatements: ['CREATE INDEX idx ON target (body)'], + }, + ]); + + expect(sql).toBe('-- idx unsafe: line one line two\n\nCREATE INDEX idx ON target (body);'); + expect(sql).not.toContain('\n--'); + }); +}); diff --git a/frontend/src/components/data-sync/dataSyncPreflightIndexes.ts b/frontend/src/components/data-sync/dataSyncPreflightIndexes.ts new file mode 100644 index 00000000..7d35b19e --- /dev/null +++ b/frontend/src/components/data-sync/dataSyncPreflightIndexes.ts @@ -0,0 +1,55 @@ +import type { DataSyncUnmigratedIndex, DataSyncValidationIssue } from './model'; + +export type DataSyncPreflightIndexItem = DataSyncUnmigratedIndex & { + mappingId?: string; +}; + +const singleLine = (value: string): string => + String(value || '').replace(/[\r\n]+/g, ' ').replace(/\s+/g, ' ').trim(); + +export const collectDataSyncPreflightIndexes = ( + issues: DataSyncValidationIssue[], +): DataSyncPreflightIndexItem[] => + issues.flatMap((issue) => { + const index = issue.detail?.unmigratedIndex; + return issue.code === 'unmigrated_index' && index + ? [{ ...index, columns: Array.isArray(index.columns) ? index.columns : [], mappingId: issue.mappingId }] + : []; + }); + +const copyableStatement = (value: string): string => { + const statement = value.trim(); + try { + const parsed = JSON.parse(statement); + if (parsed && typeof parsed === 'object') return statement; + } catch { + // SQL statements are not JSON and use a trailing semicolon below. + } + return statement.endsWith(';') ? statement : `${statement};`; +}; + +export const buildDataSyncPreflightRemediationSQL = ( + indexes: DataSyncPreflightIndexItem[], +): string => + indexes + .flatMap((index) => { + const statements = (Array.isArray(index.remediationStatements) ? index.remediationStatements : []) + .map((statement) => statement.trim()) + .filter(Boolean) + .map(copyableStatement); + return statements.length > 0 + ? [`-- ${singleLine(index.name)}: ${singleLine(index.reason)}`, ...statements] + : []; + }) + .join('\n\n'); + +export const formatDataSyncPreflightIndexColumns = ( + columns: DataSyncUnmigratedIndex['columns'], +): string => + (Array.isArray(columns) ? columns : []) + .map((column) => + column.prefixLength && column.prefixLength > 0 + ? `${column.name}(${column.prefixLength})` + : column.name, + ) + .join(', '); diff --git a/frontend/src/components/data-sync/model.ts b/frontend/src/components/data-sync/model.ts index 55b6fbec..809ee66d 100644 --- a/frontend/src/components/data-sync/model.ts +++ b/frontend/src/components/data-sync/model.ts @@ -226,8 +226,25 @@ export type DataSyncValidationCode = | 'cdc_checkpoint_required' | 'cdc_checkpoint_incompatible' | 'compare_route_unsupported' + | 'index_inspection_failed' + | 'unmigrated_index' | 'capability_unverified'; +export type DataSyncIndexColumn = { + name: string; + prefixLength?: number; +}; + +export type DataSyncUnmigratedIndex = { + name: string; + columns: DataSyncIndexColumn[]; + unique: boolean; + indexType: string; + reasonCode?: string; + reason: string; + remediationStatements?: string[]; +}; + export type DataSyncValidationIssue = { id: string; severity: DataSyncValidationSeverity; @@ -235,6 +252,9 @@ export type DataSyncValidationIssue = { stage: DataSyncTaskStage; mappingId?: string; message?: string; + detail?: { + unmigratedIndex?: DataSyncUnmigratedIndex; + }; }; export type DataSyncPreflightStatus = diff --git a/frontend/src/components/data-sync/text.ts b/frontend/src/components/data-sync/text.ts index 016efaf8..b70ff65c 100644 --- a/frontend/src/components/data-sync/text.ts +++ b/frontend/src/components/data-sync/text.ts @@ -258,6 +258,10 @@ const zhCN = { 'preflight.severity.blocker': '阻断', 'preflight.severity.warning': '警告', 'preflight.severity.info': '提示', + 'preflight.index_remediation.title': '未迁移索引（{count}）', + 'preflight.index_remediation.columns': '列：{columns}', + 'preflight.index_remediation.copy': '复制补救 DDL', + 'preflight.index_remediation.copy_failed': '复制失败，请从下方清单手动复制 DDL。', 'runs.title': '运行记录', 'runs.subtitle': '运行实例由 gateway 提供；后续接入持久化历史与 checkpoint。', 'runs.empty_title': '还没有运行记录', @@ -398,6 +402,8 @@ const zhCN = { 'validation.cdc_checkpoint_required': '选择 checkpoint 起点前必须存在兼容的持久化 checkpoint。', 'validation.cdc_checkpoint_incompatible': '已存在的 CDC checkpoint 与当前任务定义不兼容，请在暂停任务后显式重置。', 'validation.compare_route_unsupported': '当前源端与目标端组合不支持执行数据比对。', + 'validation.index_inspection_failed': '无法读取源端索引元数据，不能确认所有索引都已迁移。', + 'validation.unmigrated_index': '有索引无法自动迁移，请在执行前审核候选补救 DDL。', 'validation.capability_unverified': '当前使用静态 gateway，运行前仍需后端确认数据源组合能力。', 'validation.unknown': '预检发现未识别的配置问题。', } as const; @@ -672,6 +678,10 @@ const enUS: Record = { 'preflight.severity.blocker': 'Blocker', 'preflight.severity.warning': 'Warning', 'preflight.severity.info': 'Info', + 'preflight.index_remediation.title': 'Unmigrated indexes ({count})', + 'preflight.index_remediation.columns': 'Columns: {columns}', + 'preflight.index_remediation.copy': 'Copy remediation DDL', + 'preflight.index_remediation.copy_failed': 'Copy failed. Copy the DDL manually from the list below.', 'runs.title': 'Run history', 'runs.subtitle': 'Run instances come from the gateway; persisted history and checkpoints plug in later.', 'runs.empty_title': 'No runs yet', @@ -812,6 +822,8 @@ const enUS: Record = { 'validation.cdc_checkpoint_required': 'Checkpoint start requires a compatible durable checkpoint for this task.', 'validation.cdc_checkpoint_incompatible': 'The stored CDC checkpoint is incompatible with this task. Pause the task and reset it explicitly.', 'validation.compare_route_unsupported': 'The selected source-target route cannot execute compare tasks.', + 'validation.index_inspection_failed': 'Source index metadata could not be read, so complete index migration cannot be verified.', + 'validation.unmigrated_index': 'An index could not be migrated automatically. Review the remediation DDL before execution.', 'validation.capability_unverified': 'This task uses the static gateway. The backend must still verify the source-target capability before execution.', 'validation.unknown': 'Preflight found an unrecognized configuration issue.', }; diff --git a/frontend/src/components/data-sync/wailsDto.test.ts b/frontend/src/components/data-sync/wailsDto.test.ts index aa062236..8424156e 100644 --- a/frontend/src/components/data-sync/wailsDto.test.ts +++ b/frontend/src/components/data-sync/wailsDto.test.ts @@ -322,6 +322,23 @@ describe('data sync Wails DTO boundary', () => { stage: 'endpoints', message: 'unsupported route', }, + { + code: 'unmigrated_index', + severity: 'warning', + stage: 'mappings', + message: 'index requires review', + detail: { + unmigratedIndex: { + name: 'idx_name_prefix', + columns: [{ name: 'name', prefixLength: 12 }], + unique: false, + indexType: 'BTREE', + reasonCode: 'prefix_index_requires_review', + reason: 'index requires review', + remediationStatements: ['CREATE INDEX idx_name_prefix ON public.users (left(name, 12))'], + }, + }, + }, ], checkedAt: Date.parse('2026-08-08T00:02:00.000Z'), }, @@ -332,7 +349,18 @@ describe('data sync Wails DTO boundary', () => { expect(blocked.snapshot).toMatchObject({ status: 'blocked', approvalSatisfied: false, - issues: [{ message: 'unsupported route' }], + issues: [ + { message: 'unsupported route' }, + { + detail: { + unmigratedIndex: { + name: 'idx_name_prefix', + reasonCode: 'prefix_index_requires_review', + remediationStatements: ['CREATE INDEX idx_name_prefix ON public.users (left(name, 12))'], + }, + }, + }, + ], }); const earlyBlocked = decodeDataSyncPreflightQuery( { diff --git a/frontend/src/components/data-sync/wailsDto.ts b/frontend/src/components/data-sync/wailsDto.ts index 901a4c1c..e9d3ceec 100644 --- a/frontend/src/components/data-sync/wailsDto.ts +++ b/frontend/src/components/data-sync/wailsDto.ts @@ -7,6 +7,7 @@ import { type DataSyncErrorRow, type DataSyncFieldMetadata, type DataSyncObjectMetadata, + type DataSyncIndexColumn, type DataSyncPreflightSnapshot, type DataSyncRouteCapability, type DataSyncRunRecord, @@ -15,6 +16,7 @@ import { type DataSyncTableMapping, type DataSyncTaskDefinition, type DataSyncTaskLifecycle, + type DataSyncUnmigratedIndex, type DataSyncValidationCode, type DataSyncValidationIssue, } from './model'; @@ -883,6 +885,14 @@ export const decodeDataSyncPreflight = ( ), mappingId: optionalString(issue.mappingId, 'issue.mappingId') || undefined, message: optionalString(issue.message, 'issue.message') || undefined, + detail: isRecord(issue.detail) && issue.detail.unmigratedIndex + ? { + unmigratedIndex: decodeUnmigratedIndex( + issue.detail.unmigratedIndex, + `DataSyncJobPreflight.data.issues[${index}].detail.unmigratedIndex`, + ), + } + : undefined, }; }); let capability: DataSyncRouteCapability; @@ -1082,6 +1092,34 @@ const previewJSON = (value: unknown, limit = 480): string => { return text.length > limit ? `${text.slice(0, limit)}…` : text; }; +const decodeIndexColumns = (value: unknown, path: string): DataSyncIndexColumn[] => + array(value, path).map((item, index) => { + const column = record(item, `${path}[${index}]`); + return { + name: string(column.name, `${path}[${index}].name`, false), + prefixLength: + optionalNumber(column.prefixLength, `${path}[${index}].prefixLength`) || undefined, + }; + }); + +const decodeUnmigratedIndex = (value: unknown, path: string): DataSyncUnmigratedIndex => { + const index = record(value, path); + return { + name: string(index.name, `${path}.name`, false), + columns: decodeIndexColumns(index.columns || [], `${path}.columns`), + unique: boolean(index.unique, `${path}.unique`), + indexType: optionalString(index.indexType, `${path}.indexType`) || '', + reasonCode: optionalString(index.reasonCode, `${path}.reasonCode`) || undefined, + reason: string(index.reason, `${path}.reason`, false), + remediationStatements: array( + index.remediationStatements || [], + `${path}.remediationStatements`, + ).map((statement, indexOffset) => + string(statement, `${path}.remediationStatements[${indexOffset}]`, false), + ), + }; +}; + export const decodeErrorRow = (value: unknown): DataSyncErrorRow => { const row = record(value, 'errorRow'); const source = optionalString(row.sourceTable, 'errorRow.sourceTable'); diff --git a/internal/app/data_sync_job_api_types.go b/internal/app/data_sync_job_api_types.go index ad63fedb..c79ddcae 100644 --- a/internal/app/data_sync_job_api_types.go +++ b/internal/app/data_sync_job_api_types.go @@ -14,12 +14,17 @@ const ( DataSyncJobPreflightInfo DataSyncJobPreflightSeverity = "info" ) +type DataSyncJobPreflightIssueDetail struct { + UnmigratedIndex *sync.UnmigratedIndex `json:"unmigratedIndex,omitempty"` +} + type DataSyncJobPreflightIssue struct { - Code string `json:"code"` - Severity DataSyncJobPreflightSeverity `json:"severity"` - Stage string `json:"stage"` - Message string `json:"message"` - MappingID string `json:"mappingId,omitempty"` + Code string `json:"code"` + Severity DataSyncJobPreflightSeverity `json:"severity"` + Stage string `json:"stage"` + Message string `json:"message"` + MappingID string `json:"mappingId,omitempty"` + Detail *DataSyncJobPreflightIssueDetail `json:"detail,omitempty"` } type DataSyncJobPreflightResult struct { diff --git a/internal/app/data_sync_job_endpoint_test.go b/internal/app/data_sync_job_endpoint_test.go index 84ceb7d7..66a7508c 100644 --- a/internal/app/data_sync_job_endpoint_test.go +++ b/internal/app/data_sync_job_endpoint_test.go @@ -8,6 +8,24 @@ import ( "GoNavi-Wails/internal/syncjob" ) +func TestDataSyncJobSourceIndexLocationUsesMappingThenEndpointSelection(t *testing.T) { + endpoint := resolvedDataSyncJobEndpoint{Database: "sales", Schema: "public"} + + schema, table := dataSyncJobSourceIndexLocation(endpoint, syncjob.TableMapping{SourceSchema: " tenant ", SourceTable: " orders "}) + if schema != "tenant" || table != "orders" { + t.Fatalf("mapping schema must win: schema=%q table=%q", schema, table) + } + schema, _ = dataSyncJobSourceIndexLocation(endpoint, syncjob.TableMapping{SourceTable: "orders"}) + if schema != "public" { + t.Fatalf("endpoint schema must win over database: %q", schema) + } + endpoint.Schema = "" + schema, _ = dataSyncJobSourceIndexLocation(endpoint, syncjob.TableMapping{SourceTable: "orders"}) + if schema != "sales" { + t.Fatalf("database must be the final fallback: %q", schema) + } +} + func TestDataSyncJobEndpointFingerprintHMACTracksSecretsAndSelection(t *testing.T) { key := []byte("0123456789abcdef0123456789abcdef") base := resolvedDataSyncJobEndpoint{ diff --git a/internal/app/data_sync_job_preflight.go b/internal/app/data_sync_job_preflight.go index cb4ce401..08b3ec70 100644 --- a/internal/app/data_sync_job_preflight.go +++ b/internal/app/data_sync_job_preflight.go @@ -273,6 +273,7 @@ func (a *App) preflightDataSyncMappings(definition syncjob.JobDefinition, source issues = append(issues, preflightIssue("target_table_missing", DataSyncJobPreflightBlocker, "mappings", "target table does not exist and this mapping cannot auto-create it", mappingID)) } else { issues = append(issues, preflightIssue("target_table_will_be_created", DataSyncJobPreflightInfo, "mappings", "target table will be created by the migration planner", mappingID)) + issues = append(issues, a.preflightUnmigratedIndexes(definition, source, target, mapping)...) } continue } @@ -297,6 +298,54 @@ func (a *App) preflightDataSyncMappings(definition syncjob.JobDefinition, source return issues } +func dataSyncJobSourceIndexLocation(source resolvedDataSyncJobEndpoint, mapping syncjob.TableMapping) (string, string) { + schema := firstNonEmptySyncJob(mapping.SourceSchema, source.Schema, source.Database) + return strings.TrimSpace(schema), strings.TrimSpace(mapping.SourceTable) +} + +func (a *App) preflightUnmigratedIndexes(definition syncjob.JobDefinition, source, target resolvedDataSyncJobEndpoint, mapping syncjob.TableMapping) []DataSyncJobPreflightIssue { + if !definition.Options.CreateIndexes { + return nil + } + config, err := buildDataSyncJobEngineConfig(definition, "preflight", source, target, mapping) + if err != nil { + return []DataSyncJobPreflightIssue{preflightIssue("mapping_compile_failed", DataSyncJobPreflightBlocker, "mappings", err.Error(), dataSyncJobMappingLabel(mapping))} + } + sourceDB, sourceErr := a.getDatabase(normalizeMetadataRunConfig(source.Config, source.Database)) + if sourceErr != nil { + return []DataSyncJobPreflightIssue{preflightIssue("source_connect_failed", DataSyncJobPreflightBlocker, "endpoints", sourceErr.Error(), dataSyncJobMappingLabel(mapping))} + } + sourceSchema, sourceTable := dataSyncJobSourceIndexLocation(source, mapping) + if _, indexErr := sourceDB.GetIndexes(sourceSchema, sourceTable); indexErr != nil { + return []DataSyncJobPreflightIssue{preflightIssue("index_inspection_failed", DataSyncJobPreflightWarning, "mappings", indexErr.Error(), dataSyncJobMappingLabel(mapping))} + } + targetDB, targetErr := a.getDatabase(normalizeMetadataRunConfig(target.Config, target.Database)) + if targetErr != nil { + return []DataSyncJobPreflightIssue{preflightIssue("target_connect_failed", DataSyncJobPreflightBlocker, "endpoints", targetErr.Error(), dataSyncJobMappingLabel(mapping))} + } + qualifiedSourceTable := strings.TrimSpace(mapping.SourceTable) + if strings.TrimSpace(mapping.SourceSchema) != "" { + qualifiedSourceTable = strings.TrimSpace(mapping.SourceSchema) + "." + qualifiedSourceTable + } + plan, planErr := sync.InspectSchemaMigrationPlan(config, qualifiedSourceTable, sourceDB, targetDB) + if planErr != nil { + return []DataSyncJobPreflightIssue{preflightIssue("schema_inspection_failed", DataSyncJobPreflightWarning, "mappings", planErr.Error(), dataSyncJobMappingLabel(mapping))} + } + issues := make([]DataSyncJobPreflightIssue, 0, len(plan.UnmigratedIndexes)) + for _, index := range plan.UnmigratedIndexes { + indexCopy := index + issues = append(issues, DataSyncJobPreflightIssue{ + Code: "unmigrated_index", + Severity: DataSyncJobPreflightWarning, + Stage: "mappings", + Message: index.Reason, + MappingID: dataSyncJobMappingLabel(mapping), + Detail: &DataSyncJobPreflightIssueDetail{UnmigratedIndex: &indexCopy}, + }) + } + return issues +} + func (a *App) preflightDataSyncQueryTarget(definition syncjob.JobDefinition, mapping syncjob.TableMapping, target resolvedDataSyncJobEndpoint) []DataSyncJobPreflightIssue { mappingID := dataSyncJobMappingLabel(mapping) issues := make([]DataSyncJobPreflightIssue, 0) diff --git a/internal/sync/analyze.go b/internal/sync/analyze.go index e349cbfa..f2643f3f 100644 --- a/internal/sync/analyze.go +++ b/internal/sync/analyze.go @@ -7,22 +7,23 @@ import ( ) type TableDiffSummary struct { - Table string `json:"table"` - PKColumn string `json:"pkColumn,omitempty"` - CanSync bool `json:"canSync"` - Inserts int `json:"inserts"` - Updates int `json:"updates"` - Deletes int `json:"deletes"` - Same int `json:"same"` - SchemaDiffCount int `json:"schemaDiffCount,omitempty"` - Message string `json:"message,omitempty"` - HasSchema bool `json:"hasSchema,omitempty"` - TargetTableExists bool `json:"targetTableExists,omitempty"` - PlannedAction string `json:"plannedAction,omitempty"` - Warnings []string `json:"warnings,omitempty"` - UnsupportedObjects []string `json:"unsupportedObjects,omitempty"` - IndexesToCreate int `json:"indexesToCreate,omitempty"` - IndexesSkipped int `json:"indexesSkipped,omitempty"` + Table string `json:"table"` + PKColumn string `json:"pkColumn,omitempty"` + CanSync bool `json:"canSync"` + Inserts int `json:"inserts"` + Updates int `json:"updates"` + Deletes int `json:"deletes"` + Same int `json:"same"` + SchemaDiffCount int `json:"schemaDiffCount,omitempty"` + Message string `json:"message,omitempty"` + HasSchema bool `json:"hasSchema,omitempty"` + TargetTableExists bool `json:"targetTableExists,omitempty"` + PlannedAction string `json:"plannedAction,omitempty"` + Warnings []string `json:"warnings,omitempty"` + UnsupportedObjects []string `json:"unsupportedObjects,omitempty"` + UnmigratedIndexes []UnmigratedIndex `json:"unmigratedIndexes,omitempty"` + IndexesToCreate int `json:"indexesToCreate,omitempty"` + IndexesSkipped int `json:"indexesSkipped,omitempty"` } type SyncAnalyzeResult struct { @@ -134,6 +135,7 @@ func (s *SyncEngine) Analyze(config SyncConfig) SyncAnalyzeResult { summary.PlannedAction = plan.PlannedAction summary.Warnings = append(summary.Warnings, plan.Warnings...) summary.UnsupportedObjects = append(summary.UnsupportedObjects, plan.UnsupportedObjects...) + summary.UnmigratedIndexes = append(summary.UnmigratedIndexes, plan.UnmigratedIndexes...) summary.IndexesToCreate = plan.IndexesToCreate summary.IndexesSkipped = plan.IndexesSkipped summary.SchemaDiffCount = len(plan.PreDataSQL) + len(plan.PostDataSQL) diff --git a/internal/sync/migration_mongodb.go b/internal/sync/migration_mongodb.go index 0389740c..7074e585 100644 --- a/internal/sync/migration_mongodb.go +++ b/internal/sync/migration_mongodb.go @@ -72,13 +72,14 @@ func buildTabularToMongoPlan(config SyncConfig, tableName string, sourceDB db.Da } plan.PreDataSQL = append(plan.PreDataSQL, createCmd) if config.CreateIndexes { - indexCmds, warnings, unsupported, created, skipped, err := buildMongoIndexCommands(sourceDB, plan.SourceSchema, plan.SourceTable, plan.TargetTable) + indexCmds, warnings, unsupported, unmigrated, created, skipped, err := buildMongoIndexCommands(sourceDB, plan.SourceSchema, plan.SourceTable, plan.TargetTable) if err != nil { plan.Warnings = append(plan.Warnings, fmt.Sprintf("读取源表索引失败，已跳过索引迁移：%v", err)) } else { plan.PostDataSQL = append(plan.PostDataSQL, indexCmds...) plan.Warnings = append(plan.Warnings, warnings...) plan.UnsupportedObjects = append(plan.UnsupportedObjects, unsupported...) + plan.UnmigratedIndexes = append(plan.UnmigratedIndexes, unmigrated...) plan.IndexesToCreate = created plan.IndexesSkipped = skipped } @@ -139,13 +140,14 @@ func buildMongoToMongoPlan(config SyncConfig, tableName string, sourceDB db.Data } plan.PreDataSQL = append(plan.PreDataSQL, createCmd) if config.CreateIndexes { - indexCmds, indexWarnings, unsupported, created, skipped, err := buildMongoIndexCommands(sourceDB, plan.SourceSchema, plan.SourceTable, plan.TargetTable) + indexCmds, indexWarnings, unsupported, unmigrated, created, skipped, err := buildMongoIndexCommands(sourceDB, plan.SourceSchema, plan.SourceTable, plan.TargetTable) if err != nil { plan.Warnings = append(plan.Warnings, fmt.Sprintf("读取源集合索引失败，已跳过索引迁移：%v", err)) } else { plan.PostDataSQL = append(plan.PostDataSQL, indexCmds...) plan.Warnings = append(plan.Warnings, indexWarnings...) plan.UnsupportedObjects = append(plan.UnsupportedObjects, unsupported...) + plan.UnmigratedIndexes = append(plan.UnmigratedIndexes, unmigrated...) plan.IndexesToCreate = created plan.IndexesSkipped = skipped } @@ -245,15 +247,60 @@ func buildMongoCreateCollectionCommand(collection string) (string, error) { return string(data), nil } -func buildMongoIndexCommands(sourceDB db.Database, dbName, tableName, targetCollection string) ([]string, []string, []string, int, int, error) { +func buildMongoIndexCommand(targetCollection string, idx groupedIndex, text bool) (string, error) { + keyParts := make([]string, 0, len(idx.Columns)) + for _, column := range idx.Columns { + nameJSON, err := json.Marshal(strings.TrimSpace(column.Name)) + if err != nil { + return "", err + } + value := "1" + if text { + value = `"text"` + } + keyParts = append(keyParts, fmt.Sprintf("%s:%s", nameJSON, value)) + } + command := struct { + CreateIndexes string `json:"createIndexes"` + Indexes []struct { + Name string `json:"name"` + Key json.RawMessage `json:"key"` + Unique bool `json:"unique"` + } `json:"indexes"` + }{CreateIndexes: strings.TrimSpace(targetCollection)} + command.Indexes = append(command.Indexes, struct { + Name string `json:"name"` + Key json.RawMessage `json:"key"` + Unique bool `json:"unique"` + }{ + Name: strings.TrimSpace(idx.Name), + Key: json.RawMessage(`{` + strings.Join(keyParts, ",") + `}`), + Unique: idx.Unique, + }) + data, err := json.Marshal(command) + if err != nil { + return "", err + } + return string(data), nil +} + +func remediationStatements(command string) []string { + if strings.TrimSpace(command) == "" { + return nil + } + return []string{command} +} + +func buildMongoIndexCommands(sourceDB db.Database, dbName, tableName, targetCollection string) ([]string, []string, []string, []UnmigratedIndex, int, int, error) { indexes, err := sourceDB.GetIndexes(dbName, tableName) if err != nil { - return nil, nil, nil, 0, 0, err + return nil, nil, nil, nil, 0, 0, err } grouped := groupIndexDefinitions(indexes) cmds := make([]string, 0, len(grouped)) warnings := make([]string, 0) unsupported := make([]string, 0) + unmigrated := make([]UnmigratedIndex, 0) created := 0 skipped := 0 for _, idx := range grouped { @@ -262,41 +309,85 @@ func buildMongoIndexCommands(sourceDB db.Database, dbName, tableName, targetColl continue } if len(idx.Columns) == 0 { + reason := fmt.Sprintf("索引 %s 缺少列定义，已跳过", name) skipped++ - unsupported = append(unsupported, fmt.Sprintf("索引 %s 缺少列定义，已跳过", name)) + unsupported = append(unsupported, reason) + unmigrated = append(unmigrated, UnmigratedIndex{ + Name: name, + Columns: []IndexMigrationColumn{}, + Unique: idx.Unique, + IndexType: idx.IndexType, + ReasonCode: "missing_columns", + Reason: reason, + }) continue } kind := strings.ToLower(strings.TrimSpace(idx.IndexType)) - if idx.SubPart > 0 { + if hasIndexPrefix(idx.Columns) { + reason := fmt.Sprintf("索引 %s 使用前缀长度，MongoDB 目标暂不支持等价迁移", name) + remediation, _ := buildMongoIndexCommand(targetCollection, idx, false) skipped++ - unsupported = append(unsupported, fmt.Sprintf("索引 %s 使用前缀长度，MongoDB 目标暂不支持等价迁移", name)) + unsupported = append(unsupported, reason) + unmigrated = append(unmigrated, UnmigratedIndex{ + Name: name, + Columns: append([]IndexMigrationColumn(nil), idx.Columns...), + Unique: idx.Unique, + IndexType: idx.IndexType, + ReasonCode: "prefix_index_requires_review", + Reason: reason, + RemediationStatements: remediationStatements(remediation), + }) + continue + } + if kind == "fulltext" { + reason := fmt.Sprintf("索引 %s 类型=%s，MongoDB 目标暂不支持等价迁移", name, idx.IndexType) + remediation, _ := buildMongoIndexCommand(targetCollection, idx, true) + skipped++ + unsupported = append(unsupported, reason) + unmigrated = append(unmigrated, UnmigratedIndex{ + Name: name, + Columns: append([]IndexMigrationColumn(nil), idx.Columns...), + Unique: idx.Unique, + IndexType: idx.IndexType, + ReasonCode: "fulltext_requires_review", + Reason: reason, + RemediationStatements: remediationStatements(remediation), + }) continue } if kind != "" && kind != "btree" { - warnings = append(warnings, fmt.Sprintf("索引 %s 类型=%s 将按普通索引迁移到 MongoDB", name, idx.IndexType)) - } - keySpec := make(map[string]int) - for _, col := range idx.Columns { - keySpec[col] = 1 - } - command := map[string]interface{}{ - "createIndexes": strings.TrimSpace(targetCollection), - "indexes": []map[string]interface{}{{ - "name": name, - "key": keySpec, - "unique": idx.Unique, - }}, - } - data, err := json.Marshal(command) - if err != nil { + reason := fmt.Sprintf("索引 %s 类型=%s，MongoDB 目标暂不支持等价迁移", name, idx.IndexType) skipped++ - unsupported = append(unsupported, fmt.Sprintf("索引 %s 生成 MongoDB createIndexes 命令失败：%v", name, err)) + unsupported = append(unsupported, reason) + unmigrated = append(unmigrated, UnmigratedIndex{ + Name: name, + Columns: append([]IndexMigrationColumn(nil), idx.Columns...), + Unique: idx.Unique, + IndexType: idx.IndexType, + ReasonCode: "unsupported_index_type", + Reason: reason, + }) continue } - cmds = append(cmds, string(data)) + command, err := buildMongoIndexCommand(targetCollection, idx, false) + if err != nil { + reason := fmt.Sprintf("索引 %s 生成 MongoDB createIndexes 命令失败：%v", name, err) + skipped++ + unsupported = append(unsupported, reason) + unmigrated = append(unmigrated, UnmigratedIndex{ + Name: name, + Columns: append([]IndexMigrationColumn(nil), idx.Columns...), + Unique: idx.Unique, + IndexType: idx.IndexType, + ReasonCode: "remediation_generation_failed", + Reason: reason, + }) + continue + } + cmds = append(cmds, command) created++ } - return cmds, dedupeStrings(warnings), dedupeStrings(unsupported), created, skipped, nil + return cmds, dedupeStrings(warnings), dedupeStrings(unsupported), unmigrated, created, skipped, nil } func inferMongoCollectionColumns(sourceDB db.Database, collection string) ([]connection.ColumnDefinition, []string, error) { @@ -468,7 +559,7 @@ func buildMongoToMySQLCreateTablePlan(config SyncConfig, targetQueryTable string } quotedCols := make([]string, 0, len(idx.Columns)) for _, col := range idx.Columns { - quotedCols = append(quotedCols, quoteIdentByType("mysql", col)) + quotedCols = append(quotedCols, quoteIdentByType("mysql", col.Name)) } prefix := "CREATE INDEX" if idx.Unique { @@ -642,7 +733,7 @@ func buildMongoToPGLikeCreateTablePlan(targetType string, config SyncConfig, tar } quotedCols := make([]string, 0, len(idx.Columns)) for _, col := range idx.Columns { - quotedCols = append(quotedCols, quoteIdentByType(targetType, col)) + quotedCols = append(quotedCols, quoteIdentByType(targetType, col.Name)) } prefix := "CREATE INDEX" if idx.Unique { diff --git a/internal/sync/preview.go b/internal/sync/preview.go index ad261715..e7162242 100644 --- a/internal/sync/preview.go +++ b/internal/sync/preview.go @@ -18,18 +18,19 @@ type PreviewUpdateRow struct { } type TableDiffPreview struct { - Table string `json:"table"` - PKColumn string `json:"pkColumn"` - ColumnTypes map[string]string `json:"columnTypes,omitempty"` - SchemaSummary string `json:"schemaSummary,omitempty"` - SchemaWarnings []string `json:"schemaWarnings,omitempty"` - SchemaStatements []string `json:"schemaStatements,omitempty"` - TotalInserts int `json:"totalInserts"` - TotalUpdates int `json:"totalUpdates"` - TotalDeletes int `json:"totalDeletes"` - Inserts []PreviewRow `json:"inserts"` - Updates []PreviewUpdateRow `json:"updates"` - Deletes []PreviewRow `json:"deletes"` + Table string `json:"table"` + PKColumn string `json:"pkColumn"` + ColumnTypes map[string]string `json:"columnTypes,omitempty"` + SchemaSummary string `json:"schemaSummary,omitempty"` + SchemaWarnings []string `json:"schemaWarnings,omitempty"` + SchemaStatements []string `json:"schemaStatements,omitempty"` + UnmigratedIndexes []UnmigratedIndex `json:"unmigratedIndexes,omitempty"` + TotalInserts int `json:"totalInserts"` + TotalUpdates int `json:"totalUpdates"` + TotalDeletes int `json:"totalDeletes"` + Inserts []PreviewRow `json:"inserts"` + Updates []PreviewUpdateRow `json:"updates"` + Deletes []PreviewRow `json:"deletes"` } func (s *SyncEngine) Preview(config SyncConfig, tableName string, limit int) (TableDiffPreview, error) { @@ -94,10 +95,11 @@ func (s *SyncEngine) Preview(config SyncConfig, tableName string, limit int) (Ta contentRaw := strings.ToLower(strings.TrimSpace(config.Content)) if contentRaw == "schema" { return TableDiffPreview{ - Table: tableName, - SchemaSummary: firstNonEmpty(plan.PlannedAction, "仅同步结构"), - SchemaWarnings: append([]string(nil), plan.Warnings...), - SchemaStatements: append([]string(nil), schemaStatements...), + Table: tableName, + SchemaSummary: firstNonEmpty(plan.PlannedAction, "仅同步结构"), + SchemaWarnings: append([]string(nil), plan.Warnings...), + SchemaStatements: append([]string(nil), schemaStatements...), + UnmigratedIndexes: append([]UnmigratedIndex(nil), plan.UnmigratedIndexes...), }, nil } @@ -126,18 +128,19 @@ func (s *SyncEngine) Preview(config SyncConfig, tableName string, limit int) (Ta sourceType := resolveMigrationDBType(config.SourceConfig) targetType := resolveMigrationDBType(config.TargetConfig) out := TableDiffPreview{ - Table: tableName, - PKColumn: pkCol, - ColumnTypes: make(map[string]string, len(cols)), - SchemaSummary: firstNonEmpty(plan.PlannedAction, "结构预览"), - SchemaWarnings: append([]string(nil), plan.Warnings...), - SchemaStatements: append([]string(nil), schemaStatements...), - TotalInserts: 0, - TotalUpdates: 0, - TotalDeletes: 0, - Inserts: make([]PreviewRow, 0), - Updates: make([]PreviewUpdateRow, 0), - Deletes: make([]PreviewRow, 0), + Table: tableName, + PKColumn: pkCol, + ColumnTypes: make(map[string]string, len(cols)), + SchemaSummary: firstNonEmpty(plan.PlannedAction, "结构预览"), + SchemaWarnings: append([]string(nil), plan.Warnings...), + SchemaStatements: append([]string(nil), schemaStatements...), + UnmigratedIndexes: append([]UnmigratedIndex(nil), plan.UnmigratedIndexes...), + TotalInserts: 0, + TotalUpdates: 0, + TotalDeletes: 0, + Inserts: make([]PreviewRow, 0), + Updates: make([]PreviewUpdateRow, 0), + Deletes: make([]PreviewRow, 0), } columnTypes := cols if hasExplicitSyncMappings(config) { diff --git a/internal/sync/schema_migration.go b/internal/sync/schema_migration.go index 40e41d85..626c7df3 100644 --- a/internal/sync/schema_migration.go +++ b/internal/sync/schema_migration.go @@ -22,6 +22,7 @@ type SchemaMigrationPlan struct { PlannedAction string Warnings []string UnsupportedObjects []string + UnmigratedIndexes []UnmigratedIndex IndexesToCreate int IndexesSkipped int CreateTableSQL string @@ -29,12 +30,26 @@ type SchemaMigrationPlan struct { PostDataSQL []string } +type IndexMigrationColumn struct { + Name string `json:"name"` + PrefixLength int `json:"prefixLength,omitempty"` +} + +type UnmigratedIndex struct { + Name string `json:"name"` + Columns []IndexMigrationColumn `json:"columns"` + Unique bool `json:"unique"` + IndexType string `json:"indexType"` + ReasonCode string `json:"reasonCode"` + Reason string `json:"reason"` + RemediationStatements []string `json:"remediationStatements,omitempty"` +} + type groupedIndex struct { Name string - Columns []string + Columns []IndexMigrationColumn Unique bool IndexType string - SubPart int } func normalizeTargetTableStrategy(strategy string) string { @@ -188,7 +203,7 @@ func buildSchemaMigrationPlanLegacy(config SyncConfig, tableName string, sourceD } plan.AutoCreate = true plan.PlannedAction = "目标表不存在，将自动建表后导入" - createSQL, postSQL, warnings, unsupported, idxCreate, idxSkip, err := buildMySQLToKingbaseCreateTablePlan(config, plan.TargetQueryTable, sourceCols, sourceDB, plan.SourceSchema, plan.SourceTable) + createSQL, postSQL, warnings, unsupported, unmigrated, idxCreate, idxSkip, err := buildMySQLToKingbaseCreateTablePlan(config, plan.TargetQueryTable, sourceCols, sourceDB, plan.SourceSchema, plan.SourceTable) if err != nil { return plan, sourceCols, targetCols, err } @@ -196,6 +211,7 @@ func buildSchemaMigrationPlanLegacy(config SyncConfig, tableName string, sourceD plan.PostDataSQL = append(plan.PostDataSQL, postSQL...) plan.Warnings = append(plan.Warnings, warnings...) plan.UnsupportedObjects = append(plan.UnsupportedObjects, unsupported...) + plan.UnmigratedIndexes = append(plan.UnmigratedIndexes, unmigrated...) plan.IndexesToCreate = idxCreate plan.IndexesSkipped = idxSkip return dedupeSchemaMigrationPlan(plan), sourceCols, targetCols, nil @@ -210,6 +226,11 @@ func dedupeSchemaMigrationPlan(plan SchemaMigrationPlan) SchemaMigrationPlan { return plan } +func InspectSchemaMigrationPlan(config SyncConfig, tableName string, sourceDB db.Database, targetDB db.Database) (SchemaMigrationPlan, error) { + plan, _, _, err := buildSchemaMigrationPlan(config, tableName, sourceDB, targetDB) + return plan, err +} + func dedupeStrings(items []string) []string { if len(items) == 0 { return items @@ -291,7 +312,7 @@ func buildMySQLToKingbaseAddColumnSQL(targetQueryTable string, sourceCols, targe return sqlList, dedupeStrings(warnings) } -func buildMySQLToKingbaseCreateTablePlan(config SyncConfig, targetQueryTable string, sourceCols []connection.ColumnDefinition, sourceDB db.Database, sourceSchema, sourceTable string) (string, []string, []string, []string, int, int, error) { +func buildMySQLToKingbaseCreateTablePlan(config SyncConfig, targetQueryTable string, sourceCols []connection.ColumnDefinition, sourceDB db.Database, sourceSchema, sourceTable string) (string, []string, []string, []string, []UnmigratedIndex, int, int, error) { columnDefs := make([]string, 0, len(sourceCols)+1) warnings := make([]string, 0) unsupported := make([]string, 0) @@ -311,51 +332,16 @@ func buildMySQLToKingbaseCreateTablePlan(config SyncConfig, targetQueryTable str createSQL := fmt.Sprintf("CREATE TABLE %s (\n %s\n)", quoteQualifiedIdentByType("kingbase", targetQueryTable), strings.Join(columnDefs, ",\n ")) if !config.CreateIndexes { - return createSQL, nil, dedupeStrings(warnings), dedupeStrings(unsupported), 0, 0, nil + return createSQL, nil, dedupeStrings(warnings), dedupeStrings(unsupported), nil, 0, 0, nil } indexes, err := sourceDB.GetIndexes(sourceSchema, sourceTable) if err != nil { warnings = append(warnings, fmt.Sprintf("读取源表索引失败，已跳过索引迁移：%v", err)) - return createSQL, nil, dedupeStrings(warnings), dedupeStrings(unsupported), 0, 0, nil + return createSQL, nil, dedupeStrings(warnings), dedupeStrings(unsupported), nil, 0, 0, nil } - grouped := groupIndexDefinitions(indexes) - postSQL := make([]string, 0, len(grouped)) - created := 0 - skipped := 0 - for _, idx := range grouped { - name := strings.TrimSpace(idx.Name) - if name == "" || strings.EqualFold(name, "primary") { - continue - } - if len(idx.Columns) == 0 { - skipped++ - unsupported = append(unsupported, fmt.Sprintf("索引 %s 缺少列定义，已跳过", name)) - continue - } - kind := strings.ToLower(strings.TrimSpace(idx.IndexType)) - if idx.SubPart > 0 { - skipped++ - unsupported = append(unsupported, fmt.Sprintf("索引 %s 使用前缀长度，当前暂不支持迁移", name)) - continue - } - if kind != "" && kind != "btree" { - skipped++ - unsupported = append(unsupported, fmt.Sprintf("索引 %s 类型=%s，当前暂不支持自动迁移", name, idx.IndexType)) - continue - } - quotedCols := make([]string, 0, len(idx.Columns)) - for _, col := range idx.Columns { - quotedCols = append(quotedCols, quoteIdentByType("kingbase", col)) - } - prefix := "CREATE INDEX" - if idx.Unique { - prefix = "CREATE UNIQUE INDEX" - } - postSQL = append(postSQL, fmt.Sprintf("%s %s ON %s (%s)", prefix, quoteIdentByType("kingbase", name), quoteQualifiedIdentByType("kingbase", targetQueryTable), strings.Join(quotedCols, ", "))) - created++ - } - return createSQL, postSQL, dedupeStrings(warnings), dedupeStrings(unsupported), created, skipped, nil + postSQL, unsupported, unmigrated, created, skipped := buildMySQLSourceIndexPlan("kingbase", targetQueryTable, indexes) + return createSQL, postSQL, dedupeStrings(warnings), unsupported, unmigrated, created, skipped, nil } func buildMySQLToKingbaseColumnDefinition(col connection.ColumnDefinition) (string, []string) { @@ -534,12 +520,12 @@ func groupIndexDefinitions(indexes []connection.IndexDefinition) []groupedIndex if strings.TrimSpace(row.IndexType) != "" { gi.IndexType = row.IndexType } - if row.SubPart > 0 && gi.SubPart == 0 { - gi.SubPart = row.SubPart - } col := strings.TrimSpace(row.ColumnName) if col != "" { - gi.Columns = append(gi.Columns, col) + gi.Columns = append(gi.Columns, IndexMigrationColumn{ + Name: col, + PrefixLength: row.SubPart, + }) } } grouped = append(grouped, gi) @@ -547,18 +533,161 @@ func groupIndexDefinitions(indexes []connection.IndexDefinition) []groupedIndex return grouped } -func sameColumnNameList(a, b []string) bool { +func sameColumnNameList(a []IndexMigrationColumn, b []string) bool { if len(a) == 0 || len(a) != len(b) { return false } for i := range a { - if !strings.EqualFold(strings.TrimSpace(a[i]), strings.TrimSpace(b[i])) { + if !strings.EqualFold(strings.TrimSpace(a[i].Name), strings.TrimSpace(b[i])) { return false } } return true } +func hasIndexPrefix(columns []IndexMigrationColumn) bool { + for _, column := range columns { + if column.PrefixLength > 0 { + return true + } + } + return false +} + +func buildQuotedIndexColumns(targetType string, columns []IndexMigrationColumn, preservePrefix bool) []string { + quoted := make([]string, 0, len(columns)) + for _, column := range columns { + name := strings.TrimSpace(column.Name) + if name == "" { + continue + } + value := quoteIdentByType(targetType, name) + if preservePrefix && column.PrefixLength > 0 { + value += fmt.Sprintf("(%d)", column.PrefixLength) + } + quoted = append(quoted, value) + } + return quoted +} + +func buildCreateIndexSQL(targetType, targetQueryTable string, idx groupedIndex, preservePrefix bool) string { + prefix := "CREATE INDEX" + if idx.Unique { + prefix = "CREATE UNIQUE INDEX" + } + return fmt.Sprintf("%s %s ON %s (%s)", + prefix, + quoteIdentByType(targetType, idx.Name), + quoteQualifiedIdentByType(targetType, targetQueryTable), + strings.Join(buildQuotedIndexColumns(targetType, idx.Columns, preservePrefix), ", "), + ) +} + +func buildIndexRemediationStatements(targetType, targetQueryTable string, idx groupedIndex) []string { + kind := strings.ToLower(strings.TrimSpace(idx.IndexType)) + if isMySQLRowStoreType(targetType) { + if kind == "fulltext" { + return []string{fmt.Sprintf("CREATE FULLTEXT INDEX %s ON %s (%s)", + quoteIdentByType(targetType, idx.Name), + quoteQualifiedIdentByType(targetType, targetQueryTable), + strings.Join(buildQuotedIndexColumns(targetType, idx.Columns, true), ", "), + )} + } + if hasIndexPrefix(idx.Columns) { + return []string{buildCreateIndexSQL(targetType, targetQueryTable, idx, true)} + } + return nil + } + if kind == "fulltext" && isPGLikeSameFamilyDDLType(targetType) { + parts := make([]string, 0, len(idx.Columns)) + for _, column := range idx.Columns { + parts = append(parts, fmt.Sprintf("coalesce(CAST(%s AS text), '')", quoteIdentByType(targetType, column.Name))) + } + return []string{fmt.Sprintf("CREATE INDEX %s ON %s USING GIN (to_tsvector('simple', %s))", + quoteIdentByType(targetType, idx.Name), + quoteQualifiedIdentByType(targetType, targetQueryTable), + strings.Join(parts, " || ' ' || "), + )} + } + if hasIndexPrefix(idx.Columns) && isPGLikeSameFamilyDDLType(targetType) { + columns := make([]string, 0, len(idx.Columns)) + for _, column := range idx.Columns { + value := quoteIdentByType(targetType, column.Name) + if column.PrefixLength > 0 { + value = fmt.Sprintf("left(CAST(%s AS text), %d)", value, column.PrefixLength) + } + columns = append(columns, value) + } + prefix := "CREATE INDEX" + if idx.Unique { + prefix = "CREATE UNIQUE INDEX" + } + return []string{fmt.Sprintf("%s %s ON %s (%s)", + prefix, + quoteIdentByType(targetType, idx.Name), + quoteQualifiedIdentByType(targetType, targetQueryTable), + strings.Join(columns, ", "), + )} + } + return nil +} + +func buildMySQLSourceIndexPlan(targetType, targetQueryTable string, indexes []connection.IndexDefinition) ([]string, []string, []UnmigratedIndex, int, int) { + grouped := groupIndexDefinitions(indexes) + postSQL := make([]string, 0, len(grouped)) + unsupported := make([]string, 0) + unmigrated := make([]UnmigratedIndex, 0) + created := 0 + skipped := 0 + for _, idx := range grouped { + name := strings.TrimSpace(idx.Name) + if name == "" || strings.EqualFold(name, "primary") { + continue + } + if len(idx.Columns) == 0 { + reason := fmt.Sprintf("索引 %s 缺少列定义，已跳过", name) + unsupported = append(unsupported, reason) + unmigrated = append(unmigrated, UnmigratedIndex{ + Name: name, + Columns: []IndexMigrationColumn{}, + Unique: idx.Unique, + IndexType: idx.IndexType, + ReasonCode: "missing_columns", + Reason: reason, + }) + skipped++ + continue + } + kind := strings.ToLower(strings.TrimSpace(idx.IndexType)) + if (kind == "" || kind == "btree") && !hasIndexPrefix(idx.Columns) { + postSQL = append(postSQL, buildCreateIndexSQL(targetType, targetQueryTable, idx, false)) + created++ + continue + } + + reasonCode := "unsupported_index_type" + reason := fmt.Sprintf("索引 %s 类型=%s，当前暂不支持等价自动迁移", name, idx.IndexType) + if hasIndexPrefix(idx.Columns) { + reasonCode = "prefix_index_requires_review" + reason = fmt.Sprintf("索引 %s 使用前缀长度，当前目标方言暂不支持等价自动迁移", name) + } else if kind == "fulltext" { + reasonCode = "fulltext_requires_review" + } + unsupported = append(unsupported, reason) + unmigrated = append(unmigrated, UnmigratedIndex{ + Name: name, + Columns: append([]IndexMigrationColumn(nil), idx.Columns...), + Unique: idx.Unique, + IndexType: idx.IndexType, + ReasonCode: reasonCode, + Reason: reason, + RemediationStatements: buildIndexRemediationStatements(targetType, targetQueryTable, idx), + }) + skipped++ + } + return postSQL, dedupeStrings(unsupported), unmigrated, created, skipped +} + func intFromAny(v interface{}) int { switch typed := v.(type) { case int: @@ -673,7 +802,7 @@ func buildMySQLToMySQLPlan(config SyncConfig, tableName string, sourceDB db.Data case "smart", "auto_create_if_missing": plan.AutoCreate = true plan.PlannedAction = "目标表不存在，将自动建表后导入" - createSQL, postSQL, warnings, unsupported, idxCreate, idxSkip, err := buildMySQLToMySQLCreateTablePlan(targetType, config, plan.TargetQueryTable, sourceCols, sourceDB, plan.SourceSchema, plan.SourceTable) + createSQL, postSQL, warnings, unsupported, unmigrated, idxCreate, idxSkip, err := buildMySQLToMySQLCreateTablePlan(targetType, config, plan.TargetQueryTable, sourceCols, sourceDB, plan.SourceSchema, plan.SourceTable) if err != nil { return plan, sourceCols, targetCols, err } @@ -681,6 +810,7 @@ func buildMySQLToMySQLPlan(config SyncConfig, tableName string, sourceDB db.Data plan.PostDataSQL = append(plan.PostDataSQL, postSQL...) plan.Warnings = append(plan.Warnings, warnings...) plan.UnsupportedObjects = append(plan.UnsupportedObjects, unsupported...) + plan.UnmigratedIndexes = append(plan.UnmigratedIndexes, unmigrated...) plan.IndexesToCreate = idxCreate plan.IndexesSkipped = idxSkip return dedupeSchemaMigrationPlan(plan), sourceCols, targetCols, nil @@ -689,7 +819,7 @@ func buildMySQLToMySQLPlan(config SyncConfig, tableName string, sourceDB db.Data } } -func buildMySQLToMySQLCreateTablePlan(targetType string, config SyncConfig, targetQueryTable string, sourceCols []connection.ColumnDefinition, sourceDB db.Database, sourceSchema, sourceTable string) (string, []string, []string, []string, int, int, error) { +func buildMySQLToMySQLCreateTablePlan(targetType string, config SyncConfig, targetQueryTable string, sourceCols []connection.ColumnDefinition, sourceDB db.Database, sourceSchema, sourceTable string) (string, []string, []string, []string, []UnmigratedIndex, int, int, error) { columnDefs := make([]string, 0, len(sourceCols)+1) warnings := make([]string, 0) unsupported := make([]string, 0) @@ -707,50 +837,15 @@ func buildMySQLToMySQLCreateTablePlan(targetType string, config SyncConfig, targ } createSQL := fmt.Sprintf("CREATE TABLE %s (\n %s\n)", quoteQualifiedIdentByType(targetType, targetQueryTable), strings.Join(columnDefs, ",\n ")) if !config.CreateIndexes { - return createSQL, nil, dedupeStrings(warnings), dedupeStrings(unsupported), 0, 0, nil + return createSQL, nil, dedupeStrings(warnings), dedupeStrings(unsupported), nil, 0, 0, nil } indexes, err := sourceDB.GetIndexes(sourceSchema, sourceTable) if err != nil { warnings = append(warnings, fmt.Sprintf("读取源表索引失败，已跳过索引迁移：%v", err)) - return createSQL, nil, dedupeStrings(warnings), dedupeStrings(unsupported), 0, 0, nil + return createSQL, nil, dedupeStrings(warnings), dedupeStrings(unsupported), nil, 0, 0, nil } - grouped := groupIndexDefinitions(indexes) - postSQL := make([]string, 0, len(grouped)) - created := 0 - skipped := 0 - for _, idx := range grouped { - name := strings.TrimSpace(idx.Name) - if name == "" || strings.EqualFold(name, "primary") { - continue - } - if len(idx.Columns) == 0 { - skipped++ - unsupported = append(unsupported, fmt.Sprintf("索引 %s 缺少列定义，已跳过", name)) - continue - } - kind := strings.ToLower(strings.TrimSpace(idx.IndexType)) - if idx.SubPart > 0 { - skipped++ - unsupported = append(unsupported, fmt.Sprintf("索引 %s 使用前缀长度，当前暂不支持迁移", name)) - continue - } - if kind != "" && kind != "btree" { - skipped++ - unsupported = append(unsupported, fmt.Sprintf("索引 %s 类型=%s，当前暂不支持自动迁移", name, idx.IndexType)) - continue - } - quotedCols := make([]string, 0, len(idx.Columns)) - for _, col := range idx.Columns { - quotedCols = append(quotedCols, quoteIdentByType(targetType, col)) - } - prefix := "CREATE INDEX" - if idx.Unique { - prefix = "CREATE UNIQUE INDEX" - } - postSQL = append(postSQL, fmt.Sprintf("%s %s ON %s (%s)", prefix, quoteIdentByType(targetType, name), quoteQualifiedIdentByType(targetType, targetQueryTable), strings.Join(quotedCols, ", "))) - created++ - } - return createSQL, postSQL, dedupeStrings(warnings), dedupeStrings(unsupported), created, skipped, nil + postSQL, unsupported, unmigrated, created, skipped := buildMySQLSourceIndexPlan(targetType, targetQueryTable, indexes) + return createSQL, postSQL, dedupeStrings(warnings), unsupported, unmigrated, created, skipped, nil } func buildMySQLToMySQLColumnDefinition(col connection.ColumnDefinition) (string, []string) { @@ -980,7 +1075,7 @@ func buildPGLikeToPGLikeCreateTablePlan(targetType string, config SyncConfig, ta continue } kind := strings.ToLower(strings.TrimSpace(idx.IndexType)) - if idx.SubPart > 0 { + if hasIndexPrefix(idx.Columns) { skipped++ unsupported = append(unsupported, fmt.Sprintf("索引 %s 使用前缀长度，当前暂不支持迁移", name)) continue @@ -992,7 +1087,7 @@ func buildPGLikeToPGLikeCreateTablePlan(targetType string, config SyncConfig, ta } quotedCols := make([]string, 0, len(idx.Columns)) for _, col := range idx.Columns { - quotedCols = append(quotedCols, quoteIdentByType(targetType, col)) + quotedCols = append(quotedCols, quoteIdentByType(targetType, col.Name)) } prefix := "CREATE INDEX" if idx.Unique { @@ -1230,7 +1325,7 @@ func buildPGLikeToMySQLCreateTablePlan(config SyncConfig, targetQueryTable strin continue } kind := strings.ToLower(strings.TrimSpace(idx.IndexType)) - if idx.SubPart > 0 { + if hasIndexPrefix(idx.Columns) { skipped++ unsupported = append(unsupported, fmt.Sprintf("索引 %s 使用前缀长度，当前暂不支持迁移", name)) continue @@ -1242,7 +1337,7 @@ func buildPGLikeToMySQLCreateTablePlan(config SyncConfig, targetQueryTable strin } quotedCols := make([]string, 0, len(idx.Columns)) for _, col := range idx.Columns { - quotedCols = append(quotedCols, quoteIdentByType("mysql", col)) + quotedCols = append(quotedCols, quoteIdentByType("mysql", col.Name)) } prefix := "CREATE INDEX" if idx.Unique { @@ -1434,7 +1529,7 @@ func buildMySQLToPGLikePlan(config SyncConfig, tableName string, sourceDB db.Dat case "smart", "auto_create_if_missing": plan.AutoCreate = true plan.PlannedAction = "目标表不存在，将自动建表后导入" - createSQL, postSQL, warnings, unsupported, idxCreate, idxSkip, err := buildMySQLToPGLikeCreateTablePlan(targetType, config, plan.TargetQueryTable, sourceCols, sourceDB, plan.SourceSchema, plan.SourceTable) + createSQL, postSQL, warnings, unsupported, unmigrated, idxCreate, idxSkip, err := buildMySQLToPGLikeCreateTablePlan(targetType, config, plan.TargetQueryTable, sourceCols, sourceDB, plan.SourceSchema, plan.SourceTable) if err != nil { return plan, sourceCols, targetCols, err } @@ -1442,6 +1537,7 @@ func buildMySQLToPGLikePlan(config SyncConfig, tableName string, sourceDB db.Dat plan.PostDataSQL = append(plan.PostDataSQL, postSQL...) plan.Warnings = append(plan.Warnings, warnings...) plan.UnsupportedObjects = append(plan.UnsupportedObjects, unsupported...) + plan.UnmigratedIndexes = append(plan.UnmigratedIndexes, unmigrated...) plan.IndexesToCreate = idxCreate plan.IndexesSkipped = idxSkip return dedupeSchemaMigrationPlan(plan), sourceCols, targetCols, nil @@ -1483,7 +1579,7 @@ func buildMySQLToPGLikeAddColumnSQL(targetType string, targetQueryTable string, return sqlList, dedupeStrings(warnings) } -func buildMySQLToPGLikeCreateTablePlan(targetType string, config SyncConfig, targetQueryTable string, sourceCols []connection.ColumnDefinition, sourceDB db.Database, sourceSchema, sourceTable string) (string, []string, []string, []string, int, int, error) { +func buildMySQLToPGLikeCreateTablePlan(targetType string, config SyncConfig, targetQueryTable string, sourceCols []connection.ColumnDefinition, sourceDB db.Database, sourceSchema, sourceTable string) (string, []string, []string, []string, []UnmigratedIndex, int, int, error) { columnDefs := make([]string, 0, len(sourceCols)+1) warnings := make([]string, 0) unsupported := make([]string, 0) @@ -1501,50 +1597,15 @@ func buildMySQLToPGLikeCreateTablePlan(targetType string, config SyncConfig, tar } createSQL := fmt.Sprintf("CREATE TABLE %s (\n %s\n)", quoteQualifiedIdentByType(targetType, targetQueryTable), strings.Join(columnDefs, ",\n ")) if !config.CreateIndexes { - return createSQL, nil, dedupeStrings(warnings), dedupeStrings(unsupported), 0, 0, nil + return createSQL, nil, dedupeStrings(warnings), dedupeStrings(unsupported), nil, 0, 0, nil } indexes, err := sourceDB.GetIndexes(sourceSchema, sourceTable) if err != nil { warnings = append(warnings, fmt.Sprintf("读取源表索引失败，已跳过索引迁移：%v", err)) - return createSQL, nil, dedupeStrings(warnings), dedupeStrings(unsupported), 0, 0, nil + return createSQL, nil, dedupeStrings(warnings), dedupeStrings(unsupported), nil, 0, 0, nil } - grouped := groupIndexDefinitions(indexes) - postSQL := make([]string, 0, len(grouped)) - created := 0 - skipped := 0 - for _, idx := range grouped { - name := strings.TrimSpace(idx.Name) - if name == "" || strings.EqualFold(name, "primary") { - continue - } - if len(idx.Columns) == 0 { - skipped++ - unsupported = append(unsupported, fmt.Sprintf("索引 %s 缺少列定义，已跳过", name)) - continue - } - kind := strings.ToLower(strings.TrimSpace(idx.IndexType)) - if idx.SubPart > 0 { - skipped++ - unsupported = append(unsupported, fmt.Sprintf("索引 %s 使用前缀长度，当前暂不支持迁移", name)) - continue - } - if kind != "" && kind != "btree" { - skipped++ - unsupported = append(unsupported, fmt.Sprintf("索引 %s 类型=%s，当前暂不支持自动迁移", name, idx.IndexType)) - continue - } - quotedCols := make([]string, 0, len(idx.Columns)) - for _, col := range idx.Columns { - quotedCols = append(quotedCols, quoteIdentByType(targetType, col)) - } - prefix := "CREATE INDEX" - if idx.Unique { - prefix = "CREATE UNIQUE INDEX" - } - postSQL = append(postSQL, fmt.Sprintf("%s %s ON %s (%s)", prefix, quoteIdentByType(targetType, name), quoteQualifiedIdentByType(targetType, targetQueryTable), strings.Join(quotedCols, ", "))) - created++ - } - return createSQL, postSQL, dedupeStrings(warnings), dedupeStrings(unsupported), created, skipped, nil + postSQL, unsupported, unmigrated, created, skipped := buildMySQLSourceIndexPlan(targetType, targetQueryTable, indexes) + return createSQL, postSQL, dedupeStrings(warnings), unsupported, unmigrated, created, skipped, nil } func buildMySQLToPGLikeColumnDefinition(col connection.ColumnDefinition) (string, []string) { diff --git a/internal/sync/schema_migration_test.go b/internal/sync/schema_migration_test.go index e17141ad..0e02b6c0 100644 --- a/internal/sync/schema_migration_test.go +++ b/internal/sync/schema_migration_test.go @@ -120,7 +120,7 @@ func TestBuildMySQLToKingbaseCreateTablePlan_GeneratesAndSkipsIndexes(t *testing {Name: "note", Type: "text", Nullable: "YES"}, } cfg := SyncConfig{CreateIndexes: true} - createSQL, postSQL, warnings, unsupported, idxCreate, idxSkip, err := buildMySQLToKingbaseCreateTablePlan(cfg, "public.orders", cols, sourceDB, "shop", "orders") + createSQL, postSQL, warnings, unsupported, unmigrated, idxCreate, idxSkip, err := buildMySQLToKingbaseCreateTablePlan(cfg, "public.orders", cols, sourceDB, "shop", "orders") if err != nil { t.Fatalf("buildMySQLToKingbaseCreateTablePlan returned error: %v", err) } @@ -140,12 +140,39 @@ func TestBuildMySQLToKingbaseCreateTablePlan_GeneratesAndSkipsIndexes(t *testing t.Fatalf("unexpected warnings: %v", warnings) } wantUnsupported := []string{ - "索引 idx_name_prefix 使用前缀长度，当前暂不支持迁移", - "索引 idx_fulltext_note 类型=FULLTEXT，当前暂不支持自动迁移", + "索引 idx_name_prefix 使用前缀长度，当前目标方言暂不支持等价自动迁移", + "索引 idx_fulltext_note 类型=FULLTEXT，当前暂不支持等价自动迁移", } if !reflect.DeepEqual(unsupported, wantUnsupported) { t.Fatalf("unexpected unsupported objects: got=%v want=%v", unsupported, wantUnsupported) } + wantUnmigrated := []UnmigratedIndex{ + { + Name: "idx_name_prefix", + Columns: []IndexMigrationColumn{{Name: "name", PrefixLength: 12}}, + Unique: false, + IndexType: "BTREE", + ReasonCode: "prefix_index_requires_review", + Reason: wantUnsupported[0], + RemediationStatements: []string{ + "CREATE INDEX idx_name_prefix ON public.orders (left(CAST(name AS text), 12))", + }, + }, + { + Name: "idx_fulltext_note", + Columns: []IndexMigrationColumn{{Name: "note"}}, + Unique: false, + IndexType: "FULLTEXT", + ReasonCode: "fulltext_requires_review", + Reason: wantUnsupported[1], + RemediationStatements: []string{ + "CREATE INDEX idx_fulltext_note ON public.orders USING GIN (to_tsvector('simple', coalesce(CAST(note AS text), '')))", + }, + }, + } + if !reflect.DeepEqual(unmigrated, wantUnmigrated) { + t.Fatalf("unexpected unmigrated indexes: got=%+v want=%+v", unmigrated, wantUnmigrated) + } } func TestBuildSchemaMigrationPlan_AutoCreateWhenTargetMissing(t *testing.T) { @@ -377,6 +404,124 @@ func TestBuildSchemaMigrationPlan_MySQLToMySQLAutoCreatesMissingTarget(t *testin } } +func TestBuildMySQLToMySQLCreateTablePlan_ListsCompositePrefixRemediation(t *testing.T) { + t.Parallel() + + sourceDB := &fakeMigrationDB{ + indexes: map[string][]connection.IndexDefinition{ + "shop.users": { + {Name: "idx_users_lookup", ColumnName: "email", NonUnique: 1, SeqInIndex: 2, IndexType: "BTREE"}, + {Name: "idx_users_lookup", ColumnName: "name", NonUnique: 1, SeqInIndex: 1, IndexType: "BTREE", SubPart: 12}, + {Name: "idx_users_lookup", ColumnName: "bio", NonUnique: 1, SeqInIndex: 3, IndexType: "BTREE", SubPart: 24}, + }, + }, + } + cols := []connection.ColumnDefinition{ + {Name: "name", Type: "varchar(128)", Nullable: "NO"}, + {Name: "email", Type: "varchar(255)", Nullable: "NO"}, + {Name: "bio", Type: "text", Nullable: "YES"}, + } + + _, postSQL, warnings, unsupported, unmigrated, idxCreate, idxSkip, err := buildMySQLToMySQLCreateTablePlan( + "mysql", + SyncConfig{CreateIndexes: true}, + "app.users", + cols, + sourceDB, + "shop", + "users", + ) + if err != nil { + t.Fatalf("buildMySQLToMySQLCreateTablePlan returned error: %v", err) + } + want := "CREATE INDEX `idx_users_lookup` ON `app`.`users` (`name`(12), `email`, `bio`(24))" + if len(postSQL) != 0 || idxCreate != 0 || idxSkip != 1 { + t.Fatalf("prefix index must require review: sql=%v create=%d skip=%d", postSQL, idxCreate, idxSkip) + } + if len(warnings) != 0 || len(unsupported) != 1 || len(unmigrated) != 1 { + t.Fatalf("unexpected warnings/unsupported: warnings=%v unsupported=%v unmigrated=%v", warnings, unsupported, unmigrated) + } + if !reflect.DeepEqual(unmigrated[0].RemediationStatements, []string{want}) { + t.Fatalf("unexpected prefix remediation: got=%v want=%v", unmigrated[0].RemediationStatements, []string{want}) + } +} + +func TestBuildMySQLToMySQLCreateTablePlan_ListsFulltextRemediation(t *testing.T) { + t.Parallel() + + sourceDB := &fakeMigrationDB{ + indexes: map[string][]connection.IndexDefinition{ + "shop.articles": { + {Name: "idx_fulltext_body", ColumnName: "title", NonUnique: 1, SeqInIndex: 1, IndexType: "FULLTEXT"}, + {Name: "idx_fulltext_body", ColumnName: "body", NonUnique: 1, SeqInIndex: 2, IndexType: "FULLTEXT"}, + }, + }, + } + cols := []connection.ColumnDefinition{ + {Name: "title", Type: "varchar(255)", Nullable: "NO"}, + {Name: "body", Type: "text", Nullable: "NO"}, + } + + _, postSQL, warnings, unsupported, unmigrated, idxCreate, idxSkip, err := buildMySQLToMySQLCreateTablePlan( + "mysql", + SyncConfig{CreateIndexes: true}, + "archive.articles", + cols, + sourceDB, + "shop", + "articles", + ) + if err != nil { + t.Fatalf("buildMySQLToMySQLCreateTablePlan returned error: %v", err) + } + if len(postSQL) != 0 || idxCreate != 0 || idxSkip != 1 { + t.Fatalf("fulltext index must require review: sql=%v create=%d skip=%d", postSQL, idxCreate, idxSkip) + } + if len(warnings) != 0 || len(unsupported) != 1 || len(unmigrated) != 1 { + t.Fatalf("unexpected fulltext summary: warnings=%v unsupported=%v unmigrated=%+v", warnings, unsupported, unmigrated) + } + wantSQL := "CREATE FULLTEXT INDEX `idx_fulltext_body` ON `archive`.`articles` (`title`, `body`)" + if !reflect.DeepEqual(unmigrated[0].RemediationStatements, []string{wantSQL}) { + t.Fatalf("unexpected fulltext remediation: got=%v want=%v", unmigrated[0].RemediationStatements, []string{wantSQL}) + } +} + +func TestBuildMongoIndexCommands_ListsUnmigratedIndexes(t *testing.T) { + t.Parallel() + + sourceDB := &fakeMigrationDB{ + indexes: map[string][]connection.IndexDefinition{ + "shop.articles": { + {Name: "idx_lookup", ColumnName: "category", NonUnique: 1, SeqInIndex: 2, IndexType: "BTREE"}, + {Name: "idx_lookup", ColumnName: "author", NonUnique: 1, SeqInIndex: 1, IndexType: "BTREE"}, + {Name: "idx_title_prefix", ColumnName: "title", NonUnique: 1, SeqInIndex: 1, IndexType: "BTREE", SubPart: 12}, + {Name: "idx_fulltext_body", ColumnName: "title", NonUnique: 1, SeqInIndex: 1, IndexType: "FULLTEXT"}, + {Name: "idx_fulltext_body", ColumnName: "body", NonUnique: 1, SeqInIndex: 2, IndexType: "FULLTEXT"}, + }, + }, + } + + commands, warnings, unsupported, unmigrated, created, skipped, err := buildMongoIndexCommands(sourceDB, "shop", "articles", "articles") + if err != nil { + t.Fatalf("buildMongoIndexCommands returned error: %v", err) + } + if created != 1 || skipped != 2 || len(commands) != 1 { + t.Fatalf("unexpected Mongo index summary: commands=%v create=%d skip=%d", commands, created, skipped) + } + if !strings.Contains(commands[0], `"key":{"author":1,"category":1}`) { + t.Fatalf("Mongo compound index order was not preserved: %s", commands[0]) + } + if len(warnings) != 0 || len(unsupported) != 2 || len(unmigrated) != 2 { + t.Fatalf("unexpected Mongo index warnings: warnings=%v unsupported=%v unmigrated=%+v", warnings, unsupported, unmigrated) + } + if unmigrated[0].ReasonCode != "prefix_index_requires_review" || !strings.Contains(strings.Join(unmigrated[0].RemediationStatements, "\n"), `"key":{"title":1}`) { + t.Fatalf("unexpected prefix remediation: %+v", unmigrated[0]) + } + if unmigrated[1].ReasonCode != "fulltext_requires_review" || !strings.Contains(strings.Join(unmigrated[1].RemediationStatements, "\n"), `"key":{"title":"text","body":"text"}`) { + t.Fatalf("unexpected fulltext remediation: %+v", unmigrated[1]) + } +} + func TestBuildSchemaMigrationPlan_PGLikeToPGLikeAutoCreatesMissingTarget(t *testing.T) { t.Parallel() @@ -536,7 +681,7 @@ func TestBuildMySQLToPGLikeCreateTablePlan_GeneratesPostgresDDL(t *testing.T) { {Name: "payload", Type: "json", Nullable: "YES"}, } cfg := SyncConfig{CreateIndexes: true} - createSQL, postSQL, warnings, unsupported, idxCreate, idxSkip, err := buildMySQLToPGLikeCreateTablePlan("postgres", cfg, "public.orders", cols, sourceDB, "shop", "orders") + createSQL, postSQL, warnings, unsupported, unmigrated, idxCreate, idxSkip, err := buildMySQLToPGLikeCreateTablePlan("postgres", cfg, "public.orders", cols, sourceDB, "shop", "orders") if err != nil { t.Fatalf("buildMySQLToPGLikeCreateTablePlan returned error: %v", err) } @@ -555,8 +700,8 @@ func TestBuildMySQLToPGLikeCreateTablePlan_GeneratesPostgresDDL(t *testing.T) { if len(postSQL) != 1 || !strings.Contains(postSQL[0], `CREATE INDEX "idx_orders_user"`) { t.Fatalf("unexpected post SQL: %v", postSQL) } - if len(warnings) != 0 || len(unsupported) != 0 { - t.Fatalf("unexpected warnings/unsupported: warnings=%v unsupported=%v", warnings, unsupported) + if len(warnings) != 0 || len(unsupported) != 0 || len(unmigrated) != 0 { + t.Fatalf("unexpected warnings/unsupported: warnings=%v unsupported=%v unmigrated=%v", warnings, unsupported, unmigrated) } }