✨ feat(sync): 数据同步支持差异对比、行级选择与实时进度日志

- 新增差异分析/预览接口与前端预览抽屉（插入/更新/删除） - 支持按表勾选插入/更新/删除（删除默认不勾选） - 支持按主键选择行级同步；无主键/复合主键表跳过并提示 - 同步过程实时输出中文日志与进度条，便于定位失败步骤
2026-07-10 23:11:40 +08:00 · 2026-02-03 17:37:41 +08:00
parent 80fbfd6365
commit e3bf160072
24 changed files with 2087 additions and 280 deletions
--- a/internal/app/methods_sync.go
+++ b/internal/app/methods_sync.go
@@ -1,11 +1,99 @@
 package app

 import (
+	"fmt"
+	"strings"
+	"time"
+
+	"GoNavi-Wails/internal/connection"
 	"GoNavi-Wails/internal/sync"
+
+	"github.com/wailsapp/wails/v2/pkg/runtime"
 )

 // DataSync executes a data synchronization task
 func (a *App) DataSync(config sync.SyncConfig) sync.SyncResult {
-	engine := sync.NewSyncEngine()
-	return engine.RunSync(config)
+	jobID := strings.TrimSpace(config.JobID)
+	if jobID == "" {
+		jobID = fmt.Sprintf("sync-%d", time.Now().UnixNano())
+		config.JobID = jobID
+	}
+
+	reporter := sync.Reporter{
+		OnLog: func(event sync.SyncLogEvent) {
+			runtime.EventsEmit(a.ctx, sync.EventSyncLog, event)
+		},
+		OnProgress: func(event sync.SyncProgressEvent) {
+			runtime.EventsEmit(a.ctx, sync.EventSyncProgress, event)
+		},
+	}
+
+	runtime.EventsEmit(a.ctx, sync.EventSyncStart, map[string]any{
+		"jobId": jobID,
+		"total": len(config.Tables),
+	})
+
+	engine := sync.NewSyncEngine(reporter)
+	res := engine.RunSync(config)
+
+	runtime.EventsEmit(a.ctx, sync.EventSyncDone, map[string]any{
+		"jobId":  jobID,
+		"result": res,
+	})
+
+	return res
+}
+
+// DataSyncAnalyze analyzes differences between source and target for the given tables (dry-run).
+func (a *App) DataSyncAnalyze(config sync.SyncConfig) connection.QueryResult {
+	jobID := strings.TrimSpace(config.JobID)
+	if jobID == "" {
+		jobID = fmt.Sprintf("analyze-%d", time.Now().UnixNano())
+		config.JobID = jobID
+	}
+
+	reporter := sync.Reporter{
+		OnLog: func(event sync.SyncLogEvent) {
+			runtime.EventsEmit(a.ctx, sync.EventSyncLog, event)
+		},
+		OnProgress: func(event sync.SyncProgressEvent) {
+			runtime.EventsEmit(a.ctx, sync.EventSyncProgress, event)
+		},
+	}
+
+	runtime.EventsEmit(a.ctx, sync.EventSyncStart, map[string]any{
+		"jobId": jobID,
+		"total": len(config.Tables),
+		"type":  "analyze",
+	})
+
+	engine := sync.NewSyncEngine(reporter)
+	res := engine.Analyze(config)
+
+	runtime.EventsEmit(a.ctx, sync.EventSyncDone, map[string]any{
+		"jobId":  jobID,
+		"result": res,
+		"type":   "analyze",
+	})
+
+	if !res.Success {
+		return connection.QueryResult{Success: false, Message: res.Message, Data: res}
+	}
+	return connection.QueryResult{Success: true, Message: res.Message, Data: res}
+}
+
+// DataSyncPreview returns a limited preview of diff rows for one table.
+func (a *App) DataSyncPreview(config sync.SyncConfig, tableName string, limit int) connection.QueryResult {
+	jobID := strings.TrimSpace(config.JobID)
+	if jobID == "" {
+		jobID = fmt.Sprintf("preview-%d", time.Now().UnixNano())
+		config.JobID = jobID
+	}
+
+	engine := sync.NewSyncEngine(sync.Reporter{})
+	preview, err := engine.Preview(config, tableName, limit)
+	if err != nil {
+		return connection.QueryResult{Success: false, Message: err.Error()}
+	}
+	return connection.QueryResult{Success: true, Message: "OK", Data: preview}
 }
--- a/internal/db/custom_impl.go
+++ b/internal/db/custom_impl.go
@@ -88,19 +88,7 @@ func (c *CustomDB) Query(query string) ([]map[string]interface{}, []string, erro

 		entry := make(map[string]interface{})
 		for i, col := range columns {
-			var v interface{}
-			val := values[i]
-			b, ok := val.([]byte)
-			if ok {
-				if b == nil {
-					v = nil
-				} else {
-					v = string(b)
-				}
-			} else {
-				v = val
-			}
-			entry[col] = v
+			entry[col] = normalizeQueryValue(values[i])
 		}
 		resultData = append(resultData, entry)
 	}
--- a/internal/db/dameng_impl.go
+++ b/internal/db/dameng_impl.go
@@ -119,19 +119,7 @@ func (d *DamengDB) Query(query string) ([]map[string]interface{}, []string, erro

 		entry := make(map[string]interface{})
 		for i, col := range columns {
-			var v interface{}
-			val := values[i]
-			b, ok := val.([]byte)
-			if ok {
-				if b == nil {
-					v = nil
-				} else {
-					v = string(b)
-				}
-			} else {
-				v = val
-			}
-			entry[col] = v
+			entry[col] = normalizeQueryValue(values[i])
 		}
 		resultData = append(resultData, entry)
 	}
--- a/internal/db/kingbase_impl.go
+++ b/internal/db/kingbase_impl.go
@@ -150,19 +150,7 @@ func (k *KingbaseDB) Query(query string) ([]map[string]interface{}, []string, er

 		entry := make(map[string]interface{})
 		for i, col := range columns {
-			var v interface{}
-			val := values[i]
-			b, ok := val.([]byte)
-			if ok {
-				if b == nil {
-					v = nil
-				} else {
-					v = string(b)
-				}
-			} else {
-				v = val
-			}
-			entry[col] = v
+			entry[col] = normalizeQueryValue(values[i])
 		}
 		resultData = append(resultData, entry)
 	}
--- a/internal/db/mysql_impl.go
+++ b/internal/db/mysql_impl.go
@@ -48,7 +48,7 @@ func (m *MySQLDB) Connect(config connection.ConnectionConfig) error {
 	}
 	m.conn = db
 	m.pingTimeout = getConnectTimeout(config)
-	
+
 	// Force verification
 	if err := m.Ping(); err != nil {
 		return fmt.Errorf("连接建立后验证失败：%w", err)
@@ -107,19 +107,7 @@ func (m *MySQLDB) Query(query string) ([]map[string]interface{}, []string, error

 		entry := make(map[string]interface{})
 		for i, col := range columns {
-			var v interface{}
-			val := values[i]
-			b, ok := val.([]byte)
-			if ok {
-				if b == nil {
-					v = nil
-				} else {
-					v = string(b)
-				}
-			} else {
-				v = val
-			}
-			entry[col] = v
+			entry[col] = normalizeQueryValue(values[i])
 		}
 		resultData = append(resultData, entry)
 	}
@@ -159,12 +147,12 @@ func (m *MySQLDB) GetTables(dbName string) ([]string, error) {
 	if dbName != "" {
 		query = fmt.Sprintf("SHOW TABLES FROM `%s`", dbName)
 	}
-	
+
 	data, _, err := m.Query(query)
 	if err != nil {
 		return nil, err
 	}
-	
+
 	var tables []string
 	for _, row := range data {
 		for _, v := range row {
@@ -185,7 +173,7 @@ func (m *MySQLDB) GetCreateStatement(dbName, tableName string) (string, error) {
 	if err != nil {
 		return "", err
 	}
-	
+
 	if len(data) > 0 {
 		if val, ok := data[0]["Create Table"]; ok {
 			return fmt.Sprintf("%v", val), nil
@@ -215,12 +203,12 @@ func (m *MySQLDB) GetColumns(dbName, tableName string) ([]connection.ColumnDefin
 			Extra:    fmt.Sprintf("%v", row["Extra"]),
 			Comment:  fmt.Sprintf("%v", row["Comment"]),
 		}
-		
+
 		if row["Default"] != nil {
 			d := fmt.Sprintf("%v", row["Default"])
 			col.Default = &d
 		}
-		
+
 		columns = append(columns, col)
 	}
 	return columns, nil
@@ -248,14 +236,14 @@ func (m *MySQLDB) GetIndexes(dbName, tableName string) ([]connection.IndexDefini
 			}
 		}

-        seq := 0
-        if val, ok := row["Seq_in_index"]; ok {
+		seq := 0
+		if val, ok := row["Seq_in_index"]; ok {
 			if f, ok := val.(float64); ok {
 				seq = int(f)
 			} else if i, ok := val.(int64); ok {
 				seq = int(i)
 			}
-        }
+		}

 		idx := connection.IndexDefinition{
 			Name:       fmt.Sprintf("%v", row["Key_name"]),
@@ -345,12 +333,12 @@ func (m *MySQLDB) ApplyChanges(tableName string, changes connection.ChangeSet) e
 	for _, update := range changes.Updates {
 		var sets []string
 		var args []interface{}
-		
+
 		for k, v := range update.Values {
 			sets = append(sets, fmt.Sprintf("`%s` = ?", k))
 			args = append(args, v)
 		}
-		
+
 		if len(sets) == 0 {
 			continue
 		}
@@ -360,7 +348,7 @@ func (m *MySQLDB) ApplyChanges(tableName string, changes connection.ChangeSet) e
 			wheres = append(wheres, fmt.Sprintf("`%s` = ?", k))
 			args = append(args, v)
 		}
-		
+
 		if len(wheres) == 0 {
 			return fmt.Errorf("update requires keys")
 		}
@@ -376,13 +364,13 @@ func (m *MySQLDB) ApplyChanges(tableName string, changes connection.ChangeSet) e
 		var cols []string
 		var placeholders []string
 		var args []interface{}
-		
+
 		for k, v := range row {
 			cols = append(cols, fmt.Sprintf("`%s`", k))
 			placeholders = append(placeholders, "?")
 			args = append(args, v)
 		}
-		
+
 		if len(cols) == 0 {
 			continue
 		}
--- a/internal/db/oracle_impl.go
+++ b/internal/db/oracle_impl.go
@@ -125,19 +125,7 @@ func (o *OracleDB) Query(query string) ([]map[string]interface{}, []string, erro

 		entry := make(map[string]interface{})
 		for i, col := range columns {
-			var v interface{}
-			val := values[i]
-			b, ok := val.([]byte)
-			if ok {
-				if b == nil {
-					v = nil
-				} else {
-					v = string(b)
-				}
-			} else {
-				v = val
-			}
-			entry[col] = v
+			entry[col] = normalizeQueryValue(values[i])
 		}
 		resultData = append(resultData, entry)
 	}
--- a/internal/db/postgres_impl.go
+++ b/internal/db/postgres_impl.go
@@ -48,7 +48,7 @@ func (p *PostgresDB) Connect(config connection.ConnectionConfig) error {
 	}
 	p.conn = db
 	p.pingTimeout = getConnectTimeout(config)
-		
+
 	// Force verification
 	if err := p.Ping(); err != nil {
 		return fmt.Errorf("连接建立后验证失败：%w", err)
@@ -81,8 +81,7 @@ func (p *PostgresDB) Query(query string) ([]map[string]interface{}, []string, er
 		return nil, nil, fmt.Errorf("connection not open")
 	}

-
-rows, err := p.conn.Query(query)
+	rows, err := p.conn.Query(query)
 	if err != nil {
 		return nil, nil, err
 	}
@@ -108,19 +107,7 @@ rows, err := p.conn.Query(query)

 		entry := make(map[string]interface{})
 		for i, col := range columns {
-			var v interface{}
-			val := values[i]
-			b, ok := val.([]byte)
-			if ok {
-				if b == nil {
-					v = nil
-				} else {
-					v = string(b)
-				}
-			} else {
-				v = val
-			}
-			entry[col] = v
+			entry[col] = normalizeQueryValue(values[i])
 		}
 		resultData = append(resultData, entry)
 	}
@@ -159,7 +146,7 @@ func (p *PostgresDB) GetTables(dbName string) ([]string, error) {
 	if err != nil {
 		return nil, err
 	}
-	
+
 	var tables []string
 	for _, row := range data {
 		schema, okSchema := row["schemaname"]
--- a/internal/db/query_value.go
+++ b/internal/db/query_value.go
@@ -0,0 +1,58 @@
+package db
+
+import (
+	"encoding/hex"
+	"unicode"
+	"unicode/utf8"
+)
+
+// normalizeQueryValue normalizes driver-returned values for UI/JSON transport.
+// 当前主要处理 []byte：如果是可读文本则转为 string，否则转为十六进制字符串，避免前端出现“空白值”。
+func normalizeQueryValue(v interface{}) interface{} {
+	if b, ok := v.([]byte); ok {
+		return bytesToReadableString(b)
+	}
+	return v
+}
+
+func bytesToReadableString(b []byte) interface{} {
+	if b == nil {
+		return nil
+	}
+	if len(b) == 0 {
+		return ""
+	}
+
+	if utf8.Valid(b) {
+		s := string(b)
+		if isMostlyPrintable(s) {
+			return s
+		}
+	}
+
+	return "0x" + hex.EncodeToString(b)
+}
+
+func isMostlyPrintable(s string) bool {
+	if s == "" {
+		return true
+	}
+
+	total := 0
+	printable := 0
+	for _, r := range s {
+		total++
+		switch r {
+		case '\n', '\r', '\t':
+			printable++
+			continue
+		default:
+		}
+		if unicode.IsPrint(r) {
+			printable++
+		}
+	}
+
+	// 允许少量不可见字符，避免把正常文本误判为二进制。
+	return printable*100 >= total*90
+}
--- a/internal/db/sqlite_impl.go
+++ b/internal/db/sqlite_impl.go
@@ -17,14 +17,14 @@ type SQLiteDB struct {
 }

 func (s *SQLiteDB) Connect(config connection.ConnectionConfig) error {
-	dsn := config.Host 
+	dsn := config.Host
 	db, err := sql.Open("sqlite", dsn)
 	if err != nil {
 		return fmt.Errorf("打开数据库连接失败：%w", err)
 	}
 	s.conn = db
 	s.pingTimeout = getConnectTimeout(config)
-	
+
 	// Force verification
 	if err := s.Ping(); err != nil {
 		return fmt.Errorf("连接建立后验证失败：%w", err)
@@ -83,19 +83,7 @@ func (s *SQLiteDB) Query(query string) ([]map[string]interface{}, []string, erro

 		entry := make(map[string]interface{})
 		for i, col := range columns {
-			var v interface{}
-			val := values[i]
-			b, ok := val.([]byte)
-			if ok {
-				if b == nil {
-					v = nil
-				} else {
-					v = string(b)
-				}
-			} else {
-				v = val
-			}
-			entry[col] = v
+			entry[col] = normalizeQueryValue(values[i])
 		}
 		resultData = append(resultData, entry)
 	}
@@ -124,7 +112,7 @@ func (s *SQLiteDB) GetTables(dbName string) ([]string, error) {
 	if err != nil {
 		return nil, err
 	}
-	
+
 	var tables []string
 	for _, row := range data {
 		if val, ok := row["name"]; ok {
--- a/internal/sync/analyze.go
+++ b/internal/sync/analyze.go
@@ -0,0 +1,198 @@
+package sync
+
+import (
+	"GoNavi-Wails/internal/db"
+	"GoNavi-Wails/internal/logger"
+	"fmt"
+	"strings"
+)
+
+type TableDiffSummary struct {
+	Table     string `json:"table"`
+	PKColumn  string `json:"pkColumn,omitempty"`
+	CanSync   bool   `json:"canSync"`
+	Inserts   int    `json:"inserts"`
+	Updates   int    `json:"updates"`
+	Deletes   int    `json:"deletes"`
+	Same      int    `json:"same"`
+	Message   string `json:"message,omitempty"`
+	HasSchema bool   `json:"hasSchema,omitempty"`
+}
+
+type SyncAnalyzeResult struct {
+	Success bool               `json:"success"`
+	Message string             `json:"message"`
+	Tables  []TableDiffSummary `json:"tables"`
+}
+
+func (s *SyncEngine) Analyze(config SyncConfig) SyncAnalyzeResult {
+	result := SyncAnalyzeResult{Success: true, Tables: []TableDiffSummary{}}
+
+	contentRaw := strings.ToLower(strings.TrimSpace(config.Content))
+	syncSchema := false
+	syncData := true
+	switch contentRaw {
+	case "", "data":
+		syncData = true
+	case "schema":
+		syncSchema = true
+		syncData = false
+	case "both":
+		syncSchema = true
+		syncData = true
+	default:
+		s.appendLog(config.JobID, nil, "warn", fmt.Sprintf("未知同步内容 %q，已自动使用仅同步数据", config.Content))
+		syncData = true
+	}
+
+	totalTables := len(config.Tables)
+	s.progress(config.JobID, 0, totalTables, "", "差异分析开始")
+
+	sourceDB, err := db.NewDatabase(config.SourceConfig.Type)
+	if err != nil {
+		logger.Error(err, "初始化源数据库驱动失败：类型=%s", config.SourceConfig.Type)
+		return SyncAnalyzeResult{Success: false, Message: "初始化源数据库驱动失败: " + err.Error()}
+	}
+	targetDB, err := db.NewDatabase(config.TargetConfig.Type)
+	if err != nil {
+		logger.Error(err, "初始化目标数据库驱动失败：类型=%s", config.TargetConfig.Type)
+		return SyncAnalyzeResult{Success: false, Message: "初始化目标数据库驱动失败: " + err.Error()}
+	}
+
+	// Connect Source
+	if err := sourceDB.Connect(config.SourceConfig); err != nil {
+		logger.Error(err, "源数据库连接失败：%s", formatConnSummaryForSync(config.SourceConfig))
+		return SyncAnalyzeResult{Success: false, Message: "源数据库连接失败: " + err.Error()}
+	}
+	defer sourceDB.Close()
+
+	// Connect Target
+	if err := targetDB.Connect(config.TargetConfig); err != nil {
+		logger.Error(err, "目标数据库连接失败：%s", formatConnSummaryForSync(config.TargetConfig))
+		return SyncAnalyzeResult{Success: false, Message: "目标数据库连接失败: " + err.Error()}
+	}
+	defer targetDB.Close()
+
+	for i, tableName := range config.Tables {
+		func() {
+			s.progress(config.JobID, i, totalTables, tableName, fmt.Sprintf("分析表(%d/%d)", i+1, totalTables))
+
+			summary := TableDiffSummary{
+				Table:     tableName,
+				CanSync:   false,
+				Inserts:   0,
+				Updates:   0,
+				Deletes:   0,
+				Same:      0,
+				Message:   "",
+				HasSchema: syncSchema,
+			}
+
+			sourceSchema, sourceTable := normalizeSchemaAndTable(config.SourceConfig.Type, config.SourceConfig.Database, tableName)
+			targetSchema, targetTable := normalizeSchemaAndTable(config.TargetConfig.Type, config.TargetConfig.Database, tableName)
+			sourceQueryTable := qualifiedNameForQuery(config.SourceConfig.Type, sourceSchema, sourceTable, tableName)
+			targetQueryTable := qualifiedNameForQuery(config.TargetConfig.Type, targetSchema, targetTable, tableName)
+
+			cols, err := sourceDB.GetColumns(sourceSchema, sourceTable)
+			if err != nil {
+				summary.Message = "获取源表字段失败: " + err.Error()
+				result.Tables = append(result.Tables, summary)
+				return
+			}
+
+			if !syncData {
+				summary.CanSync = true
+				summary.Message = "仅同步结构，未执行数据差异分析"
+				result.Tables = append(result.Tables, summary)
+				return
+			}
+
+			pkCols := make([]string, 0, 2)
+			for _, c := range cols {
+				if c.Key == "PRI" || c.Key == "PK" {
+					pkCols = append(pkCols, c.Name)
+				}
+			}
+			if len(pkCols) == 0 {
+				summary.Message = "无主键，不支持数据对比/同步"
+				result.Tables = append(result.Tables, summary)
+				return
+			}
+			if len(pkCols) > 1 {
+				summary.Message = fmt.Sprintf("复合主键（%s），暂不支持数据对比/同步", strings.Join(pkCols, ","))
+				result.Tables = append(result.Tables, summary)
+				return
+			}
+			summary.PKColumn = pkCols[0]
+
+			// Query data for diff
+			sourceRows, _, err := sourceDB.Query(fmt.Sprintf("SELECT * FROM %s", quoteQualifiedIdentByType(config.SourceConfig.Type, sourceQueryTable)))
+			if err != nil {
+				summary.Message = "读取源表失败: " + err.Error()
+				result.Tables = append(result.Tables, summary)
+				return
+			}
+			targetRows, _, err := targetDB.Query(fmt.Sprintf("SELECT * FROM %s", quoteQualifiedIdentByType(config.TargetConfig.Type, targetQueryTable)))
+			if err != nil {
+				summary.Message = "读取目标表失败: " + err.Error()
+				result.Tables = append(result.Tables, summary)
+				return
+			}
+
+			pkCol := summary.PKColumn
+			targetMap := make(map[string]map[string]interface{}, len(targetRows))
+			for _, row := range targetRows {
+				if row[pkCol] == nil {
+					continue
+				}
+				pkVal := strings.TrimSpace(fmt.Sprintf("%v", row[pkCol]))
+				if pkVal == "" || pkVal == "<nil>" {
+					continue
+				}
+				targetMap[pkVal] = row
+			}
+
+			sourcePKSet := make(map[string]struct{}, len(sourceRows))
+			for _, sRow := range sourceRows {
+				if sRow[pkCol] == nil {
+					continue
+				}
+				pkVal := strings.TrimSpace(fmt.Sprintf("%v", sRow[pkCol]))
+				if pkVal == "" || pkVal == "<nil>" {
+					continue
+				}
+				sourcePKSet[pkVal] = struct{}{}
+
+				if tRow, exists := targetMap[pkVal]; exists {
+					changed := false
+					for k, v := range sRow {
+						if fmt.Sprintf("%v", v) != fmt.Sprintf("%v", tRow[k]) {
+							changed = true
+							break
+						}
+					}
+					if changed {
+						summary.Updates++
+					} else {
+						summary.Same++
+					}
+				} else {
+					summary.Inserts++
+				}
+			}
+
+			for pkVal := range targetMap {
+				if _, ok := sourcePKSet[pkVal]; !ok {
+					summary.Deletes++
+				}
+			}
+
+			summary.CanSync = true
+			result.Tables = append(result.Tables, summary)
+		}()
+	}
+
+	s.progress(config.JobID, totalTables, totalTables, "", "差异分析完成")
+	result.Message = fmt.Sprintf("已完成 %d 张表的差异分析", len(result.Tables))
+	return result
+}
--- a/internal/sync/preview.go
+++ b/internal/sync/preview.go
@@ -0,0 +1,164 @@
+package sync
+
+import (
+	"GoNavi-Wails/internal/db"
+	"fmt"
+	"strings"
+)
+
+type PreviewRow struct {
+	PK  string                 `json:"pk"`
+	Row map[string]interface{} `json:"row"`
+}
+
+type PreviewUpdateRow struct {
+	PK             string                 `json:"pk"`
+	ChangedColumns []string               `json:"changedColumns"`
+	Source         map[string]interface{} `json:"source"`
+	Target         map[string]interface{} `json:"target"`
+}
+
+type TableDiffPreview struct {
+	Table        string             `json:"table"`
+	PKColumn     string             `json:"pkColumn"`
+	TotalInserts int                `json:"totalInserts"`
+	TotalUpdates int                `json:"totalUpdates"`
+	TotalDeletes int                `json:"totalDeletes"`
+	Inserts      []PreviewRow       `json:"inserts"`
+	Updates      []PreviewUpdateRow `json:"updates"`
+	Deletes      []PreviewRow       `json:"deletes"`
+}
+
+func (s *SyncEngine) Preview(config SyncConfig, tableName string, limit int) (TableDiffPreview, error) {
+	if limit <= 0 {
+		limit = 200
+	}
+	if limit > 500 {
+		limit = 500
+	}
+
+	sourceDB, err := db.NewDatabase(config.SourceConfig.Type)
+	if err != nil {
+		return TableDiffPreview{}, fmt.Errorf("初始化源数据库驱动失败: %w", err)
+	}
+	targetDB, err := db.NewDatabase(config.TargetConfig.Type)
+	if err != nil {
+		return TableDiffPreview{}, fmt.Errorf("初始化目标数据库驱动失败: %w", err)
+	}
+
+	if err := sourceDB.Connect(config.SourceConfig); err != nil {
+		return TableDiffPreview{}, fmt.Errorf("源数据库连接失败: %w", err)
+	}
+	defer sourceDB.Close()
+
+	if err := targetDB.Connect(config.TargetConfig); err != nil {
+		return TableDiffPreview{}, fmt.Errorf("目标数据库连接失败: %w", err)
+	}
+	defer targetDB.Close()
+
+	sourceSchema, sourceTable := normalizeSchemaAndTable(config.SourceConfig.Type, config.SourceConfig.Database, tableName)
+	targetSchema, targetTable := normalizeSchemaAndTable(config.TargetConfig.Type, config.TargetConfig.Database, tableName)
+	sourceQueryTable := qualifiedNameForQuery(config.SourceConfig.Type, sourceSchema, sourceTable, tableName)
+	targetQueryTable := qualifiedNameForQuery(config.TargetConfig.Type, targetSchema, targetTable, tableName)
+
+	cols, err := sourceDB.GetColumns(sourceSchema, sourceTable)
+	if err != nil {
+		return TableDiffPreview{}, fmt.Errorf("获取源表字段失败: %w", err)
+	}
+
+	pkCols := make([]string, 0, 2)
+	for _, c := range cols {
+		if c.Key == "PRI" || c.Key == "PK" {
+			pkCols = append(pkCols, c.Name)
+		}
+	}
+	if len(pkCols) == 0 {
+		return TableDiffPreview{}, fmt.Errorf("无主键，不支持数据预览")
+	}
+	if len(pkCols) > 1 {
+		return TableDiffPreview{}, fmt.Errorf("复合主键（%s），暂不支持数据预览", strings.Join(pkCols, ","))
+	}
+	pkCol := pkCols[0]
+
+	sourceRows, _, err := sourceDB.Query(fmt.Sprintf("SELECT * FROM %s", quoteQualifiedIdentByType(config.SourceConfig.Type, sourceQueryTable)))
+	if err != nil {
+		return TableDiffPreview{}, fmt.Errorf("读取源表失败: %w", err)
+	}
+	targetRows, _, err := targetDB.Query(fmt.Sprintf("SELECT * FROM %s", quoteQualifiedIdentByType(config.TargetConfig.Type, targetQueryTable)))
+	if err != nil {
+		return TableDiffPreview{}, fmt.Errorf("读取目标表失败: %w", err)
+	}
+
+	targetMap := make(map[string]map[string]interface{}, len(targetRows))
+	for _, row := range targetRows {
+		if row[pkCol] == nil {
+			continue
+		}
+		pkVal := strings.TrimSpace(fmt.Sprintf("%v", row[pkCol]))
+		if pkVal == "" || pkVal == "<nil>" {
+			continue
+		}
+		targetMap[pkVal] = row
+	}
+
+	out := TableDiffPreview{
+		Table:        tableName,
+		PKColumn:     pkCol,
+		TotalInserts: 0,
+		TotalUpdates: 0,
+		TotalDeletes: 0,
+		Inserts:      make([]PreviewRow, 0),
+		Updates:      make([]PreviewUpdateRow, 0),
+		Deletes:      make([]PreviewRow, 0),
+	}
+
+	sourcePKSet := make(map[string]struct{}, len(sourceRows))
+	for _, sRow := range sourceRows {
+		if sRow[pkCol] == nil {
+			continue
+		}
+		pkVal := strings.TrimSpace(fmt.Sprintf("%v", sRow[pkCol]))
+		if pkVal == "" || pkVal == "<nil>" {
+			continue
+		}
+		sourcePKSet[pkVal] = struct{}{}
+
+		if tRow, exists := targetMap[pkVal]; exists {
+			changedColumns := make([]string, 0)
+			for k, v := range sRow {
+				if fmt.Sprintf("%v", v) != fmt.Sprintf("%v", tRow[k]) {
+					changedColumns = append(changedColumns, k)
+				}
+			}
+			if len(changedColumns) > 0 {
+				out.TotalUpdates++
+				if len(out.Updates) < limit {
+					out.Updates = append(out.Updates, PreviewUpdateRow{
+						PK:             pkVal,
+						ChangedColumns: changedColumns,
+						Source:         sRow,
+						Target:         tRow,
+					})
+				}
+			}
+			continue
+		}
+
+		out.TotalInserts++
+		if len(out.Inserts) < limit {
+			out.Inserts = append(out.Inserts, PreviewRow{PK: pkVal, Row: sRow})
+		}
+	}
+
+	for pkVal, row := range targetMap {
+		if _, ok := sourcePKSet[pkVal]; ok {
+			continue
+		}
+		out.TotalDeletes++
+		if len(out.Deletes) < limit {
+			out.Deletes = append(out.Deletes, PreviewRow{PK: pkVal, Row: row})
+		}
+	}
+
+	return out, nil
+}
--- a/internal/sync/row_selection.go
+++ b/internal/sync/row_selection.go
@@ -0,0 +1,58 @@
+package sync
+
+import (
+	"GoNavi-Wails/internal/connection"
+	"fmt"
+)
+
+func filterRowsByPKSelection(pkCol string, rows []map[string]interface{}, enabled bool, selectedPKs []string) []map[string]interface{} {
+	if !enabled {
+		return nil
+	}
+	if len(rows) == 0 {
+		return rows
+	}
+	if len(selectedPKs) == 0 {
+		return rows
+	}
+
+	set := make(map[string]struct{}, len(selectedPKs))
+	for _, pk := range selectedPKs {
+		set[pk] = struct{}{}
+	}
+
+	out := make([]map[string]interface{}, 0, len(rows))
+	for _, row := range rows {
+		pkStr := fmt.Sprintf("%v", row[pkCol])
+		if _, ok := set[pkStr]; ok {
+			out = append(out, row)
+		}
+	}
+	return out
+}
+
+func filterUpdatesByPKSelection(pkCol string, updates []connection.UpdateRow, enabled bool, selectedPKs []string) []connection.UpdateRow {
+	if !enabled {
+		return nil
+	}
+	if len(updates) == 0 {
+		return updates
+	}
+	if len(selectedPKs) == 0 {
+		return updates
+	}
+
+	set := make(map[string]struct{}, len(selectedPKs))
+	for _, pk := range selectedPKs {
+		set[pk] = struct{}{}
+	}
+
+	out := make([]connection.UpdateRow, 0, len(updates))
+	for _, u := range updates {
+		pkStr := fmt.Sprintf("%v", u.Keys[pkCol])
+		if _, ok := set[pkStr]; ok {
+			out = append(out, u)
+		}
+	}
+	return out
+}
--- a/internal/sync/schema_align.go
+++ b/internal/sync/schema_align.go
@@ -0,0 +1,97 @@
+package sync
+
+import (
+	"GoNavi-Wails/internal/connection"
+	"strings"
+)
+
+func collectRequiredColumns(inserts []map[string]interface{}, updates []connection.UpdateRow) map[string]string {
+	// key: lower(columnName), value: original columnName
+	required := make(map[string]string)
+	for _, row := range inserts {
+		for k := range row {
+			key := strings.ToLower(strings.TrimSpace(k))
+			if key == "" {
+				continue
+			}
+			if _, exists := required[key]; !exists {
+				required[key] = k
+			}
+		}
+	}
+	for _, u := range updates {
+		for k := range u.Values {
+			key := strings.ToLower(strings.TrimSpace(k))
+			if key == "" {
+				continue
+			}
+			if _, exists := required[key]; !exists {
+				required[key] = k
+			}
+		}
+	}
+	return required
+}
+
+func filterInsertRows(inserts []map[string]interface{}, allowedLower map[string]struct{}) []map[string]interface{} {
+	if len(inserts) == 0 || len(allowedLower) == 0 {
+		return inserts
+	}
+
+	out := make([]map[string]interface{}, 0, len(inserts))
+	for _, row := range inserts {
+		if len(row) == 0 {
+			out = append(out, row)
+			continue
+		}
+		n := make(map[string]interface{}, len(row))
+		for k, v := range row {
+			if _, ok := allowedLower[strings.ToLower(strings.TrimSpace(k))]; ok {
+				n[k] = v
+			}
+		}
+		out = append(out, n)
+	}
+	return out
+}
+
+func filterUpdateRows(updates []connection.UpdateRow, allowedLower map[string]struct{}) []connection.UpdateRow {
+	if len(updates) == 0 || len(allowedLower) == 0 {
+		return updates
+	}
+
+	out := make([]connection.UpdateRow, 0, len(updates))
+	for _, u := range updates {
+		if len(u.Values) == 0 {
+			continue
+		}
+
+		values := make(map[string]interface{}, len(u.Values))
+		for k, v := range u.Values {
+			if _, ok := allowedLower[strings.ToLower(strings.TrimSpace(k))]; ok {
+				values[k] = v
+			}
+		}
+		if len(values) == 0 {
+			continue
+		}
+		out = append(out, connection.UpdateRow{
+			Keys:   u.Keys,
+			Values: values,
+		})
+	}
+	return out
+}
+
+func sanitizeMySQLColumnType(t string) string {
+	tt := strings.TrimSpace(t)
+	if tt == "" {
+		return "TEXT"
+	}
+
+	// 基础防护：避免把元数据中异常内容拼进 SQL。
+	if strings.ContainsAny(tt, "`;\n\r") {
+		return "TEXT"
+	}
+	return tt
+}
--- a/internal/sync/schema_sync.go
+++ b/internal/sync/schema_sync.go
@@ -0,0 +1,101 @@
+package sync
+
+import (
+	"GoNavi-Wails/internal/db"
+	"fmt"
+	"strings"
+)
+
+func (s *SyncEngine) syncTableSchema(config SyncConfig, res *SyncResult, sourceDB db.Database, targetDB db.Database, tableName string) error {
+	targetType := strings.ToLower(strings.TrimSpace(config.TargetConfig.Type))
+	if targetType != "mysql" {
+		s.appendLog(config.JobID, res, "warn", fmt.Sprintf("目标数据库类型=%s 暂不支持结构同步，已跳过表 %s", config.TargetConfig.Type, tableName))
+		return nil
+	}
+
+	sourceSchema, sourceTable := normalizeSchemaAndTable(config.SourceConfig.Type, config.SourceConfig.Database, tableName)
+	targetSchema, targetTable := normalizeSchemaAndTable(config.TargetConfig.Type, config.TargetConfig.Database, tableName)
+	targetQueryTable := qualifiedNameForQuery(config.TargetConfig.Type, targetSchema, targetTable, tableName)
+
+	// 1) 获取源表字段
+	sourceCols, err := sourceDB.GetColumns(sourceSchema, sourceTable)
+	if err != nil {
+		return fmt.Errorf("获取源表字段失败: %w", err)
+	}
+
+	// 2) 确保目标表存在
+	targetCols, err := targetDB.GetColumns(targetSchema, targetTable)
+	if err != nil {
+		sourceType := strings.ToLower(strings.TrimSpace(config.SourceConfig.Type))
+		if sourceType != "mysql" {
+			return fmt.Errorf("目标表不存在且源类型=%s 暂不支持自动建表: %w", config.SourceConfig.Type, err)
+		}
+
+		s.appendLog(config.JobID, res, "warn", fmt.Sprintf("目标表 %s 不存在，开始尝试创建表结构", tableName))
+		createSQL, errCreate := sourceDB.GetCreateStatement(sourceSchema, sourceTable)
+		if errCreate != nil || strings.TrimSpace(createSQL) == "" {
+			if errCreate == nil {
+				errCreate = fmt.Errorf("建表语句为空")
+			}
+			return fmt.Errorf("获取源表建表语句失败: %w", errCreate)
+		}
+
+		if _, errExec := targetDB.Exec(createSQL); errExec != nil {
+			return fmt.Errorf("创建目标表失败: %w", errExec)
+		}
+		s.appendLog(config.JobID, res, "info", fmt.Sprintf("目标表创建成功：%s", tableName))
+
+		targetCols, err = targetDB.GetColumns(targetSchema, targetTable)
+		if err != nil {
+			return fmt.Errorf("创建目标表后获取字段失败: %w", err)
+		}
+	}
+
+	targetColSet := make(map[string]struct{}, len(targetCols))
+	for _, c := range targetCols {
+		name := strings.ToLower(strings.TrimSpace(c.Name))
+		if name == "" {
+			continue
+		}
+		targetColSet[name] = struct{}{}
+	}
+
+	// 3) 补齐目标缺失字段（安全策略：新增字段统一允许 NULL）
+	missing := make([]string, 0)
+	sourceType := strings.ToLower(strings.TrimSpace(config.SourceConfig.Type))
+	for _, c := range sourceCols {
+		colName := strings.TrimSpace(c.Name)
+		if colName == "" {
+			continue
+		}
+		lower := strings.ToLower(colName)
+		if _, ok := targetColSet[lower]; ok {
+			continue
+		}
+		missing = append(missing, colName)
+
+		colType := "TEXT"
+		if sourceType == "mysql" {
+			colType = sanitizeMySQLColumnType(c.Type)
+		}
+
+		alterSQL := fmt.Sprintf("ALTER TABLE %s ADD COLUMN %s %s NULL",
+			quoteQualifiedIdentByType("mysql", targetQueryTable),
+			quoteIdentByType("mysql", colName),
+			colType,
+		)
+		if _, err := targetDB.Exec(alterSQL); err != nil {
+			s.appendLog(config.JobID, res, "error", fmt.Sprintf("  -> 补字段失败：表=%s 字段=%s 错误=%v", tableName, colName, err))
+			continue
+		}
+		s.appendLog(config.JobID, res, "info", fmt.Sprintf("  -> 已补齐字段：表=%s 字段=%s 类型=%s", tableName, colName, colType))
+	}
+
+	if len(missing) == 0 {
+		s.appendLog(config.JobID, res, "info", fmt.Sprintf("表结构一致：%s", tableName))
+	} else {
+		s.appendLog(config.JobID, res, "info", fmt.Sprintf("表结构同步完成：%s（新增字段 %d 个）", tableName, len(missing)))
+	}
+
+	return nil
+}
--- a/internal/sync/sql_helpers.go
+++ b/internal/sync/sql_helpers.go
@@ -0,0 +1,109 @@
+package sync
+
+import "strings"
+
+func normalizeSyncMode(mode string) string {
+	m := strings.ToLower(strings.TrimSpace(mode))
+	switch m {
+	case "", "insert_update":
+		return "insert_update"
+	case "insert_only":
+		return "insert_only"
+	case "full_overwrite":
+		return "full_overwrite"
+	default:
+		return "insert_update"
+	}
+}
+
+func quoteIdentByType(dbType string, ident string) string {
+	if ident == "" {
+		return ident
+	}
+
+	switch dbType {
+	case "mysql":
+		return "`" + strings.ReplaceAll(ident, "`", "``") + "`"
+	default:
+		return `"` + strings.ReplaceAll(ident, `"`, `""`) + `"`
+	}
+}
+
+func quoteQualifiedIdentByType(dbType string, ident string) string {
+	raw := strings.TrimSpace(ident)
+	if raw == "" {
+		return raw
+	}
+
+	parts := strings.Split(raw, ".")
+	if len(parts) <= 1 {
+		return quoteIdentByType(dbType, raw)
+	}
+
+	quotedParts := make([]string, 0, len(parts))
+	for _, part := range parts {
+		part = strings.TrimSpace(part)
+		if part == "" {
+			continue
+		}
+		quotedParts = append(quotedParts, quoteIdentByType(dbType, part))
+	}
+
+	if len(quotedParts) == 0 {
+		return quoteIdentByType(dbType, raw)
+	}
+	return strings.Join(quotedParts, ".")
+}
+
+func normalizeSchemaAndTable(dbType string, dbName string, tableName string) (string, string) {
+	rawTable := strings.TrimSpace(tableName)
+	rawDB := strings.TrimSpace(dbName)
+	if rawTable == "" {
+		return rawDB, rawTable
+	}
+
+	if parts := strings.SplitN(rawTable, ".", 2); len(parts) == 2 {
+		schema := strings.TrimSpace(parts[0])
+		table := strings.TrimSpace(parts[1])
+		if schema != "" && table != "" {
+			return schema, table
+		}
+	}
+
+	switch strings.ToLower(strings.TrimSpace(dbType)) {
+	case "postgres", "kingbase":
+		return "public", rawTable
+	default:
+		return rawDB, rawTable
+	}
+}
+
+func qualifiedNameForQuery(dbType string, schema string, table string, original string) string {
+	raw := strings.TrimSpace(original)
+	if raw == "" {
+		return raw
+	}
+	if strings.Contains(raw, ".") {
+		return raw
+	}
+
+	switch strings.ToLower(strings.TrimSpace(dbType)) {
+	case "postgres", "kingbase":
+		s := strings.TrimSpace(schema)
+		if s == "" {
+			s = "public"
+		}
+		if table == "" {
+			return raw
+		}
+		return s + "." + table
+	case "mysql":
+		s := strings.TrimSpace(schema)
+		if s == "" || table == "" {
+			return table
+		}
+		return s + "." + table
+	default:
+		return table
+	}
+}
--- a/internal/sync/sync_engine.go
+++ b/internal/sync/sync_engine.go
@@ -5,15 +5,21 @@ import (
 	"GoNavi-Wails/internal/db"
 	"GoNavi-Wails/internal/logger"
 	"fmt"
+	"sort"
 	"strings"
+	"time"
 )

 // SyncConfig defines the parameters for a synchronization task
 type SyncConfig struct {
-	SourceConfig connection.ConnectionConfig `json:"sourceConfig"`
-	TargetConfig connection.ConnectionConfig `json:"targetConfig"`
-	Tables       []string                    `json:"tables"` // Tables to sync
-	Mode         string                      `json:"mode"`   // "insert_update", "full_overwrite"
+	SourceConfig   connection.ConnectionConfig `json:"sourceConfig"`
+	TargetConfig   connection.ConnectionConfig `json:"targetConfig"`
+	Tables         []string                    `json:"tables"`            // Tables to sync
+	Content        string                      `json:"content,omitempty"` // "data", "schema", "both"
+	Mode           string                      `json:"mode"`              // "insert_update", "insert_only", "full_overwrite"
+	JobID          string                      `json:"jobId,omitempty"`
+	AutoAddColumns bool                        `json:"autoAddColumns,omitempty"` // 自动补齐缺失字段（当前仅 MySQL 目标支持）
+	TableOptions   map[string]TableOptions     `json:"tableOptions,omitempty"`
 }

 // SyncResult holds the result of the sync operation
@@ -28,21 +34,55 @@ type SyncResult struct {
 }

 type SyncEngine struct {
+	reporter Reporter
 }

-func NewSyncEngine() *SyncEngine {
-	return &SyncEngine{}
+func NewSyncEngine(reporter Reporter) *SyncEngine {
+	return &SyncEngine{reporter: reporter}
 }

 // CompareAndSync performs the synchronization
 func (s *SyncEngine) RunSync(config SyncConfig) SyncResult {
 	result := SyncResult{Success: true, Logs: []string{}}
 	logger.Infof("开始数据同步：源=%s 目标=%s 表数量=%d", formatConnSummaryForSync(config.SourceConfig), formatConnSummaryForSync(config.TargetConfig), len(config.Tables))
+	totalTables := len(config.Tables)
+	s.progress(config.JobID, 0, totalTables, "", "开始同步")
+
+	contentRaw := strings.ToLower(strings.TrimSpace(config.Content))
+	syncSchema := false
+	syncData := true
+	switch contentRaw {
+	case "", "data":
+		syncData = true
+	case "schema":
+		syncSchema = true
+		syncData = false
+	case "both":
+		syncSchema = true
+		syncData = true
+	default:
+		s.appendLog(config.JobID, &result, "warn", fmt.Sprintf("未知同步内容 %q，已自动使用仅同步数据", config.Content))
+		syncData = true
+	}
+
+	modeRaw := strings.ToLower(strings.TrimSpace(config.Mode))
+	if modeRaw != "" && modeRaw != "insert_update" && modeRaw != "insert_only" && modeRaw != "full_overwrite" {
+		s.appendLog(config.JobID, &result, "warn", fmt.Sprintf("未知同步模式 %q，已自动使用 insert_update", config.Mode))
+	}
+	defaultMode := normalizeSyncMode(config.Mode)
+
+	contentLabel := "仅同步数据"
+	if syncSchema && syncData {
+		contentLabel = "同步结构+数据"
+	} else if syncSchema {
+		contentLabel = "仅同步结构"
+	}
+	s.appendLog(config.JobID, &result, "info", fmt.Sprintf("同步内容：%s；模式：%s；自动补字段：%v", contentLabel, defaultMode, config.AutoAddColumns))

 	sourceDB, err := db.NewDatabase(config.SourceConfig.Type)
 	if err != nil {
 		logger.Error(err, "初始化源数据库驱动失败：类型=%s", config.SourceConfig.Type)
-		return s.fail(result, "初始化源数据库驱动失败: "+err.Error())
+		return s.fail(config.JobID, totalTables, result, "初始化源数据库驱动失败: "+err.Error())
 	}
 	if config.SourceConfig.Type == "custom" {
 		// Custom DB setup would go here if needed
@@ -51,133 +91,402 @@ func (s *SyncEngine) RunSync(config SyncConfig) SyncResult {
 	targetDB, err := db.NewDatabase(config.TargetConfig.Type)
 	if err != nil {
 		logger.Error(err, "初始化目标数据库驱动失败：类型=%s", config.TargetConfig.Type)
-		return s.fail(result, "初始化目标数据库驱动失败: "+err.Error())
+		return s.fail(config.JobID, totalTables, result, "初始化目标数据库驱动失败: "+err.Error())
 	}

 	// Connect Source
-	result.Logs = append(result.Logs, fmt.Sprintf("正在连接源数据库: %s...", config.SourceConfig.Host))
+	s.appendLog(config.JobID, &result, "info", fmt.Sprintf("正在连接源数据库: %s...", config.SourceConfig.Host))
+	s.progress(config.JobID, 0, totalTables, "", "连接源数据库")
 	if err := sourceDB.Connect(config.SourceConfig); err != nil {
 		logger.Error(err, "源数据库连接失败：%s", formatConnSummaryForSync(config.SourceConfig))
-		return s.fail(result, "源数据库连接失败: "+err.Error())
+		return s.fail(config.JobID, totalTables, result, "源数据库连接失败: "+err.Error())
 	}
 	defer sourceDB.Close()

 	// Connect Target
-	result.Logs = append(result.Logs, fmt.Sprintf("正在连接目标数据库: %s...", config.TargetConfig.Host))
+	s.appendLog(config.JobID, &result, "info", fmt.Sprintf("正在连接目标数据库: %s...", config.TargetConfig.Host))
+	s.progress(config.JobID, 0, totalTables, "", "连接目标数据库")
 	if err := targetDB.Connect(config.TargetConfig); err != nil {
 		logger.Error(err, "目标数据库连接失败：%s", formatConnSummaryForSync(config.TargetConfig))
-		return s.fail(result, "目标数据库连接失败: "+err.Error())
+		return s.fail(config.JobID, totalTables, result, "目标数据库连接失败: "+err.Error())
 	}
 	defer targetDB.Close()

 	// Iterate Tables
-	for _, tableName := range config.Tables {
-		result.Logs = append(result.Logs, fmt.Sprintf("正在同步表: %s", tableName))
+	for i, tableName := range config.Tables {
+		func() {
+			tableMode := defaultMode
+			s.appendLog(config.JobID, &result, "info", fmt.Sprintf("正在同步表: %s", tableName))
+			s.progress(config.JobID, i, totalTables, tableName, fmt.Sprintf("同步表(%d/%d)", i+1, totalTables))
+			defer s.progress(config.JobID, i+1, totalTables, tableName, "表处理完成")

-		// 1. Get Columns & PKs (Naive approach: assume same schema)
-		cols, err := sourceDB.GetColumns(config.SourceConfig.Database, tableName)
-		if err != nil {
-			logger.Error(err, "获取源表列信息失败：表=%s", tableName)
-			result.Logs = append(result.Logs, fmt.Sprintf("获取表 %s 的列信息失败: %v", tableName, err))
-			continue
-		}
-
-		pkCol := ""
-		for _, col := range cols {
-			if col.Key == "PRI" || col.Key == "PK" {
-				pkCol = col.Name
-				break
+			if syncSchema {
+				s.progress(config.JobID, i, totalTables, tableName, "同步表结构")
+				if err := s.syncTableSchema(config, &result, sourceDB, targetDB, tableName); err != nil {
+					s.appendLog(config.JobID, &result, "error", fmt.Sprintf("表结构同步失败：表=%s 错误=%v", tableName, err))
+					return
+				}
+			}
+			if !syncData {
+				result.TablesSynced++
+				return
 			}
-		}

-		if pkCol == "" {
-			result.Logs = append(result.Logs, fmt.Sprintf("跳过表 %s: 未找到主键 (同步需要主键)", tableName))
-			continue
-		}
+			sourceSchema, sourceTable := normalizeSchemaAndTable(config.SourceConfig.Type, config.SourceConfig.Database, tableName)
+			targetSchema, targetTable := normalizeSchemaAndTable(config.TargetConfig.Type, config.TargetConfig.Database, tableName)
+			sourceQueryTable := qualifiedNameForQuery(config.SourceConfig.Type, sourceSchema, sourceTable, tableName)
+			targetQueryTable := qualifiedNameForQuery(config.TargetConfig.Type, targetSchema, targetTable, tableName)

-		// 2. Fetch Data (MEMORY INTENSIVE - PROTOTYPE ONLY)
-		// TODO: Implement paging/streaming
-		sourceRows, _, err := sourceDB.Query(fmt.Sprintf("SELECT * FROM %s", tableName))
-		if err != nil {
-			logger.Error(err, "读取源表失败：表=%s", tableName)
-			result.Logs = append(result.Logs, fmt.Sprintf("读取源表 %s 失败: %v", tableName, err))
-			continue
-		}
+			// 1. Get Columns & PKs
+			cols, err := sourceDB.GetColumns(sourceSchema, sourceTable)
+			if err != nil {
+				logger.Error(err, "获取源表列信息失败：表=%s", tableName)
+				s.appendLog(config.JobID, &result, "error", fmt.Sprintf("获取表 %s 的列信息失败: %v", tableName, err))
+				return
+			}
+			sourceColsByLower := make(map[string]connection.ColumnDefinition, len(cols))
+			for _, col := range cols {
+				if strings.TrimSpace(col.Name) == "" {
+					continue
+				}
+				sourceColsByLower[strings.ToLower(strings.TrimSpace(col.Name))] = col
+			}

-		targetRows, _, err := targetDB.Query(fmt.Sprintf("SELECT * FROM %s", tableName))
-		if err != nil {
-			logger.Error(err, "读取目标表失败：表=%s", tableName)
-			// Table might not exist in target?
-			// Check if error is "table not found" -> Try to Create?
-			// For now, assume table exists.
-			result.Logs = append(result.Logs, fmt.Sprintf("读取目标表 %s 失败: %v", tableName, err))
-			continue
-		}
+			pkCols := make([]string, 0, 2)
+			for _, col := range cols {
+				if col.Key == "PRI" || col.Key == "PK" {
+					pkCols = append(pkCols, col.Name)
+				}
+			}

-		// 3. Compare (In-Memory Hash Map)
-		targetMap := make(map[string]map[string]interface{})
-		for _, row := range targetRows {
-			pkVal := fmt.Sprintf("%v", row[pkCol])
-			targetMap[pkVal] = row
-		}
+			if len(pkCols) == 0 {
+				s.appendLog(config.JobID, &result, "warn", fmt.Sprintf("表 %s 未找到主键，已跳过数据同步（避免产生重复数据）", tableName))
+				return
+			}
+			if len(pkCols) > 1 {
+				s.appendLog(config.JobID, &result, "warn", fmt.Sprintf("表 %s 为复合主键（%s），当前暂不支持数据同步", tableName, strings.Join(pkCols, ",")))
+				return
+			}
+			pkCol := pkCols[0]

-		var inserts []map[string]interface{}
-		var updates []connection.UpdateRow
-		// var deletes []map[string]interface{} // Not implemented in "insert_update" mode usually
-
-		for _, sRow := range sourceRows {
-			pkVal := fmt.Sprintf("%v", sRow[pkCol])
-
-			if tRow, exists := targetMap[pkVal]; exists {
-				// Update? Compare values
-				// Simplified: Compare string representations or iterate keys
-				// For prototype: assume update if exists
-				// Optimization: Check diff
-				changes := make(map[string]interface{})
-				for k, v := range sRow {
-					if fmt.Sprintf("%v", v) != fmt.Sprintf("%v", tRow[k]) {
-						changes[k] = v
+			opts := TableOptions{Insert: true, Update: true, Delete: false}
+			if config.TableOptions != nil {
+				if t, ok := config.TableOptions[tableName]; ok {
+					opts = t
+					// 默认防护：如用户未设置任意一个字段，保持 insert/update 默认 true、delete 默认 false
+					if !t.Insert && !t.Update && !t.Delete {
+						opts = t
 					}
 				}
-				if len(changes) > 0 {
-					updates = append(updates, connection.UpdateRow{
-						Keys:   map[string]interface{}{pkCol: pkVal},
-						Values: changes,
-					})
-				}
-			} else {
-				// Insert
-				inserts = append(inserts, sRow)
 			}
-		}
+			if !opts.Insert && !opts.Update && !opts.Delete {
+				s.appendLog(config.JobID, &result, "info", fmt.Sprintf("表 %s 未勾选任何操作，已跳过", tableName))
+				return
+			}

-		// 4. Apply Changes
-		changeSet := connection.ChangeSet{
-			Inserts: inserts,
-			Updates: updates,
-		}
+			// 2. Fetch Data (MEMORY INTENSIVE - PROTOTYPE ONLY)
+			// TODO: Implement paging/streaming
+			s.progress(config.JobID, i, totalTables, tableName, "读取源表数据")
+			sourceRows, _, err := sourceDB.Query(fmt.Sprintf("SELECT * FROM %s", quoteQualifiedIdentByType(config.SourceConfig.Type, sourceQueryTable)))
+			if err != nil {
+				logger.Error(err, "读取源表失败：表=%s", tableName)
+				s.appendLog(config.JobID, &result, "error", fmt.Sprintf("读取源表 %s 失败: %v", tableName, err))
+				return
+			}

-		if len(inserts) > 0 || len(updates) > 0 {
-			result.Logs = append(result.Logs, fmt.Sprintf("  -> 需插入: %d 行, 需更新: %d 行", len(inserts), len(updates)))
+			var inserts []map[string]interface{}
+			var updates []connection.UpdateRow

-			// We need a BatchApplier interface or assume Database implements ApplyChanges
-			if applier, ok := targetDB.(db.BatchApplier); ok {
-				if err := applier.ApplyChanges(tableName, changeSet); err != nil {
-					result.Logs = append(result.Logs, fmt.Sprintf("  -> 应用变更失败: %v", err))
+			if tableMode == "insert_update" {
+				s.progress(config.JobID, i, totalTables, tableName, "读取目标表数据")
+				targetRows, _, err := targetDB.Query(fmt.Sprintf("SELECT * FROM %s", quoteQualifiedIdentByType(config.TargetConfig.Type, targetQueryTable)))
+				if err != nil {
+					logger.Error(err, "读取目标表失败：表=%s", tableName)
+					s.appendLog(config.JobID, &result, "error", fmt.Sprintf("读取目标表 %s 失败: %v", tableName, err))
+					return
+				}
+
+				// 3. Compare (In-Memory Hash Map)
+				s.progress(config.JobID, i, totalTables, tableName, "对比差异")
+				targetMap := make(map[string]map[string]interface{})
+				for _, row := range targetRows {
+					if row[pkCol] == nil {
+						continue
+					}
+					pkVal := fmt.Sprintf("%v", row[pkCol])
+					if strings.TrimSpace(pkVal) == "" || pkVal == "<nil>" {
+						continue
+					}
+					targetMap[pkVal] = row
+				}
+				sourcePKSet := make(map[string]struct{}, len(sourceRows))
+
+				for _, sRow := range sourceRows {
+					if sRow[pkCol] == nil {
+						continue
+					}
+					pkVal := fmt.Sprintf("%v", sRow[pkCol])
+					if strings.TrimSpace(pkVal) == "" || pkVal == "<nil>" {
+						continue
+					}
+					sourcePKSet[pkVal] = struct{}{}
+
+					if tRow, exists := targetMap[pkVal]; exists {
+						changes := make(map[string]interface{})
+						for k, v := range sRow {
+							if fmt.Sprintf("%v", v) != fmt.Sprintf("%v", tRow[k]) {
+								changes[k] = v
+							}
+						}
+						if len(changes) > 0 {
+							updates = append(updates, connection.UpdateRow{
+								Keys:   map[string]interface{}{pkCol: sRow[pkCol]},
+								Values: changes,
+							})
+						}
+					} else {
+						inserts = append(inserts, sRow)
+					}
+				}
+
+				var deletes []map[string]interface{}
+				if opts.Delete {
+					for pkStr, row := range targetMap {
+						if _, ok := sourcePKSet[pkStr]; ok {
+							continue
+						}
+						deletes = append(deletes, map[string]interface{}{pkCol: row[pkCol]})
+					}
+				}
+
+				// apply operation selection
+				inserts = filterRowsByPKSelection(pkCol, inserts, opts.Insert, opts.SelectedInsertPKs)
+				updates = filterUpdatesByPKSelection(pkCol, updates, opts.Update, opts.SelectedUpdatePKs)
+				deletes = filterRowsByPKSelection(pkCol, deletes, opts.Delete, opts.SelectedDeletePKs)
+
+				changeSet := connection.ChangeSet{
+					Inserts: inserts,
+					Updates: updates,
+					Deletes: deletes,
+				}
+
+				// 4. Align schema (target missing columns)
+				s.progress(config.JobID, i, totalTables, tableName, "检查字段一致性")
+				requiredCols := collectRequiredColumns(changeSet.Inserts, changeSet.Updates)
+				targetCols, err := targetDB.GetColumns(targetSchema, targetTable)
+				if err != nil {
+					s.appendLog(config.JobID, &result, "warn", fmt.Sprintf("  -> 获取目标表字段失败，已跳过字段一致性检查: %v", err))
 				} else {
-					result.RowsInserted += len(inserts)
-					result.RowsUpdated += len(updates)
+					targetColSet := make(map[string]struct{}, len(targetCols))
+					for _, c := range targetCols {
+						name := strings.ToLower(strings.TrimSpace(c.Name))
+						if name == "" {
+							continue
+						}
+						targetColSet[name] = struct{}{}
+					}
+
+					missing := make([]string, 0)
+					for lower, original := range requiredCols {
+						if _, ok := targetColSet[lower]; !ok {
+							missing = append(missing, original)
+						}
+					}
+					sort.Strings(missing)
+
+					if len(missing) > 0 {
+						if config.AutoAddColumns && strings.ToLower(strings.TrimSpace(config.TargetConfig.Type)) == "mysql" {
+							s.appendLog(config.JobID, &result, "warn", fmt.Sprintf("  -> 目标表缺少字段 %d 个，开始自动补齐: %s", len(missing), strings.Join(missing, ", ")))
+							added := 0
+							for _, colName := range missing {
+								colLower := strings.ToLower(strings.TrimSpace(colName))
+								colType := "TEXT"
+								if strings.ToLower(strings.TrimSpace(config.SourceConfig.Type)) == "mysql" {
+									if srcCol, ok := sourceColsByLower[colLower]; ok {
+										colType = sanitizeMySQLColumnType(srcCol.Type)
+									}
+								}
+
+								alterSQL := fmt.Sprintf("ALTER TABLE %s ADD COLUMN %s %s NULL",
+									quoteQualifiedIdentByType("mysql", targetQueryTable),
+									quoteIdentByType("mysql", colName),
+									colType,
+								)
+								if _, err := targetDB.Exec(alterSQL); err != nil {
+									s.appendLog(config.JobID, &result, "error", fmt.Sprintf("  -> 自动补字段失败：字段=%s 错误=%v", colName, err))
+									continue
+								}
+								added++
+							}
+							s.appendLog(config.JobID, &result, "info", fmt.Sprintf("  -> 自动补字段完成：成功=%d 失败=%d", added, len(missing)-added))
+
+							// refresh columns
+							targetCols, err = targetDB.GetColumns(targetSchema, targetTable)
+							if err == nil {
+								targetColSet = make(map[string]struct{}, len(targetCols))
+								for _, c := range targetCols {
+									name := strings.ToLower(strings.TrimSpace(c.Name))
+									if name == "" {
+										continue
+									}
+									targetColSet[name] = struct{}{}
+								}
+							}
+						} else {
+							s.appendLog(config.JobID, &result, "warn", fmt.Sprintf("  -> 目标表缺少字段 %d 个（未开启自动补齐），将自动忽略：%s", len(missing), strings.Join(missing, ", ")))
+						}
+
+						// filter out still-missing columns to avoid apply failure
+						changeSet.Inserts = filterInsertRows(changeSet.Inserts, targetColSet)
+						changeSet.Updates = filterUpdateRows(changeSet.Updates, targetColSet)
+					}
+				}
+
+				// 5. Apply Changes
+				s.progress(config.JobID, i, totalTables, tableName, "应用变更")
+
+				if len(changeSet.Inserts) > 0 || len(changeSet.Updates) > 0 || len(changeSet.Deletes) > 0 {
+					s.appendLog(config.JobID, &result, "info", fmt.Sprintf("  -> 需插入: %d 行, 需更新: %d 行, 需删除: %d 行", len(changeSet.Inserts), len(changeSet.Updates), len(changeSet.Deletes)))
+
+					if applier, ok := targetDB.(db.BatchApplier); ok {
+						if err := applier.ApplyChanges(targetTable, changeSet); err != nil {
+							s.appendLog(config.JobID, &result, "error", fmt.Sprintf("  -> 应用变更失败: %v", err))
+						} else {
+							result.RowsInserted += len(changeSet.Inserts)
+							result.RowsUpdated += len(changeSet.Updates)
+							result.RowsDeleted += len(changeSet.Deletes)
+						}
+					} else {
+						s.appendLog(config.JobID, &result, "warn", "  -> 目标驱动不支持应用数据变更 (ApplyChanges).")
+					}
+				} else {
+					s.appendLog(config.JobID, &result, "info", "  -> 数据一致，无需变更.")
+				}
+
+				result.TablesSynced++
+				return
+			} else {
+				// insert_only / full_overwrite: do not compare target, just insert source rows
+				inserts = sourceRows
+			}
+
+			// full_overwrite: clear target table first
+			if tableMode == "full_overwrite" {
+				s.appendLog(config.JobID, &result, "warn", fmt.Sprintf("  -> 全量覆盖模式：即将清空目标表 %s", tableName))
+				s.progress(config.JobID, i, totalTables, tableName, "清空目标表")
+				clearSQL := ""
+				if strings.ToLower(strings.TrimSpace(config.TargetConfig.Type)) == "mysql" {
+					clearSQL = fmt.Sprintf("TRUNCATE TABLE %s", quoteQualifiedIdentByType(config.TargetConfig.Type, targetQueryTable))
+				} else {
+					clearSQL = fmt.Sprintf("DELETE FROM %s", quoteQualifiedIdentByType(config.TargetConfig.Type, targetQueryTable))
+				}
+				if _, err := targetDB.Exec(clearSQL); err != nil {
+					s.appendLog(config.JobID, &result, "error", fmt.Sprintf("  -> 清空目标表失败: %v", err))
+					return
+				}
+			}
+
+			// 4. Align schema (target missing columns)
+			s.progress(config.JobID, i, totalTables, tableName, "检查字段一致性")
+			requiredCols := collectRequiredColumns(inserts, updates)
+			targetCols, err := targetDB.GetColumns(targetSchema, targetTable)
+			if err != nil {
+				s.appendLog(config.JobID, &result, "warn", fmt.Sprintf("  -> 获取目标表字段失败，已跳过字段一致性检查: %v", err))
+			} else {
+				targetColSet := make(map[string]struct{}, len(targetCols))
+				for _, c := range targetCols {
+					name := strings.ToLower(strings.TrimSpace(c.Name))
+					if name == "" {
+						continue
+					}
+					targetColSet[name] = struct{}{}
+				}
+
+				missing := make([]string, 0)
+				for lower, original := range requiredCols {
+					if _, ok := targetColSet[lower]; !ok {
+						missing = append(missing, original)
+					}
+				}
+				sort.Strings(missing)
+
+				if len(missing) > 0 {
+					if config.AutoAddColumns && strings.ToLower(strings.TrimSpace(config.TargetConfig.Type)) == "mysql" {
+						s.appendLog(config.JobID, &result, "warn", fmt.Sprintf("  -> 目标表缺少字段 %d 个，开始自动补齐: %s", len(missing), strings.Join(missing, ", ")))
+						added := 0
+						for _, colName := range missing {
+							colLower := strings.ToLower(strings.TrimSpace(colName))
+							colType := "TEXT"
+							if strings.ToLower(strings.TrimSpace(config.SourceConfig.Type)) == "mysql" {
+								if srcCol, ok := sourceColsByLower[colLower]; ok {
+									colType = sanitizeMySQLColumnType(srcCol.Type)
+								}
+							}
+
+							alterSQL := fmt.Sprintf("ALTER TABLE %s ADD COLUMN %s %s NULL",
+								quoteQualifiedIdentByType("mysql", targetQueryTable),
+								quoteIdentByType("mysql", colName),
+								colType,
+							)
+							if _, err := targetDB.Exec(alterSQL); err != nil {
+								s.appendLog(config.JobID, &result, "error", fmt.Sprintf("  -> 自动补字段失败：字段=%s 错误=%v", colName, err))
+								continue
+							}
+							added++
+						}
+						s.appendLog(config.JobID, &result, "info", fmt.Sprintf("  -> 自动补字段完成：成功=%d 失败=%d", added, len(missing)-added))
+
+						// refresh columns
+						targetCols, err = targetDB.GetColumns(targetSchema, targetTable)
+						if err == nil {
+							targetColSet = make(map[string]struct{}, len(targetCols))
+							for _, c := range targetCols {
+								name := strings.ToLower(strings.TrimSpace(c.Name))
+								if name == "" {
+									continue
+								}
+								targetColSet[name] = struct{}{}
+							}
+						}
+					} else {
+						s.appendLog(config.JobID, &result, "warn", fmt.Sprintf("  -> 目标表缺少字段 %d 个（未开启自动补齐），将自动忽略：%s", len(missing), strings.Join(missing, ", ")))
+					}
+
+					// filter out still-missing columns to avoid apply failure
+					inserts = filterInsertRows(inserts, targetColSet)
+					updates = filterUpdateRows(updates, targetColSet)
+				}
+			}
+
+			// 5. Apply Changes
+			s.progress(config.JobID, i, totalTables, tableName, "应用变更")
+			changeSet := connection.ChangeSet{
+				Inserts: inserts,
+				Updates: updates,
+			}
+
+			if len(changeSet.Inserts) > 0 || len(changeSet.Updates) > 0 {
+				s.appendLog(config.JobID, &result, "info", fmt.Sprintf("  -> 需插入: %d 行, 需更新: %d 行", len(changeSet.Inserts), len(changeSet.Updates)))
+
+				if applier, ok := targetDB.(db.BatchApplier); ok {
+					if err := applier.ApplyChanges(targetTable, changeSet); err != nil {
+						s.appendLog(config.JobID, &result, "error", fmt.Sprintf("  -> 应用变更失败: %v", err))
+					} else {
+						result.RowsInserted += len(changeSet.Inserts)
+						result.RowsUpdated += len(changeSet.Updates)
+					}
+				} else {
+					s.appendLog(config.JobID, &result, "warn", "  -> 目标驱动不支持应用数据变更 (ApplyChanges).")
 				}
 			} else {
-				result.Logs = append(result.Logs, "  -> 目标驱动不支持应用数据变更 (ApplyChanges).")
+				s.appendLog(config.JobID, &result, "info", "  -> 数据一致，无需变更.")
 			}
-		} else {
-			result.Logs = append(result.Logs, "  -> 数据一致，无需变更.")
-		}

-		result.TablesSynced++
+			result.TablesSynced++
+		}()
 	}

+	s.progress(config.JobID, totalTables, totalTables, "", "同步完成")
 	return result
 }

@@ -196,9 +505,52 @@ func formatConnSummaryForSync(config connection.ConnectionConfig) string {
 		config.Type, config.Host, config.Port, dbName, config.User, timeoutSeconds)
 }

-func (s *SyncEngine) fail(res SyncResult, msg string) SyncResult {
+func (s *SyncEngine) appendLog(jobID string, res *SyncResult, level string, msg string) {
+	if res != nil {
+		res.Logs = append(res.Logs, msg)
+	}
+	if s.reporter.OnLog != nil && strings.TrimSpace(jobID) != "" {
+		s.reporter.OnLog(SyncLogEvent{
+			JobID:   jobID,
+			Level:   level,
+			Message: msg,
+			Ts:      time.Now().UnixMilli(),
+		})
+	}
+}
+
+func (s *SyncEngine) progress(jobID string, current, total int, table string, stage string) {
+	if s.reporter.OnProgress == nil || strings.TrimSpace(jobID) == "" {
+		return
+	}
+	percent := 0
+	if total <= 0 {
+		if current > 0 {
+			percent = 100
+		}
+	} else {
+		if current < 0 {
+			current = 0
+		}
+		if current > total {
+			current = total
+		}
+		percent = (current * 100) / total
+	}
+	s.reporter.OnProgress(SyncProgressEvent{
+		JobID:   jobID,
+		Percent: percent,
+		Current: current,
+		Total:   total,
+		Table:   table,
+		Stage:   stage,
+	})
+}
+
+func (s *SyncEngine) fail(jobID string, totalTables int, res SyncResult, msg string) SyncResult {
 	res.Success = false
 	res.Message = msg
-	res.Logs = append(res.Logs, "致命错误: "+msg)
+	s.appendLog(jobID, &res, "error", "致命错误: "+msg)
+	s.progress(jobID, res.TablesSynced, totalTables, "", "同步失败")
 	return res
 }
--- a/internal/sync/sync_events.go
+++ b/internal/sync/sync_events.go
@@ -0,0 +1,30 @@
+package sync
+
+const (
+	EventSyncStart    = "sync:start"
+	EventSyncProgress = "sync:progress"
+	EventSyncLog      = "sync:log"
+	EventSyncDone     = "sync:done"
+)
+
+type SyncLogEvent struct {
+	JobID   string `json:"jobId"`
+	Level   string `json:"level"` // info/warn/error
+	Message string `json:"message"`
+	Ts      int64  `json:"ts"` // Unix milli
+}
+
+type SyncProgressEvent struct {
+	JobID   string `json:"jobId"`
+	Percent int    `json:"percent"`
+	Current int    `json:"current"` // 已完成表数
+	Total   int    `json:"total"`   // 总表数
+	Table   string `json:"table,omitempty"`
+	Stage   string `json:"stage,omitempty"`
+}
+
+type Reporter struct {
+	OnLog      func(event SyncLogEvent)
+	OnProgress func(event SyncProgressEvent)
+}
+
--- a/internal/sync/table_options.go
+++ b/internal/sync/table_options.go
@@ -0,0 +1,13 @@
+package sync
+
+// TableOptions controls which operations to apply per table, and optional row selection.
+// 注意：如未指定 Selected*PKs，则表示“同步全部该类型差异数据”；如指定为空数组，则同样表示全部。
+type TableOptions struct {
+	Insert bool `json:"insert,omitempty"`
+	Update bool `json:"update,omitempty"`
+	Delete bool `json:"delete,omitempty"`
+
+	SelectedInsertPKs []string `json:"selectedInsertPks,omitempty"`
+	SelectedUpdatePKs []string `json:"selectedUpdatePks,omitempty"`
+	SelectedDeletePKs []string `json:"selectedDeletePks,omitempty"`
+}