✨ feat(sync): 扩展跨库迁移自动建表能力

- 新增 MySQL、PG-like、ClickHouse、MongoDB 同类库迁移规划器
- 支持可映射库对自动建表、补字段及兼容索引迁移
- 修复 MongoDB 创建集合时建表 SQL 为空的执行判断
- 避免 PG-like 主键索引重复迁移并保留默认值表达式
- 更新 Data Sync 自动建表能力提示与回归测试
Refs #465

frontend/src/components/DataSyncModal.tsx

@@ -975,7 +975,7 @@ const DataSyncModal: React.FC<{ open: boolean; onClose: () => void }> = ({ open,
                       )}
                       <Form.Item>
                           <Checkbox checked={autoAddColumns} onChange={(e) => setAutoAddColumns(e.target.checked)} disabled={isSourceQueryMode}>
-                              自动补齐目标表缺失字段（当前支持 MySQL 目标及 MySQL → Kingbase；SQL 结果集模式暂不支持）
+                              自动补齐目标表缺失字段（按源/目标数据源选择可兼容规划器；SQL 结果集模式暂不支持）
                           </Checkbox>
                       </Form.Item>
                       <Form.Item>
@@ -987,7 +987,7 @@ const DataSyncModal: React.FC<{ open: boolean; onClose: () => void }> = ({ open,
                           <Alert
                               type="info"
                               showIcon
-                              message="自动建表模式首期仅支持 MySQL → Kingbase；将迁移字段、主键、普通/唯一/联合索引，并显式跳过全文、空间、前缀、函数类索引。"
+                              message="自动建表模式会按源/目标数据模型选择可用规划器；当前覆盖 MySQL/PG-like/ClickHouse/TDengine/MongoDB 的可映射库对，不兼容索引和表级语义会显式提示或跳过。"
                               style={{ marginBottom: 12 }}
                           />
                       )}

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}

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 {

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()
 

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)

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)
 	}

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

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)

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()
 

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))