diff --git a/frontend/src/components/DataSyncModal.tsx b/frontend/src/components/DataSyncModal.tsx
index 82f339f..4fcadc3 100644
--- a/frontend/src/components/DataSyncModal.tsx
+++ b/frontend/src/components/DataSyncModal.tsx
@@ -975,7 +975,7 @@ const DataSyncModal: React.FC<{ open: boolean; onClose: () => void }> = ({ open,
)}
setAutoAddColumns(e.target.checked)} disabled={isSourceQueryMode}>
- 自动补齐目标表缺失字段（当前支持 MySQL 目标及 MySQL → Kingbase；SQL 结果集模式暂不支持）
+ 自动补齐目标表缺失字段（按源/目标数据源选择可兼容规划器；SQL 结果集模式暂不支持）
@@ -987,7 +987,7 @@ const DataSyncModal: React.FC<{ open: boolean; onClose: () => void }> = ({ open,
)}
diff --git a/internal/sync/migration_clickhouse.go b/internal/sync/migration_clickhouse.go
index d67fcef..1c5f665 100644
--- a/internal/sync/migration_clickhouse.go
+++ b/internal/sync/migration_clickhouse.go
@@ -243,6 +243,104 @@ func buildClickHouseToPGLikePlan(config SyncConfig, tableName string, sourceDB d
}
}
+func buildClickHouseToClickHousePlan(config SyncConfig, tableName string, sourceDB db.Database, targetDB db.Database) (SchemaMigrationPlan, []connection.ColumnDefinition, []connection.ColumnDefinition, error) {
+ plan := SchemaMigrationPlan{}
+ sourceType := resolveMigrationDBType(config.SourceConfig)
+ targetType := resolveMigrationDBType(config.TargetConfig)
+ plan.SourceSchema, plan.SourceTable = normalizeSchemaAndTable(sourceType, config.SourceConfig.Database, tableName)
+ plan.TargetSchema, plan.TargetTable = normalizeSchemaAndTable(targetType, config.TargetConfig.Database, tableName)
+ plan.SourceQueryTable = qualifiedNameForQuery(sourceType, plan.SourceSchema, plan.SourceTable, tableName)
+ plan.TargetQueryTable = qualifiedNameForQuery(targetType, plan.TargetSchema, plan.TargetTable, tableName)
+ plan.PlannedAction = "使用已有目标表导入"
+
+ sourceCols, sourceExists, err := inspectTableColumns(sourceDB, plan.SourceSchema, plan.SourceTable)
+ if err != nil {
+ return plan, nil, nil, fmt.Errorf("获取源表字段失败: %w", err)
+ }
+ if !sourceExists {
+ return plan, nil, nil, fmt.Errorf("源表不存在或无列定义: %s", tableName)
+ }
+
+ targetCols, targetExists, err := inspectTableColumns(targetDB, plan.TargetSchema, plan.TargetTable)
+ if err != nil {
+ return plan, sourceCols, nil, fmt.Errorf("获取目标表字段失败: %w", err)
+ }
+ plan.TargetTableExists = targetExists
+
+ strategy := normalizeTargetTableStrategy(config.TargetTableStrategy)
+ if targetExists {
+ missing := diffMissingColumnNames(sourceCols, targetCols)
+ if len(missing) > 0 {
+ plan.Warnings = append(plan.Warnings, fmt.Sprintf("目标表缺失字段 %d 个：%s", len(missing), strings.Join(missing, ", ")))
+ }
+ if len(missing) == 0 {
+ plan.PlannedAction = "表结构已一致"
+ } else if config.AutoAddColumns {
+ addSQL, addWarnings := buildClickHouseToClickHouseAddColumnSQL(plan.TargetQueryTable, sourceCols, targetCols)
+ plan.PreDataSQL = append(plan.PreDataSQL, addSQL...)
+ plan.Warnings = append(plan.Warnings, addWarnings...)
+ if len(addSQL) > 0 {
+ plan.PlannedAction = fmt.Sprintf("补齐缺失字段(%d)后导入", len(addSQL))
+ }
+ } else {
+ plan.PlannedAction = fmt.Sprintf("目标表缺失字段(%d)，未开启自动补齐", len(missing))
+ }
+ if strategy != "existing_only" {
+ plan.Warnings = append(plan.Warnings, "目标表已存在，当前仅执行数据导入；不会自动重建 ClickHouse ORDER BY/PARTITION/TTL 等表级语义")
+ }
+ return dedupeSchemaMigrationPlan(plan), sourceCols, targetCols, nil
+ }
+
+ switch strategy {
+ case "existing_only":
+ plan.PlannedAction = "目标表不存在，需先手工创建"
+ plan.Warnings = append(plan.Warnings, "当前策略要求目标表已存在，执行时不会自动建表")
+ return dedupeSchemaMigrationPlan(plan), sourceCols, targetCols, nil
+ case "smart", "auto_create_if_missing":
+ plan.AutoCreate = true
+ plan.PlannedAction = "目标表不存在，将自动建表后导入"
+ createSQL, warnings, unsupported := buildClickHouseToClickHouseCreateTableSQL(plan.TargetQueryTable, sourceCols)
+ plan.CreateTableSQL = createSQL
+ plan.Warnings = append(plan.Warnings, warnings...)
+ plan.UnsupportedObjects = append(plan.UnsupportedObjects, unsupported...)
+ return dedupeSchemaMigrationPlan(plan), sourceCols, targetCols, nil
+ default:
+ return dedupeSchemaMigrationPlan(plan), sourceCols, targetCols, nil
+ }
+}
+
+func buildClickHouseToClickHouseAddColumnSQL(targetQueryTable string, sourceCols, targetCols []connection.ColumnDefinition) ([]string, []string) {
+ targetSet := make(map[string]struct{}, len(targetCols))
+ for _, col := range targetCols {
+ key := strings.ToLower(strings.TrimSpace(col.Name))
+ if key == "" {
+ continue
+ }
+ targetSet[key] = struct{}{}
+ }
+ var sqlList []string
+ var warnings []string
+ for _, col := range sourceCols {
+ key := strings.ToLower(strings.TrimSpace(col.Name))
+ if key == "" {
+ continue
+ }
+ if _, ok := targetSet[key]; ok {
+ continue
+ }
+ colType := sanitizeClickHouseColumnType(col.Type)
+ sqlList = append(sqlList, fmt.Sprintf("ALTER TABLE %s ADD COLUMN %s %s",
+ quoteQualifiedIdentByType("clickhouse", targetQueryTable),
+ quoteIdentByType("clickhouse", col.Name),
+ colType,
+ ))
+ if strings.TrimSpace(col.Type) != colType {
+ warnings = append(warnings, fmt.Sprintf("字段 %s 类型为空或包含不安全字符，已降级为 %s", col.Name, colType))
+ }
+ }
+ return sqlList, dedupeStrings(warnings)
+}
+
func buildPGLikeToClickHouseAddColumnSQL(targetQueryTable string, sourceCols, targetCols []connection.ColumnDefinition) ([]string, []string) {
targetSet := make(map[string]struct{}, len(targetCols))
for _, col := range targetCols {
@@ -454,6 +552,64 @@ func buildClickHouseToMySQLCreateTableSQL(targetQueryTable string, sourceCols []
return createSQL, dedupeStrings(warnings)
}
+func buildClickHouseToClickHouseCreateTableSQL(targetQueryTable string, sourceCols []connection.ColumnDefinition) (string, []string, []string) {
+ columnDefs := make([]string, 0, len(sourceCols))
+ warnings := make([]string, 0)
+ unsupported := []string{"ClickHouse 同库迁移当前按列元数据重建基础 MergeTree 表；PARTITION BY/TTL/Projection/物化视图/表设置不会自动迁移"}
+ orderByCols := make([]string, 0)
+ for _, col := range sourceCols {
+ def, colWarnings := buildClickHouseToClickHouseColumnDefinition(col)
+ warnings = append(warnings, colWarnings...)
+ columnDefs = append(columnDefs, fmt.Sprintf("%s %s", quoteIdentByType("clickhouse", col.Name), def))
+ if col.Key == "PRI" || col.Key == "PK" || col.Key == "MUL" {
+ orderByCols = append(orderByCols, quoteIdentByType("clickhouse", col.Name))
+ }
+ }
+ orderExpr := "tuple()"
+ if len(orderByCols) > 0 {
+ orderExpr = "(" + strings.Join(orderByCols, ", ") + ")"
+ } else {
+ warnings = append(warnings, "源表未返回排序键，ClickHouse 将使用 ORDER BY tuple() 建表")
+ }
+ createSQL := fmt.Sprintf("CREATE TABLE %s (\n %s\n) ENGINE = MergeTree() ORDER BY %s", quoteQualifiedIdentByType("clickhouse", targetQueryTable), strings.Join(columnDefs, ",\n "), orderExpr)
+ return createSQL, dedupeStrings(warnings), dedupeStrings(unsupported)
+}
+
+func buildClickHouseToClickHouseColumnDefinition(col connection.ColumnDefinition) (string, []string) {
+ targetType := sanitizeClickHouseColumnType(col.Type)
+ parts := []string{targetType}
+ warnings := make([]string, 0)
+ if strings.TrimSpace(col.Type) != targetType {
+ warnings = append(warnings, fmt.Sprintf("字段 %s 类型为空或包含不安全字符，已降级为 %s", col.Name, targetType))
+ }
+ extra := strings.ToUpper(strings.TrimSpace(col.Extra))
+ if extra == "MATERIALIZED" || extra == "ALIAS" {
+ warnings = append(warnings, fmt.Sprintf("字段 %s 为 %s 表达式列，当前仅迁移字段类型，不内联表达式", col.Name, extra))
+ } else if col.Default != nil {
+ rawDefault := strings.TrimSpace(*col.Default)
+ if rawDefault != "" && !strings.ContainsAny(rawDefault, ";\n\r") {
+ parts = append(parts, "DEFAULT "+rawDefault)
+ } else if rawDefault != "" {
+ warnings = append(warnings, fmt.Sprintf("字段 %s 的默认值包含不安全字符，当前未自动迁移", col.Name))
+ }
+ }
+ if comment := strings.TrimSpace(col.Comment); comment != "" {
+ parts = append(parts, "COMMENT '"+escapeMySQLStringLiteral(comment)+"'")
+ }
+ return strings.Join(parts, " "), dedupeStrings(warnings)
+}
+
+func sanitizeClickHouseColumnType(t string) string {
+ tt := strings.TrimSpace(t)
+ if tt == "" {
+ return "String"
+ }
+ if strings.ContainsAny(tt, "`;\n\r") {
+ return "String"
+ }
+ return tt
+}
+
func buildPGLikeToClickHouseColumnDefinition(col connection.ColumnDefinition) (string, []string) {
targetType, warnings := mapPGLikeColumnToClickHouse(col)
parts := []string{targetType}
diff --git a/internal/sync/migration_kernel_router.go b/internal/sync/migration_kernel_router.go
index bd1c8b1..aff5973 100644
--- a/internal/sync/migration_kernel_router.go
+++ b/internal/sync/migration_kernel_router.go
@@ -9,6 +9,14 @@ import (
type genericLegacyPlanner struct{}
+type mysqlToMySQLPlanner struct{}
+
+type pgLikeToPGLikePlanner struct{}
+
+type clickHouseToClickHousePlanner struct{}
+
+type mongoToMongoPlanner struct{}
+
type mysqlToPGLikePlanner struct{}
type mysqlToClickHousePlanner struct{}
@@ -55,6 +63,10 @@ func buildSchemaMigrationPlan(config SyncConfig, tableName string, sourceDB db.D
func resolveMigrationPlanner(ctx MigrationBuildContext) MigrationPlanner {
planners := []MigrationPlanner{
+ mysqlToMySQLPlanner{},
+ pgLikeToPGLikePlanner{},
+ clickHouseToClickHousePlanner{},
+ mongoToMongoPlanner{},
mysqlToPGLikePlanner{},
mySQLLikeToTDenginePlanner{},
pgLikeToTDenginePlanner{},
@@ -133,6 +145,66 @@ func (genericLegacyPlanner) BuildPlan(ctx MigrationBuildContext) (SchemaMigratio
return buildSchemaMigrationPlanLegacy(ctx.Config, ctx.TableName, ctx.SourceDB, ctx.TargetDB)
}
+func (mysqlToMySQLPlanner) Name() string { return "mysql-mysql-planner" }
+
+func (mysqlToMySQLPlanner) SupportLevel(ctx MigrationBuildContext) MigrationSupportLevel {
+ sourceType := resolveMigrationDBType(ctx.Config.SourceConfig)
+ targetType := resolveMigrationDBType(ctx.Config.TargetConfig)
+ if isMySQLRowStoreType(sourceType) && isMySQLRowStoreType(targetType) {
+ return MigrationSupportLevelFull
+ }
+ return MigrationSupportLevelUnsupported
+}
+
+func (mysqlToMySQLPlanner) BuildPlan(ctx MigrationBuildContext) (SchemaMigrationPlan, []connection.ColumnDefinition, []connection.ColumnDefinition, error) {
+ return buildMySQLToMySQLPlan(ctx.Config, ctx.TableName, ctx.SourceDB, ctx.TargetDB)
+}
+
+func (pgLikeToPGLikePlanner) Name() string { return "pglike-pglike-planner" }
+
+func (pgLikeToPGLikePlanner) SupportLevel(ctx MigrationBuildContext) MigrationSupportLevel {
+ sourceType := resolveMigrationDBType(ctx.Config.SourceConfig)
+ targetType := resolveMigrationDBType(ctx.Config.TargetConfig)
+ if isPGLikeSameFamilyDDLType(sourceType) && isPGLikeSameFamilyDDLType(targetType) {
+ return MigrationSupportLevelFull
+ }
+ return MigrationSupportLevelUnsupported
+}
+
+func (pgLikeToPGLikePlanner) BuildPlan(ctx MigrationBuildContext) (SchemaMigrationPlan, []connection.ColumnDefinition, []connection.ColumnDefinition, error) {
+ return buildPGLikeToPGLikePlan(ctx.Config, ctx.TableName, ctx.SourceDB, ctx.TargetDB)
+}
+
+func (clickHouseToClickHousePlanner) Name() string { return "clickhouse-clickhouse-planner" }
+
+func (clickHouseToClickHousePlanner) SupportLevel(ctx MigrationBuildContext) MigrationSupportLevel {
+ sourceType := resolveMigrationDBType(ctx.Config.SourceConfig)
+ targetType := resolveMigrationDBType(ctx.Config.TargetConfig)
+ if sourceType == "clickhouse" && targetType == "clickhouse" {
+ return MigrationSupportLevelFull
+ }
+ return MigrationSupportLevelUnsupported
+}
+
+func (clickHouseToClickHousePlanner) BuildPlan(ctx MigrationBuildContext) (SchemaMigrationPlan, []connection.ColumnDefinition, []connection.ColumnDefinition, error) {
+ return buildClickHouseToClickHousePlan(ctx.Config, ctx.TableName, ctx.SourceDB, ctx.TargetDB)
+}
+
+func (mongoToMongoPlanner) Name() string { return "mongo-mongo-planner" }
+
+func (mongoToMongoPlanner) SupportLevel(ctx MigrationBuildContext) MigrationSupportLevel {
+ sourceType := resolveMigrationDBType(ctx.Config.SourceConfig)
+ targetType := resolveMigrationDBType(ctx.Config.TargetConfig)
+ if sourceType == "mongodb" && targetType == "mongodb" {
+ return MigrationSupportLevelFull
+ }
+ return MigrationSupportLevelUnsupported
+}
+
+func (mongoToMongoPlanner) BuildPlan(ctx MigrationBuildContext) (SchemaMigrationPlan, []connection.ColumnDefinition, []connection.ColumnDefinition, error) {
+ return buildMongoToMongoPlan(ctx.Config, ctx.TableName, ctx.SourceDB, ctx.TargetDB)
+}
+
func (mysqlToPGLikePlanner) Name() string { return "mysql-pglike-planner" }
func (mysqlToPGLikePlanner) SupportLevel(ctx MigrationBuildContext) MigrationSupportLevel {
diff --git a/internal/sync/migration_kernel_router_test.go b/internal/sync/migration_kernel_router_test.go
index 2d372be..b812aa3 100644
--- a/internal/sync/migration_kernel_router_test.go
+++ b/internal/sync/migration_kernel_router_test.go
@@ -102,6 +102,91 @@ func TestResolveMigrationPlanner_UsesPGLikeMySQLPlannerForKingbaseToMySQL(t *tes
}
}
+func TestResolveMigrationPlanner_UsesMySQLMySQLPlannerForMySQLToMySQL(t *testing.T) {
+ t.Parallel()
+
+ planner := resolveMigrationPlanner(MigrationBuildContext{
+ Config: SyncConfig{
+ SourceConfig: connection.ConnectionConfig{Type: "mysql"},
+ TargetConfig: connection.ConnectionConfig{Type: "mysql"},
+ },
+ })
+ if planner == nil {
+ t.Fatalf("expected planner")
+ }
+ if planner.Name() != "mysql-mysql-planner" {
+ t.Fatalf("unexpected planner: %s", planner.Name())
+ }
+}
+
+func TestResolveMigrationPlanner_UsesPGLikePGLikePlannerForPostgresToKingbase(t *testing.T) {
+ t.Parallel()
+
+ planner := resolveMigrationPlanner(MigrationBuildContext{
+ Config: SyncConfig{
+ SourceConfig: connection.ConnectionConfig{Type: "postgres"},
+ TargetConfig: connection.ConnectionConfig{Type: "kingbase"},
+ },
+ })
+ if planner == nil {
+ t.Fatalf("expected planner")
+ }
+ if planner.Name() != "pglike-pglike-planner" {
+ t.Fatalf("unexpected planner: %s", planner.Name())
+ }
+}
+
+func TestResolveMigrationPlanner_DoesNotUsePGLikePGLikePlannerForPostgresToDuckDB(t *testing.T) {
+ t.Parallel()
+
+ planner := resolveMigrationPlanner(MigrationBuildContext{
+ Config: SyncConfig{
+ SourceConfig: connection.ConnectionConfig{Type: "postgres"},
+ TargetConfig: connection.ConnectionConfig{Type: "duckdb"},
+ },
+ })
+ if planner == nil {
+ t.Fatalf("expected planner")
+ }
+ if planner.Name() == "pglike-pglike-planner" {
+ t.Fatalf("duckdb should not use pglike same-family planner")
+ }
+}
+
+func TestResolveMigrationPlanner_UsesClickHouseClickHousePlanner(t *testing.T) {
+ t.Parallel()
+
+ planner := resolveMigrationPlanner(MigrationBuildContext{
+ Config: SyncConfig{
+ SourceConfig: connection.ConnectionConfig{Type: "clickhouse"},
+ TargetConfig: connection.ConnectionConfig{Type: "clickhouse"},
+ },
+ })
+ if planner == nil {
+ t.Fatalf("expected planner")
+ }
+ if planner.Name() != "clickhouse-clickhouse-planner" {
+ t.Fatalf("unexpected planner: %s", planner.Name())
+ }
+}
+
+func TestResolveMigrationPlanner_UsesMongoMongoPlanner(t *testing.T) {
+ t.Parallel()
+
+ planner := resolveMigrationPlanner(MigrationBuildContext{
+ Config: SyncConfig{
+ SourceConfig: connection.ConnectionConfig{Type: "mongodb"},
+ TargetConfig: connection.ConnectionConfig{Type: "mongodb"},
+ },
+ })
+ if planner == nil {
+ t.Fatalf("expected planner")
+ }
+ if planner.Name() != "mongo-mongo-planner" {
+ t.Fatalf("unexpected planner: %s", planner.Name())
+ }
+}
+
func TestResolveMigrationPlanner_UsesMySQLPGLikePlannerForMySQLToPostgres(t *testing.T) {
t.Parallel()
diff --git a/internal/sync/migration_mongodb.go b/internal/sync/migration_mongodb.go
index 23a97c4..8cb1422 100644
--- a/internal/sync/migration_mongodb.go
+++ b/internal/sync/migration_mongodb.go
@@ -87,6 +87,71 @@ func buildTabularToMongoPlan(config SyncConfig, tableName string, sourceDB db.Da
}
}
+func buildMongoToMongoPlan(config SyncConfig, tableName string, sourceDB db.Database, targetDB db.Database) (SchemaMigrationPlan, []connection.ColumnDefinition, []connection.ColumnDefinition, error) {
+ plan := SchemaMigrationPlan{}
+ sourceType := resolveMigrationDBType(config.SourceConfig)
+ targetType := resolveMigrationDBType(config.TargetConfig)
+ plan.SourceSchema, plan.SourceTable = normalizeSchemaAndTable(sourceType, config.SourceConfig.Database, tableName)
+ plan.TargetSchema, plan.TargetTable = normalizeSchemaAndTable(targetType, config.TargetConfig.Database, tableName)
+ plan.SourceQueryTable = qualifiedNameForQuery(sourceType, plan.SourceSchema, plan.SourceTable, tableName)
+ plan.TargetQueryTable = qualifiedNameForQuery(targetType, plan.TargetSchema, plan.TargetTable, tableName)
+ plan.PlannedAction = "使用已有目标集合导入"
+
+ sourceCols, warnings, err := inferMongoCollectionColumns(sourceDB, plan.SourceTable)
+ if err != nil {
+ return plan, nil, nil, err
+ }
+ plan.Warnings = append(plan.Warnings, warnings...)
+ if len(sourceCols) == 0 {
+ return plan, nil, nil, fmt.Errorf("源集合未推断出可迁移字段: %s", tableName)
+ }
+
+ targetExists, err := inspectMongoCollection(targetDB, plan.TargetSchema, plan.TargetTable)
+ if err != nil {
+ return plan, sourceCols, nil, fmt.Errorf("检查目标集合失败: %w", err)
+ }
+ plan.TargetTableExists = targetExists
+
+ strategy := normalizeTargetTableStrategy(config.TargetTableStrategy)
+ if targetExists {
+ plan.Warnings = append(plan.Warnings, "MongoDB 为弱 schema 目标，字段结构以写入文档为准，不执行目标列校验")
+ if strategy != "existing_only" {
+ plan.Warnings = append(plan.Warnings, "目标集合已存在，当前仅执行数据导入；不会自动重建已有索引")
+ }
+ return dedupeSchemaMigrationPlan(plan), sourceCols, nil, nil
+ }
+
+ switch strategy {
+ case "existing_only":
+ plan.PlannedAction = "目标集合不存在，需先手工创建"
+ plan.Warnings = append(plan.Warnings, "当前策略要求目标集合已存在，执行时不会自动创建")
+ return dedupeSchemaMigrationPlan(plan), sourceCols, nil, nil
+ case "smart", "auto_create_if_missing":
+ plan.AutoCreate = true
+ plan.PlannedAction = "目标集合不存在，将自动创建集合后导入"
+ createCmd, err := buildMongoCreateCollectionCommand(plan.TargetTable)
+ if err != nil {
+ return plan, sourceCols, nil, err
+ }
+ plan.PreDataSQL = append(plan.PreDataSQL, createCmd)
+ if config.CreateIndexes {
+ indexCmds, indexWarnings, unsupported, 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.IndexesToCreate = created
+ plan.IndexesSkipped = skipped
+ }
+ }
+ return dedupeSchemaMigrationPlan(plan), sourceCols, nil, nil
+ default:
+ return dedupeSchemaMigrationPlan(plan), sourceCols, nil, nil
+ }
+}
+
func buildMongoToMySQLPlan(config SyncConfig, tableName string, sourceDB db.Database, targetDB db.Database) (SchemaMigrationPlan, []connection.ColumnDefinition, []connection.ColumnDefinition, error) {
plan := SchemaMigrationPlan{}
plan.SourceSchema, plan.SourceTable = normalizeSchemaAndTable(config.SourceConfig.Type, config.SourceConfig.Database, tableName)
diff --git a/internal/sync/migration_runtime_helpers.go b/internal/sync/migration_runtime_helpers.go
index 418080c..cd3ae84 100644
--- a/internal/sync/migration_runtime_helpers.go
+++ b/internal/sync/migration_runtime_helpers.go
@@ -12,9 +12,15 @@ func supportsAutoAddColumnsForPair(sourceType string, targetType string) bool {
if isMySQLLikeWritableTargetType(target) {
return isMySQLCoreType(source)
}
+ if isPGLikeSameFamilyDDLType(source) && isPGLikeSameFamilyDDLType(target) {
+ return true
+ }
if isPGLikeTarget(target) {
return isMySQLLikeSourceType(source)
}
+ if source == "clickhouse" && target == "clickhouse" {
+ return true
+ }
return false
}
@@ -39,6 +45,20 @@ func buildAddColumnSQLForPair(sourceType string, targetType string, targetQueryT
quoteIdentByType(target, sourceCol.Name),
colType,
), nil
+ case isPGLikeSameFamilyDDLType(source) && isPGLikeSameFamilyDDLType(target):
+ colType := sanitizePGLikeColumnType(sourceCol.Type)
+ return fmt.Sprintf("ALTER TABLE %s ADD COLUMN %s %s NULL",
+ quoteQualifiedIdentByType(target, targetQueryTable),
+ quoteIdentByType(target, sourceCol.Name),
+ colType,
+ ), nil
+ case source == "clickhouse" && target == "clickhouse":
+ colType := sanitizeClickHouseColumnType(sourceCol.Type)
+ return fmt.Sprintf("ALTER TABLE %s ADD COLUMN %s %s",
+ quoteQualifiedIdentByType(target, targetQueryTable),
+ quoteIdentByType(target, sourceCol.Name),
+ colType,
+ ), nil
default:
return "", fmt.Errorf("当前不支持 source=%s target=%s 的自动补字段", sourceType, targetType)
}
diff --git a/internal/sync/migration_type_resolver.go b/internal/sync/migration_type_resolver.go
index 3a59907..44e9185 100644
--- a/internal/sync/migration_type_resolver.go
+++ b/internal/sync/migration_type_resolver.go
@@ -102,6 +102,15 @@ func isMySQLCoreType(dbType string) bool {
}
}
+func isMySQLRowStoreType(dbType string) bool {
+ switch normalizeMigrationDBType(dbType) {
+ case "mysql", "mariadb", "oceanbase":
+ return true
+ default:
+ return false
+ }
+}
+
func isMySQLLikeSourceType(dbType string) bool {
if isMySQLCoreType(dbType) {
return true
diff --git a/internal/sync/schema_migration.go b/internal/sync/schema_migration.go
index 39a6acd..9cde70f 100644
--- a/internal/sync/schema_migration.go
+++ b/internal/sync/schema_migration.go
@@ -179,7 +179,7 @@ func buildSchemaMigrationPlanLegacy(config SyncConfig, tableName string, sourceD
case "smart", "auto_create_if_missing":
if !supportsAutoCreateMigration(config.SourceConfig.Type, config.TargetConfig.Type) {
plan.PlannedAction = "当前库对暂不支持自动建表"
- plan.Warnings = append(plan.Warnings, fmt.Sprintf("当前仅支持 MySQL -> Kingbase 自动建表，当前组合=%s -> %s", config.SourceConfig.Type, config.TargetConfig.Type))
+ plan.Warnings = append(plan.Warnings, fmt.Sprintf("当前组合未接入专用自动建表规划器：%s -> %s", config.SourceConfig.Type, config.TargetConfig.Type))
return dedupeSchemaMigrationPlan(plan), sourceCols, targetCols, nil
}
plan.AutoCreate = true
@@ -543,6 +543,18 @@ func groupIndexDefinitions(indexes []connection.IndexDefinition) []groupedIndex
return grouped
}
+func sameColumnNameList(a, 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])) {
+ return false
+ }
+ }
+ return true
+}
+
func intFromAny(v interface{}) int {
switch typed := v.(type) {
case int:
@@ -568,6 +580,500 @@ func isPGLikeSource(dbType string) bool {
}
}
+func isPGLikeSameFamilyDDLType(dbType string) bool {
+ switch normalizeMigrationDBType(dbType) {
+ case "postgres", "kingbase", "highgo", "vastbase", "opengauss":
+ return true
+ default:
+ return false
+ }
+}
+
+func buildMySQLToMySQLPlan(config SyncConfig, tableName string, sourceDB db.Database, targetDB db.Database) (SchemaMigrationPlan, []connection.ColumnDefinition, []connection.ColumnDefinition, error) {
+ plan := SchemaMigrationPlan{}
+ sourceType := resolveMigrationDBType(config.SourceConfig)
+ targetType := resolveMigrationDBType(config.TargetConfig)
+ plan.SourceSchema, plan.SourceTable = normalizeSchemaAndTable(sourceType, config.SourceConfig.Database, tableName)
+ plan.TargetSchema, plan.TargetTable = normalizeSchemaAndTable(targetType, config.TargetConfig.Database, tableName)
+ plan.SourceQueryTable = qualifiedNameForQuery(sourceType, plan.SourceSchema, plan.SourceTable, tableName)
+ plan.TargetQueryTable = qualifiedNameForQuery(targetType, plan.TargetSchema, plan.TargetTable, tableName)
+ plan.PlannedAction = "使用已有目标表导入"
+
+ sourceCols, sourceExists, err := inspectTableColumns(sourceDB, plan.SourceSchema, plan.SourceTable)
+ if err != nil {
+ return plan, nil, nil, fmt.Errorf("获取源表字段失败: %w", err)
+ }
+ if !sourceExists {
+ return plan, nil, nil, fmt.Errorf("源表不存在或无列定义: %s", tableName)
+ }
+
+ targetCols, targetExists, err := inspectTableColumns(targetDB, plan.TargetSchema, plan.TargetTable)
+ if err != nil {
+ return plan, sourceCols, nil, fmt.Errorf("获取目标表字段失败: %w", err)
+ }
+ plan.TargetTableExists = targetExists
+
+ strategy := normalizeTargetTableStrategy(config.TargetTableStrategy)
+ if targetExists {
+ missing := diffMissingColumnNames(sourceCols, targetCols)
+ if len(missing) > 0 {
+ plan.Warnings = append(plan.Warnings, fmt.Sprintf("目标表缺失字段 %d 个：%s", len(missing), strings.Join(missing, ", ")))
+ }
+ if len(missing) == 0 {
+ plan.PlannedAction = "表结构已一致"
+ } else if config.AutoAddColumns {
+ targetSet := make(map[string]struct{}, len(targetCols))
+ for _, col := range targetCols {
+ key := strings.ToLower(strings.TrimSpace(col.Name))
+ if key == "" {
+ continue
+ }
+ targetSet[key] = struct{}{}
+ }
+ for _, col := range sourceCols {
+ key := strings.ToLower(strings.TrimSpace(col.Name))
+ if key == "" {
+ continue
+ }
+ if _, ok := targetSet[key]; ok {
+ continue
+ }
+ addSQL, err := buildAddColumnSQLForPair(sourceType, targetType, plan.TargetQueryTable, col)
+ if err != nil {
+ plan.Warnings = append(plan.Warnings, fmt.Sprintf("字段 %s 自动补齐 SQL 生成失败：%v", col.Name, err))
+ continue
+ }
+ plan.PreDataSQL = append(plan.PreDataSQL, addSQL)
+ }
+ if len(plan.PreDataSQL) > 0 {
+ plan.PlannedAction = fmt.Sprintf("补齐缺失字段(%d)后导入", len(plan.PreDataSQL))
+ } else {
+ plan.PlannedAction = fmt.Sprintf("目标表缺失字段(%d)，但未生成可执行补齐 SQL", len(missing))
+ }
+ } else {
+ plan.PlannedAction = fmt.Sprintf("目标表缺失字段(%d)，未开启自动补齐", len(missing))
+ }
+ if strategy != "existing_only" {
+ plan.Warnings = append(plan.Warnings, "目标表已存在，当前仅执行数据导入；不会自动重建已有索引/约束")
+ }
+ return dedupeSchemaMigrationPlan(plan), sourceCols, targetCols, nil
+ }
+
+ switch strategy {
+ case "existing_only":
+ plan.PlannedAction = "目标表不存在，需先手工创建"
+ plan.Warnings = append(plan.Warnings, "当前策略要求目标表已存在，执行时不会自动建表")
+ return dedupeSchemaMigrationPlan(plan), sourceCols, targetCols, nil
+ 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)
+ if err != nil {
+ return plan, sourceCols, targetCols, err
+ }
+ plan.CreateTableSQL = createSQL
+ plan.PostDataSQL = append(plan.PostDataSQL, postSQL...)
+ plan.Warnings = append(plan.Warnings, warnings...)
+ plan.UnsupportedObjects = append(plan.UnsupportedObjects, unsupported...)
+ plan.IndexesToCreate = idxCreate
+ plan.IndexesSkipped = idxSkip
+ return dedupeSchemaMigrationPlan(plan), sourceCols, targetCols, nil
+ default:
+ return dedupeSchemaMigrationPlan(plan), sourceCols, targetCols, nil
+ }
+}
+
+func buildMySQLToMySQLCreateTablePlan(targetType string, config SyncConfig, targetQueryTable string, sourceCols []connection.ColumnDefinition, sourceDB db.Database, sourceSchema, sourceTable string) (string, []string, []string, []string, int, int, error) {
+ columnDefs := make([]string, 0, len(sourceCols)+1)
+ warnings := make([]string, 0)
+ unsupported := make([]string, 0)
+ pkCols := make([]string, 0, 2)
+ for _, col := range sourceCols {
+ def, colWarnings := buildMySQLToMySQLColumnDefinition(col)
+ warnings = append(warnings, colWarnings...)
+ columnDefs = append(columnDefs, fmt.Sprintf("%s %s", quoteIdentByType(targetType, col.Name), def))
+ if strings.EqualFold(col.Key, "PRI") || strings.EqualFold(col.Key, "PK") {
+ pkCols = append(pkCols, quoteIdentByType(targetType, col.Name))
+ }
+ }
+ if len(pkCols) > 0 {
+ columnDefs = append(columnDefs, fmt.Sprintf("PRIMARY KEY (%s)", strings.Join(pkCols, ", ")))
+ }
+ 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
+ }
+ 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
+ }
+ 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
+}
+
+func buildMySQLToMySQLColumnDefinition(col connection.ColumnDefinition) (string, []string) {
+ targetType := sanitizeMySQLColumnType(col.Type)
+ parts := []string{targetType}
+ warnings := make([]string, 0)
+ if strings.EqualFold(strings.TrimSpace(col.Nullable), "NO") {
+ parts = append(parts, "NOT NULL")
+ } else {
+ parts = append(parts, "NULL")
+ }
+ isAutoIncrement := strings.Contains(strings.ToLower(strings.TrimSpace(col.Extra)), "auto_increment")
+ if isAutoIncrement {
+ if canUseMySQLAutoIncrement(targetType) {
+ parts = append(parts, "AUTO_INCREMENT")
+ } else {
+ warnings = append(warnings, fmt.Sprintf("字段 %s 的类型 %s 不适合保留 AUTO_INCREMENT，已跳过", col.Name, targetType))
+ }
+ } else if defaultSQL, ok, warningText := mapMySQLDefaultToMySQL(col, targetType); warningText != "" {
+ warnings = append(warnings, warningText)
+ } else if ok {
+ parts = append(parts, "DEFAULT "+defaultSQL)
+ }
+ extra := strings.ToLower(strings.TrimSpace(col.Extra))
+ if strings.Contains(extra, "on update current_timestamp") {
+ parts = append(parts, "ON UPDATE CURRENT_TIMESTAMP")
+ }
+ if comment := strings.TrimSpace(col.Comment); comment != "" {
+ parts = append(parts, "COMMENT '"+escapeMySQLStringLiteral(comment)+"'")
+ }
+ return strings.Join(parts, " "), dedupeStrings(warnings)
+}
+
+func mapMySQLDefaultToMySQL(col connection.ColumnDefinition, targetType string) (string, bool, string) {
+ if col.Default == nil {
+ return "", false, ""
+ }
+ raw := strings.TrimSpace(*col.Default)
+ if raw == "" {
+ if isMySQLStringLikeTargetType(targetType) {
+ return "''", true, ""
+ }
+ return "", false, fmt.Sprintf("字段 %s 的空字符串默认值未保留", col.Name)
+ }
+ lower := strings.ToLower(raw)
+ if lower == "null" {
+ return "", false, ""
+ }
+ if strings.ContainsAny(raw, ";\n\r") {
+ return "", false, fmt.Sprintf("字段 %s 的默认值包含不安全字符，当前未自动迁移", col.Name)
+ }
+ switch {
+ case strings.HasPrefix(lower, "current_timestamp"):
+ return "CURRENT_TIMESTAMP", true, ""
+ case lower == "current_date":
+ return "CURRENT_DATE", true, ""
+ case lower == "current_time":
+ return "CURRENT_TIME", true, ""
+ }
+ if numericPattern.MatchString(raw) && !isMySQLStringLikeTargetType(targetType) {
+ return raw, true, ""
+ }
+ if strings.ContainsAny(raw, "()") && !strings.HasPrefix(lower, "current_timestamp") {
+ return "", false, fmt.Sprintf("字段 %s 的默认值 %s 包含表达式，当前未自动迁移", col.Name, raw)
+ }
+ return "'" + escapeMySQLStringLiteral(raw) + "'", true, ""
+}
+
+func isMySQLStringLikeTargetType(targetType string) bool {
+ text := strings.ToLower(strings.TrimSpace(targetType))
+ return strings.Contains(text, "char") ||
+ strings.Contains(text, "text") ||
+ strings.Contains(text, "json") ||
+ strings.Contains(text, "blob") ||
+ strings.Contains(text, "binary") ||
+ strings.Contains(text, "enum") ||
+ strings.Contains(text, "set")
+}
+
+func escapeMySQLStringLiteral(value string) string {
+ return strings.ReplaceAll(value, "'", "''")
+}
+
+func buildPGLikeToPGLikePlan(config SyncConfig, tableName string, sourceDB db.Database, targetDB db.Database) (SchemaMigrationPlan, []connection.ColumnDefinition, []connection.ColumnDefinition, error) {
+ plan := SchemaMigrationPlan{}
+ sourceType := resolveMigrationDBType(config.SourceConfig)
+ targetType := resolveMigrationDBType(config.TargetConfig)
+ plan.SourceSchema, plan.SourceTable = normalizeSchemaAndTable(sourceType, config.SourceConfig.Database, tableName)
+ plan.TargetSchema, plan.TargetTable = normalizeSchemaAndTable(targetType, config.TargetConfig.Database, tableName)
+ plan.SourceQueryTable = qualifiedNameForQuery(sourceType, plan.SourceSchema, plan.SourceTable, tableName)
+ plan.TargetQueryTable = qualifiedNameForQuery(targetType, plan.TargetSchema, plan.TargetTable, tableName)
+ plan.PlannedAction = "使用已有目标表导入"
+
+ sourceCols, sourceExists, err := inspectTableColumns(sourceDB, plan.SourceSchema, plan.SourceTable)
+ if err != nil {
+ return plan, nil, nil, fmt.Errorf("获取源表字段失败: %w", err)
+ }
+ if !sourceExists {
+ return plan, nil, nil, fmt.Errorf("源表不存在或无列定义: %s", tableName)
+ }
+
+ targetCols, targetExists, err := inspectTableColumns(targetDB, plan.TargetSchema, plan.TargetTable)
+ if err != nil {
+ return plan, sourceCols, nil, fmt.Errorf("获取目标表字段失败: %w", err)
+ }
+ plan.TargetTableExists = targetExists
+
+ strategy := normalizeTargetTableStrategy(config.TargetTableStrategy)
+ if targetExists {
+ missing := diffMissingColumnNames(sourceCols, targetCols)
+ if len(missing) > 0 {
+ plan.Warnings = append(plan.Warnings, fmt.Sprintf("目标表缺失字段 %d 个：%s", len(missing), strings.Join(missing, ", ")))
+ }
+ if len(missing) == 0 {
+ plan.PlannedAction = "表结构已一致"
+ } else if config.AutoAddColumns {
+ targetSet := make(map[string]struct{}, len(targetCols))
+ for _, col := range targetCols {
+ key := strings.ToLower(strings.TrimSpace(col.Name))
+ if key == "" {
+ continue
+ }
+ targetSet[key] = struct{}{}
+ }
+ for _, col := range sourceCols {
+ key := strings.ToLower(strings.TrimSpace(col.Name))
+ if key == "" {
+ continue
+ }
+ if _, ok := targetSet[key]; ok {
+ continue
+ }
+ addSQL, err := buildAddColumnSQLForPair(sourceType, targetType, plan.TargetQueryTable, col)
+ if err != nil {
+ plan.Warnings = append(plan.Warnings, fmt.Sprintf("字段 %s 自动补齐 SQL 生成失败：%v", col.Name, err))
+ continue
+ }
+ plan.PreDataSQL = append(plan.PreDataSQL, addSQL)
+ }
+ if len(plan.PreDataSQL) > 0 {
+ plan.PlannedAction = fmt.Sprintf("补齐缺失字段(%d)后导入", len(plan.PreDataSQL))
+ } else {
+ plan.PlannedAction = fmt.Sprintf("目标表缺失字段(%d)，但未生成可执行补齐 SQL", len(missing))
+ }
+ } else {
+ plan.PlannedAction = fmt.Sprintf("目标表缺失字段(%d)，未开启自动补齐", len(missing))
+ }
+ if strategy != "existing_only" {
+ plan.Warnings = append(plan.Warnings, "目标表已存在，当前仅执行数据导入；不会自动重建已有索引/约束")
+ }
+ return dedupeSchemaMigrationPlan(plan), sourceCols, targetCols, nil
+ }
+
+ switch strategy {
+ case "existing_only":
+ plan.PlannedAction = "目标表不存在，需先手工创建"
+ plan.Warnings = append(plan.Warnings, "当前策略要求目标表已存在，执行时不会自动建表")
+ return dedupeSchemaMigrationPlan(plan), sourceCols, targetCols, nil
+ case "smart", "auto_create_if_missing":
+ plan.AutoCreate = true
+ plan.PlannedAction = "目标表不存在，将自动建表后导入"
+ createSQL, postSQL, warnings, unsupported, idxCreate, idxSkip, err := buildPGLikeToPGLikeCreateTablePlan(targetType, config, plan.TargetQueryTable, sourceCols, sourceDB, plan.SourceSchema, plan.SourceTable)
+ if err != nil {
+ return plan, sourceCols, targetCols, err
+ }
+ plan.CreateTableSQL = createSQL
+ plan.PostDataSQL = append(plan.PostDataSQL, postSQL...)
+ plan.Warnings = append(plan.Warnings, warnings...)
+ plan.UnsupportedObjects = append(plan.UnsupportedObjects, unsupported...)
+ plan.IndexesToCreate = idxCreate
+ plan.IndexesSkipped = idxSkip
+ return dedupeSchemaMigrationPlan(plan), sourceCols, targetCols, nil
+ default:
+ return dedupeSchemaMigrationPlan(plan), sourceCols, targetCols, nil
+ }
+}
+
+func buildPGLikeToPGLikeCreateTablePlan(targetType string, config SyncConfig, targetQueryTable string, sourceCols []connection.ColumnDefinition, sourceDB db.Database, sourceSchema, sourceTable string) (string, []string, []string, []string, int, int, error) {
+ columnDefs := make([]string, 0, len(sourceCols)+1)
+ warnings := make([]string, 0)
+ unsupported := make([]string, 0)
+ pkCols := make([]string, 0, 2)
+ pkColNames := make([]string, 0, 2)
+ for _, col := range sourceCols {
+ def, colWarnings := buildPGLikeToPGLikeColumnDefinition(col)
+ warnings = append(warnings, colWarnings...)
+ columnDefs = append(columnDefs, fmt.Sprintf("%s %s", quoteIdentByType(targetType, col.Name), def))
+ if strings.EqualFold(col.Key, "PRI") || strings.EqualFold(col.Key, "PK") {
+ pkCols = append(pkCols, quoteIdentByType(targetType, col.Name))
+ pkColNames = append(pkColNames, col.Name)
+ }
+ }
+ if len(pkCols) > 0 {
+ columnDefs = append(columnDefs, fmt.Sprintf("PRIMARY KEY (%s)", strings.Join(pkCols, ", ")))
+ }
+ 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
+ }
+ 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
+ }
+ 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 idx.Unique && sameColumnNameList(idx.Columns, pkColNames) {
+ 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
+}
+
+func buildPGLikeToPGLikeColumnDefinition(col connection.ColumnDefinition) (string, []string) {
+ targetType := sanitizePGLikeColumnType(col.Type)
+ parts := []string{targetType}
+ warnings := make([]string, 0)
+ if strings.Contains(strings.ToLower(strings.TrimSpace(col.Extra)), "auto_increment") {
+ if canUsePGLikeIdentity(targetType) {
+ parts = append(parts, "GENERATED BY DEFAULT AS IDENTITY")
+ } else {
+ warnings = append(warnings, fmt.Sprintf("字段 %s 的类型 %s 不适合保留 identity/sequence 语义，已跳过", col.Name, targetType))
+ }
+ } else if defaultSQL, ok, warningText := mapPGLikeDefaultToPGLike(col, targetType); warningText != "" {
+ warnings = append(warnings, warningText)
+ } else if ok {
+ parts = append(parts, "DEFAULT "+defaultSQL)
+ }
+ if strings.EqualFold(strings.TrimSpace(col.Nullable), "NO") {
+ parts = append(parts, "NOT NULL")
+ }
+ if comment := strings.TrimSpace(col.Comment); comment != "" {
+ warnings = append(warnings, fmt.Sprintf("字段 %s 注释未内联到 CREATE TABLE，请按需使用 COMMENT ON COLUMN 补充", col.Name))
+ }
+ return strings.Join(parts, " "), dedupeStrings(warnings)
+}
+
+func sanitizePGLikeColumnType(t string) string {
+ tt := strings.TrimSpace(t)
+ if tt == "" {
+ return "text"
+ }
+ if strings.ContainsAny(tt, "\";\n\r") {
+ return "text"
+ }
+ return tt
+}
+
+func canUsePGLikeIdentity(targetType string) bool {
+ text := strings.ToLower(strings.TrimSpace(targetType))
+ switch {
+ case strings.HasPrefix(text, "smallint"), strings.HasPrefix(text, "integer"), strings.HasPrefix(text, "int"), strings.HasPrefix(text, "bigint"):
+ return true
+ default:
+ return false
+ }
+}
+
+func mapPGLikeDefaultToPGLike(col connection.ColumnDefinition, targetType string) (string, bool, string) {
+ if col.Default == nil {
+ return "", false, ""
+ }
+ raw := strings.TrimSpace(*col.Default)
+ if raw == "" || strings.EqualFold(raw, "null") {
+ return "", false, ""
+ }
+ lower := strings.ToLower(raw)
+ if strings.HasPrefix(lower, "nextval(") {
+ return "", false, ""
+ }
+ if strings.ContainsAny(raw, ";\n\r") {
+ return "", false, fmt.Sprintf("字段 %s 的默认值包含不安全字符，当前未自动迁移", col.Name)
+ }
+ if strings.Contains(lower, "current_timestamp") || strings.Contains(lower, "now()") {
+ return "CURRENT_TIMESTAMP", true, ""
+ }
+ if lower == "current_date" {
+ return "CURRENT_DATE", true, ""
+ }
+ if lower == "current_time" {
+ return "CURRENT_TIME", true, ""
+ }
+ if targetType == "boolean" {
+ switch lower {
+ case "true", "1":
+ return "TRUE", true, ""
+ case "false", "0":
+ return "FALSE", true, ""
+ }
+ }
+ if numericPattern.MatchString(raw) && !isStringLikeTargetType(targetType) {
+ return raw, true, ""
+ }
+ return raw, true, ""
+}
+
func buildPGLikeToMySQLPlan(config SyncConfig, tableName string, sourceDB db.Database, targetDB db.Database) (SchemaMigrationPlan, []connection.ColumnDefinition, []connection.ColumnDefinition, error) {
plan := SchemaMigrationPlan{}
sourceType := resolveMigrationDBType(config.SourceConfig)
diff --git a/internal/sync/schema_migration_test.go b/internal/sync/schema_migration_test.go
index d016b6c..5c216c4 100644
--- a/internal/sync/schema_migration_test.go
+++ b/internal/sync/schema_migration_test.go
@@ -11,6 +11,7 @@ import (
type fakeMigrationDB struct {
columns map[string][]connection.ColumnDefinition
indexes map[string][]connection.IndexDefinition
+ tables map[string][]string
queryData map[string][]map[string]interface{}
queryCols map[string][]string
}
@@ -27,6 +28,9 @@ func (f *fakeMigrationDB) Query(query string) ([]map[string]interface{}, []strin
func (f *fakeMigrationDB) Exec(query string) (int64, error) { return 0, nil }
func (f *fakeMigrationDB) GetDatabases() ([]string, error) { return nil, nil }
func (f *fakeMigrationDB) GetTables(dbName string) ([]string, error) {
+ if rows, ok := f.tables[dbName]; ok {
+ return rows, nil
+ }
return nil, nil
}
func (f *fakeMigrationDB) GetCreateStatement(dbName, tableName string) (string, error) {
@@ -314,6 +318,204 @@ func TestBuildSchemaMigrationPlan_MySQLToMySQLAddsMissingColumnsForExistingTarge
}
}
+func TestBuildSchemaMigrationPlan_MySQLToMySQLAutoCreatesMissingTarget(t *testing.T) {
+ t.Parallel()
+
+ sourceDB := &fakeMigrationDB{
+ columns: map[string][]connection.ColumnDefinition{
+ "shop.users": {
+ {Name: "id", Type: "bigint", Nullable: "NO", Key: "PRI", Extra: "auto_increment"},
+ {Name: "name", Type: "varchar(128)", Nullable: "YES", Default: stringPtr("")},
+ },
+ },
+ indexes: map[string][]connection.IndexDefinition{
+ "shop.users": {
+ {Name: "idx_users_name", ColumnName: "name", NonUnique: 1, SeqInIndex: 1, IndexType: "BTREE"},
+ },
+ },
+ }
+ targetDB := &fakeMigrationDB{columns: map[string][]connection.ColumnDefinition{}}
+ cfg := SyncConfig{
+ SourceConfig: connection.ConnectionConfig{Type: "mysql", Database: "shop"},
+ TargetConfig: connection.ConnectionConfig{Type: "mysql", Database: "app"},
+ TargetTableStrategy: "smart",
+ CreateIndexes: true,
+ }
+
+ plan, sourceCols, targetCols, err := buildSchemaMigrationPlan(cfg, "users", sourceDB, targetDB)
+ if err != nil {
+ t.Fatalf("buildSchemaMigrationPlan returned error: %v", err)
+ }
+ if len(sourceCols) != 2 || len(targetCols) != 0 {
+ t.Fatalf("unexpected columns lengths: source=%d target=%d", len(sourceCols), len(targetCols))
+ }
+ if plan.TargetTableExists {
+ t.Fatalf("expected target table missing")
+ }
+ if !plan.AutoCreate {
+ t.Fatalf("expected mysql->mysql auto create enabled, plan=%+v", plan)
+ }
+ if strings.Contains(strings.Join(plan.Warnings, " "), "MySQL -> Kingbase") {
+ t.Fatalf("mysql->mysql should not warn with old mysql->kingbase-only message: %v", plan.Warnings)
+ }
+ if !strings.Contains(plan.PlannedAction, "自动建表") {
+ t.Fatalf("unexpected planned action: %s", plan.PlannedAction)
+ }
+ if !strings.Contains(plan.CreateTableSQL, "CREATE TABLE `app`.`users`") {
+ t.Fatalf("unexpected create table SQL: %s", plan.CreateTableSQL)
+ }
+ if !strings.Contains(plan.CreateTableSQL, "`id` bigint NOT NULL AUTO_INCREMENT") {
+ t.Fatalf("unexpected id definition: %s", plan.CreateTableSQL)
+ }
+ if !strings.Contains(plan.CreateTableSQL, "PRIMARY KEY (`id`)") {
+ t.Fatalf("missing primary key: %s", plan.CreateTableSQL)
+ }
+ if len(plan.PostDataSQL) != 1 || !strings.Contains(plan.PostDataSQL[0], "CREATE INDEX `idx_users_name`") {
+ t.Fatalf("unexpected index SQL: %v", plan.PostDataSQL)
+ }
+}
+
+func TestBuildSchemaMigrationPlan_PGLikeToPGLikeAutoCreatesMissingTarget(t *testing.T) {
+ t.Parallel()
+
+ sourceDB := &fakeMigrationDB{
+ columns: map[string][]connection.ColumnDefinition{
+ "public.accounts": {
+ {Name: "id", Type: "integer", Nullable: "NO", Key: "PRI", Extra: "auto_increment"},
+ {Name: "email", Type: "character varying(128)", Nullable: "NO"},
+ {Name: "status", Type: "text", Nullable: "NO", Default: stringPtr("'active'::text")},
+ },
+ },
+ indexes: map[string][]connection.IndexDefinition{
+ "public.accounts": {
+ {Name: "accounts_pkey", ColumnName: "id", NonUnique: 0, SeqInIndex: 1, IndexType: "btree"},
+ {Name: "accounts_email_key", ColumnName: "email", NonUnique: 0, SeqInIndex: 1, IndexType: "btree"},
+ },
+ },
+ }
+ targetDB := &fakeMigrationDB{columns: map[string][]connection.ColumnDefinition{}}
+ cfg := SyncConfig{
+ SourceConfig: connection.ConnectionConfig{Type: "postgres", Database: "public"},
+ TargetConfig: connection.ConnectionConfig{Type: "kingbase", Database: "app"},
+ TargetTableStrategy: "smart",
+ CreateIndexes: true,
+ }
+
+ plan, sourceCols, targetCols, err := buildSchemaMigrationPlan(cfg, "accounts", sourceDB, targetDB)
+ if err != nil {
+ t.Fatalf("buildSchemaMigrationPlan returned error: %v", err)
+ }
+ if len(sourceCols) != 3 || len(targetCols) != 0 {
+ t.Fatalf("unexpected columns lengths: source=%d target=%d", len(sourceCols), len(targetCols))
+ }
+ if !plan.AutoCreate {
+ t.Fatalf("expected pglike->pglike auto create enabled, plan=%+v", plan)
+ }
+ if strings.Contains(strings.Join(plan.Warnings, " "), "MySQL -> Kingbase") {
+ t.Fatalf("pglike->pglike should not warn with old mysql->kingbase-only message: %v", plan.Warnings)
+ }
+ if !strings.Contains(plan.CreateTableSQL, "CREATE TABLE public.accounts") {
+ t.Fatalf("unexpected create table SQL: %s", plan.CreateTableSQL)
+ }
+ if !strings.Contains(plan.CreateTableSQL, "id integer GENERATED BY DEFAULT AS IDENTITY NOT NULL") {
+ t.Fatalf("unexpected id definition: %s", plan.CreateTableSQL)
+ }
+ if !strings.Contains(plan.CreateTableSQL, "status text DEFAULT 'active'::text NOT NULL") {
+ t.Fatalf("unexpected default definition: %s", plan.CreateTableSQL)
+ }
+ if strings.Contains(plan.CreateTableSQL, "'''active''::text'") {
+ t.Fatalf("default expression should not be quoted as a string literal: %s", plan.CreateTableSQL)
+ }
+ if len(plan.PostDataSQL) != 1 || !strings.Contains(plan.PostDataSQL[0], "CREATE UNIQUE INDEX accounts_email_key") {
+ t.Fatalf("unexpected index SQL: %v", plan.PostDataSQL)
+ }
+ if strings.Contains(strings.Join(plan.PostDataSQL, "\n"), "accounts_pkey") {
+ t.Fatalf("primary key index should not be recreated: %v", plan.PostDataSQL)
+ }
+}
+
+func TestBuildSchemaMigrationPlan_ClickHouseToClickHouseAutoCreatesMissingTarget(t *testing.T) {
+ t.Parallel()
+
+ sourceDB := &fakeMigrationDB{
+ columns: map[string][]connection.ColumnDefinition{
+ "analytics.events": {
+ {Name: "ts", Type: "DateTime", Nullable: "NO", Key: "PRI"},
+ {Name: "user_id", Type: "UInt64", Nullable: "NO"},
+ {Name: "payload", Type: "String", Nullable: "YES"},
+ },
+ },
+ }
+ targetDB := &fakeMigrationDB{columns: map[string][]connection.ColumnDefinition{}}
+ cfg := SyncConfig{
+ SourceConfig: connection.ConnectionConfig{Type: "clickhouse", Database: "analytics"},
+ TargetConfig: connection.ConnectionConfig{Type: "clickhouse", Database: "archive"},
+ TargetTableStrategy: "smart",
+ }
+
+ plan, sourceCols, targetCols, err := buildSchemaMigrationPlan(cfg, "events", sourceDB, targetDB)
+ if err != nil {
+ t.Fatalf("buildSchemaMigrationPlan returned error: %v", err)
+ }
+ if len(sourceCols) != 3 || len(targetCols) != 0 {
+ t.Fatalf("unexpected columns lengths: source=%d target=%d", len(sourceCols), len(targetCols))
+ }
+ if !plan.AutoCreate {
+ t.Fatalf("expected clickhouse->clickhouse auto create enabled, plan=%+v", plan)
+ }
+ if !strings.Contains(plan.CreateTableSQL, "CREATE TABLE `archive`.`events`") {
+ t.Fatalf("unexpected create table SQL: %s", plan.CreateTableSQL)
+ }
+ if !strings.Contains(plan.CreateTableSQL, "ENGINE = MergeTree() ORDER BY (`ts`)") {
+ t.Fatalf("unexpected order by: %s", plan.CreateTableSQL)
+ }
+}
+
+func TestBuildSchemaMigrationPlan_MongoToMongoAutoCreatesMissingTarget(t *testing.T) {
+ t.Parallel()
+
+ sourceQuery := `{"find":"users","filter":{},"limit":200}`
+ sourceDB := &fakeMigrationDB{
+ queryData: map[string][]map[string]interface{}{
+ sourceQuery: {
+ {"_id": "u1", "name": "Ada", "age": 37},
+ },
+ },
+ indexes: map[string][]connection.IndexDefinition{
+ "app.users": {
+ {Name: "idx_users_name", ColumnName: "name", NonUnique: 1, SeqInIndex: 1, IndexType: "BTREE"},
+ },
+ },
+ }
+ targetDB := &fakeMigrationDB{tables: map[string][]string{"archive": []string{}}}
+ cfg := SyncConfig{
+ SourceConfig: connection.ConnectionConfig{Type: "mongodb", Database: "app"},
+ TargetConfig: connection.ConnectionConfig{Type: "mongodb", Database: "archive"},
+ TargetTableStrategy: "smart",
+ CreateIndexes: true,
+ }
+
+ plan, sourceCols, targetCols, err := buildSchemaMigrationPlan(cfg, "users", sourceDB, targetDB)
+ if err != nil {
+ t.Fatalf("buildSchemaMigrationPlan returned error: %v", err)
+ }
+ if len(sourceCols) == 0 || len(targetCols) != 0 {
+ t.Fatalf("unexpected columns lengths: source=%d target=%d", len(sourceCols), len(targetCols))
+ }
+ if !plan.AutoCreate {
+ t.Fatalf("expected mongo->mongo auto create enabled, plan=%+v", plan)
+ }
+ if strings.TrimSpace(plan.CreateTableSQL) != "" {
+ t.Fatalf("mongo create should be pre-data command only: %s", plan.CreateTableSQL)
+ }
+ if len(plan.PreDataSQL) != 1 || !strings.Contains(plan.PreDataSQL[0], `"create":"users"`) {
+ t.Fatalf("unexpected create collection command: %v", plan.PreDataSQL)
+ }
+ if len(plan.PostDataSQL) != 1 || !strings.Contains(plan.PostDataSQL[0], `"createIndexes":"users"`) {
+ t.Fatalf("unexpected index command: %v", plan.PostDataSQL)
+ }
+}
+
func TestBuildMySQLToPGLikeCreateTablePlan_GeneratesPostgresDDL(t *testing.T) {
t.Parallel()
diff --git a/internal/sync/sync_engine.go b/internal/sync/sync_engine.go
index e945ec4..7bb4ec2 100644
--- a/internal/sync/sync_engine.go
+++ b/internal/sync/sync_engine.go
@@ -162,15 +162,17 @@ func (s *SyncEngine) RunSync(config SyncConfig) SyncResult {
return
}
}
- if strings.TrimSpace(plan.CreateTableSQL) == "" {
- s.appendLog(config.JobID, &result, "error", fmt.Sprintf("表 %s 自动建表失败：建表 SQL 为空", tableName))
+ if strings.TrimSpace(plan.CreateTableSQL) == "" && len(plan.PreDataSQL) == 0 {
+ s.appendLog(config.JobID, &result, "error", fmt.Sprintf("表 %s 自动建表失败：建表/建集合 SQL 为空", tableName))
return
}
- if _, err := targetDB.Exec(plan.CreateTableSQL); err != nil {
- s.appendLog(config.JobID, &result, "error", fmt.Sprintf("创建目标表失败：表=%s 错误=%v", tableName, err))
- return
+ if strings.TrimSpace(plan.CreateTableSQL) != "" {
+ if _, err := targetDB.Exec(plan.CreateTableSQL); err != nil {
+ s.appendLog(config.JobID, &result, "error", fmt.Sprintf("创建目标表失败：表=%s 错误=%v", tableName, err))
+ return
+ }
}
- s.appendLog(config.JobID, &result, "info", fmt.Sprintf("目标表创建成功：%s", tableName))
+ s.appendLog(config.JobID, &result, "info", fmt.Sprintf("目标对象创建成功：%s", tableName))
targetCols, err = targetDB.GetColumns(plan.TargetSchema, plan.TargetTable)
if err != nil {
s.appendLog(config.JobID, &result, "error", fmt.Sprintf("创建目标表后获取字段失败：表=%s 错误=%v", tableName, err))