GithubBackup/MyGoNavi
1
0
Fork 0
mirror of https://github.com/Syngnat/GoNavi.git synced 2026-08-15 11:14:31 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(sync): support composite primary keys

Browse Source
This commit is contained in:
Kunghim
2026-08-09 19:53:24 +08:00
parent bf5a1b7681
commit 52cd1936f7
14 changed files with 808 additions and 386 deletions
Download Patch File Download Diff File Expand all files Collapse all files

98
frontend/src/components/DataSyncModal.sql-preview.test.ts Normal file
View File

@@ -0,0 +1,98 @@
import { describe, expect, it } from 'vitest';
import { buildSqlPreview } from './DataSyncModal';
describe('buildSqlPreview', () => {
it('uses every composite primary-key column for updates and deletes', () => {
const result = buildSqlPreview(
{
pkColumn: 'tenant_id,order_id',
pkColumns: ['tenant_id', 'order_id'],
columnTypes: {
tenant_id: 'bigint',
order_id: 'bigint',
status: 'varchar(32)',
},
inserts: [],
updates: [
{
pk: '[1,7]',
changedColumns: ['status'],
source: { tenant_id: 1, order_id: 7, status: 'paid' },
target: { tenant_id: 1, order_id: 7, status: 'pending' },
},
],
deletes: [
{
pk: '[3,7]',
row: { tenant_id: 3, order_id: 7, status: 'old' },
},
],
},
'orders',
'mysql',
{ insert: true, update: true, delete: true },
);
expect(result.statementCount).toBe(2);
expect(result.sqlText).toContain(
"UPDATE `orders` SET `status` = 'paid' WHERE `tenant_id` = 1 AND `order_id` = 7;",
);
expect(result.sqlText).toContain(
'DELETE FROM `orders` WHERE `tenant_id` = 3 AND `order_id` = 7;',
);
expect(result.sqlText).not.toContain('`tenant_id,order_id`');
});
it('keeps legacy single-column preview responses compatible', () => {
const result = buildSqlPreview(
{
pkColumn: 'id',
columnTypes: { id: 'bigint', name: 'varchar(32)' },
inserts: [],
updates: [
{
pk: '9',
changedColumns: ['name'],
source: { name: 'new' },
},
],
deletes: [],
},
'users',
'postgresql',
{ insert: true, update: true, delete: false },
);
expect(result.sqlText).toBe(
`UPDATE "users" SET "name" = 'new' WHERE "id" = '9';`,
);
});
it('quotes every PostgreSQL composite-key predicate', () => {
const result = buildSqlPreview(
{
pkColumn: 'tenant_id,order_id',
pkColumns: ['tenant_id', 'order_id'],
columnTypes: { tenant_id: 'bigint', order_id: 'bigint', status: 'text' },
inserts: [],
updates: [
{
pk: '[1,7]',
changedColumns: ['status'],
source: { tenant_id: 1, order_id: 7, status: 'paid' },
target: { tenant_id: 1, order_id: 7, status: 'pending' },
},
],
deletes: [],
},
'public.orders',
'postgresql',
{ insert: true, update: true, delete: false },
);
expect(result.sqlText).toContain(
`WHERE "tenant_id" = 1 AND "order_id" = 7;`,
);
});
});

51
frontend/src/components/DataSyncModal.tsx
View File

@@ -264,7 +264,7 @@ const isServiceNameBackedSyncConnection = (conn?: SavedConnection): boolean => {
return protocol === "oracle";
};
const buildSqlPreview = (
export const buildSqlPreview = (
previewData: any,
tableName: string,
dbType: string,
@@ -273,6 +273,13 @@ const buildSqlPreview = (
if (!previewData || !tableName) return { sqlText: "", statementCount: 0 };
const tableExpr = quoteSqlTable(dbType, tableName);
const pkCol = String(previewData.pkColumn || "id");
const pkColumns = Array.isArray(previewData.pkColumns)
? previewData.pkColumns
.map((column: unknown) => String(column || "").trim())
.filter((column: string) => column.length > 0)
: [];
if (pkColumns.length === 0) pkColumns.push(pkCol);
const pkColumnSet = new Set(pkColumns);
const columnTypesByLowerName =
previewData?.columnTypes && typeof previewData.columnTypes === "object"
? (previewData.columnTypes as Record<string, string>)
@@ -308,6 +315,30 @@ const buildSqlPreview = (
(ops?.selectedDeletePks || []).map((v) => String(v)),
);
const buildWhereExpr = (rowWrap: any, fallbackPk: string): string => {
const locator = {
...(rowWrap?.source || {}),
...(rowWrap?.target || {}),
...(rowWrap?.row || {}),
} as Record<string, unknown>;
const conditions: string[] = [];
for (const column of pkColumns) {
let value = locator[column];
if (
value === undefined &&
pkColumns.length === 1 &&
fallbackPk !== ""
) {
value = fallbackPk;
}
if (value === undefined) return "";
conditions.push(
`${quoteSqlIdent(dbType, column)} = ${toTypedSqlLiteral(value, dbType, columnTypesByLowerName[column.toLowerCase()])}`,
);
}
return conditions.join(" AND ");
};
if (ops?.insert !== false) {
insertRows.forEach((rowWrap: any) => {
const pk = String(rowWrap?.pk ?? "");
@@ -338,8 +369,10 @@ const buildSqlPreview = (
const source = rowWrap?.source || {};
const changedColumns = Array.isArray(rowWrap?.changedColumns)
? rowWrap.changedColumns
: Object.keys(source).filter((k) => k !== pkCol);
const setCols = changedColumns.filter((c: string) => String(c) !== pkCol);
: Object.keys(source).filter((k) => !pkColumnSet.has(k));
const setCols = changedColumns.filter(
(c: string) => !pkColumnSet.has(String(c)),
);
if (setCols.length === 0) return;
const setExpr = setCols
.map(
@@ -347,9 +380,9 @@ const buildSqlPreview = (
`${quoteSqlIdent(dbType, c)} = ${toTypedSqlLiteral(source[c], dbType, columnTypesByLowerName[String(c).toLowerCase()])}`,
)
.join(", ");
statements.push(
`UPDATE ${tableExpr} SET ${setExpr} WHERE ${quoteSqlIdent(dbType, pkCol)} = ${toTypedSqlLiteral(pk, dbType, columnTypesByLowerName[String(pkCol).toLowerCase()])};`,
);
const whereExpr = buildWhereExpr(rowWrap, pk);
if (whereExpr === "") return;
statements.push(`UPDATE ${tableExpr} SET ${setExpr} WHERE ${whereExpr};`);
});
}
@@ -357,9 +390,9 @@ const buildSqlPreview = (
deleteRows.forEach((rowWrap: any) => {
const pk = String(rowWrap?.pk ?? "");
if (selectedDelete.size > 0 && !selectedDelete.has(pk)) return;
statements.push(
`DELETE FROM ${tableExpr} WHERE ${quoteSqlIdent(dbType, pkCol)} = ${toTypedSqlLiteral(pk, dbType, columnTypesByLowerName[String(pkCol).toLowerCase()])};`,
);
const whereExpr = buildWhereExpr(rowWrap, pk);
if (whereExpr === "") return;
statements.push(`DELETE FROM ${tableExpr} WHERE ${whereExpr};`);
});
}

57
internal/db/composite_key_applychanges_test.go Normal file
View File

@@ -0,0 +1,57 @@
package db
import (
"GoNavi-Wails/internal/connection"
"strings"
"testing"
)
func TestMySQLApplyChangesUsesEveryCompositeKeyColumn(t *testing.T) {
dbConn, state := openOracleRecordingDB(t)
database := &MySQLDB{conn: dbConn}
changes := connection.ChangeSet{
Deletes: []map[string]interface{}{{"tenant_id": 3, "order_id": 7}},
Updates: []connection.UpdateRow{{
Keys: map[string]interface{}{"tenant_id": 1, "order_id": 7},
Values: map[string]interface{}{"status": "paid"},
}},
}
if err := database.ApplyChanges("orders", changes); err != nil {
t.Fatalf("ApplyChanges() error = %v", err)
}
queries := state.snapshotExecQueries()
if len(queries) != 2 {
t.Fatalf("exec queries = %#v", queries)
}
for _, query := range queries {
if !strings.Contains(query, "`tenant_id` = ?") || !strings.Contains(query, "`order_id` = ?") || !strings.Contains(query, " AND ") {
t.Fatalf("composite-key predicate missing from %q", query)
}
}
}
func TestPostgresApplyChangesUsesEveryCompositeKeyColumn(t *testing.T) {
dbConn, state := openOracleRecordingDB(t)
database := &PostgresDB{conn: dbConn}
changes := connection.ChangeSet{
Deletes: []map[string]interface{}{{"tenant_id": 3, "order_id": 7}},
Updates: []connection.UpdateRow{{
Keys: map[string]interface{}{"tenant_id": 1, "order_id": 7},
Values: map[string]interface{}{"status": "paid"},
}},
}
if err := database.ApplyChanges("public.orders", changes); err != nil {
t.Fatalf("ApplyChanges() error = %v", err)
}
queries := state.snapshotExecQueries()
if len(queries) != 2 {
t.Fatalf("exec queries = %#v", queries)
}
for _, query := range queries {
if !strings.Contains(query, `"tenant_id" = $`) || !strings.Contains(query, `"order_id" = $`) || !strings.Contains(query, " AND ") {
t.Fatalf("composite-key predicate missing from %q", query)
}
}
}

76
internal/sync/analyze.go
View File

@@ -204,31 +204,26 @@ func (s *SyncEngine) Analyze(config SyncConfig) SyncAnalyzeResult {
result.Tables = append(result.Tables, summary)
return
}
if len(pkCols) > 1 {
summary.Message = localizedSyncBackendText("data_sync.backend.error.diff_composite_pk_unsupported", map[string]any{
"columns": strings.Join(pkCols, ","),
})
result.Tables = append(result.Tables, summary)
return
}
sourcePKCol := pkCols[0]
comparisonPKCol := sourcePKCol
comparisonPKCols := append([]string(nil), pkCols...)
if hasExplicitSyncMappings(config) {
mappedPK, ok := projection.TargetColumn(sourcePKCol)
if !ok || strings.TrimSpace(mappedPK) == "" {
summary.Message = fmt.Sprintf("表 %s 的主键字段 %s 未映射到目标字段，无法执行差异分析", tableName, sourcePKCol)
result.Tables = append(result.Tables, summary)
return
for index, sourceKey := range pkCols {
mappedPK, ok := projection.TargetColumn(sourceKey)
if !ok || strings.TrimSpace(mappedPK) == "" {
summary.Message = fmt.Sprintf("表 %s 的主键字段 %s 未映射到目标字段，无法执行差异分析", tableName, sourceKey)
result.Tables = append(result.Tables, summary)
return
}
comparisonPKCols[index] = mappedPK
}
comparisonPKCol = mappedPK
}
summary.PKColumn = comparisonPKCol
summary.PKColumn = strings.Join(comparisonPKCols, ",")
targetColSet := buildTargetColumnSet(targetCols)
handled := false
counts := pagedDiffCounts{}
var scanErr error
if !hasExplicitSyncMappings(config) {
if !hasExplicitSyncMappings(config) && len(pkCols) == 1 {
handled, counts, scanErr = scanTableDiffInPages(sourceDB, targetDB, sourceType, targetType, plan, cols, targetCols, sourcePKCol, targetColSet, true, nil)
}
if handled {
@@ -270,53 +265,8 @@ func (s *SyncEngine) Analyze(config SyncConfig) SyncAnalyzeResult {
return
}
pkCol := comparisonPKCol
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++
}
}
inserts, updates, deletes, same := diffRowsByKeyColumns(comparisonPKCols, sourceRows, targetRows)
summary.Inserts, summary.Updates, summary.Deletes, summary.Same = len(inserts), len(updates), len(deletes), same
summary.CanSync = true
if strings.TrimSpace(summary.Message) == "" {

28
internal/sync/analyze_i18n_test.go
View File

@@ -42,7 +42,6 @@ func assertNoLegacyAnalyzeChinese(t *testing.T, text string) {
}
}
func TestAnalyzeCatalogKeysExist(t *testing.T) {
catalogs, err := i18n.LoadCatalogs()
if err != nil {
@@ -232,11 +231,6 @@ func TestAnalyzeUsesCurrentLanguageForReadAndPKMessages(t *testing.T) {
{Name: "id", Type: "bigint", Nullable: "NO"},
{Name: "name", Type: "varchar(64)", Nullable: "YES"},
}
compositePKCols := []connection.ColumnDefinition{
{Name: "id", Type: "bigint", Nullable: "NO", Key: "PRI"},
{Name: "tenant_id", Type: "bigint", Nullable: "NO", Key: "PRI"},
{Name: "name", Type: "varchar(64)", Nullable: "YES"},
}
singlePKCols := []connection.ColumnDefinition{
{Name: "id", Type: "bigint", Nullable: "NO", Key: "PRI"},
{Name: "name", Type: "varchar(64)", Nullable: "YES"},
@@ -290,28 +284,6 @@ func TestAnalyzeUsesCurrentLanguageForReadAndPKMessages(t *testing.T) {
},
wantKey: "data_sync.backend.error.diff_pk_required",
},
{
name: "composite primary key unsupported",
sourceDB: &fakeMigrationDB{
columns: map[string][]connection.ColumnDefinition{
"app.users": compositePKCols,
},
queryData: map[string][]map[string]interface{}{
countQuery: {
{"__gonavi_count__": 2},
},
},
},
targetDB: &fakeMigrationDB{
columns: map[string][]connection.ColumnDefinition{
"app.users": compositePKCols,
},
},
wantKey: "data_sync.backend.error.diff_composite_pk_unsupported",
wantParams: map[string]any{
"columns": "id,tenant_id",
},
},
}
for _, tc := range cases {

228
internal/sync/composite_key_sync_test.go Normal file
View File

@@ -0,0 +1,228 @@
package sync
import (
"GoNavi-Wails/internal/connection"
"reflect"
"strings"
"testing"
)
func compositeKeyColumns() []connection.ColumnDefinition {
return []connection.ColumnDefinition{
{Name: "tenant_id", Type: "bigint", Nullable: "NO", Key: "PRI"},
{Name: "order_id", Type: "bigint", Nullable: "NO", Key: "PRI"},
{Name: "status", Type: "varchar(32)", Nullable: "YES"},
}
}
func compositeKeyRows() ([]map[string]interface{}, []map[string]interface{}) {
return []map[string]interface{}{
{"tenant_id": int64(1), "order_id": int64(7), "status": "paid"},
{"tenant_id": int64(2), "order_id": int64(7), "status": "new"},
}, []map[string]interface{}{
{"tenant_id": int64(1), "order_id": int64(7), "status": "pending"},
{"tenant_id": int64(3), "order_id": int64(7), "status": "old"},
}
}
func compositeTableConfig() SyncConfig {
return SyncConfig{
SourceConfig: connection.ConnectionConfig{Type: "mysql", Database: "source_db", Host: "source.local"},
TargetConfig: connection.ConnectionConfig{Type: "mysql", Database: "target_db", Host: "target.local"},
Tables: []string{"orders"},
Content: "data",
Mode: "insert_update",
TargetTableStrategy: "existing_only",
TableOptions: map[string]TableOptions{
"orders": {Insert: true, Update: true, Delete: true},
},
}
}
func compositeTableDatabases() (*fakeMigrationDB, *fakeQuerySyncTargetDB) {
sourceRows, targetRows := compositeKeyRows()
columns := compositeKeyColumns()
source := &fakeMigrationDB{
columns: map[string][]connection.ColumnDefinition{"source_db.orders": columns},
queryData: map[string][]map[string]interface{}{
"SELECT * FROM `source_db`.`orders`": sourceRows,
},
}
target := &fakeQuerySyncTargetDB{fakeMigrationDB: fakeMigrationDB{
columns: map[string][]connection.ColumnDefinition{"target_db.orders": columns},
queryData: map[string][]map[string]interface{}{
"SELECT * FROM `target_db`.`orders`": targetRows,
},
}}
return source, target
}
func TestAnalyzeAndPreviewAcceptCompositePrimaryKey(t *testing.T) {
t.Run("analyze", func(t *testing.T) {
source, target := compositeTableDatabases()
useSyncDatabaseFactorySequence(t,
syncDatabaseFactoryStep{db: source},
syncDatabaseFactoryStep{db: target},
)
result := NewSyncEngine(Reporter{}).Analyze(compositeTableConfig())
if !result.Success || len(result.Tables) != 1 {
t.Fatalf("Analyze() = %+v", result)
}
summary := result.Tables[0]
if !summary.CanSync || summary.PKColumn != "tenant_id,order_id" || summary.Inserts != 1 || summary.Updates != 1 || summary.Deletes != 1 {
t.Fatalf("composite-key summary = %+v", summary)
}
})
t.Run("preview", func(t *testing.T) {
source, target := compositeTableDatabases()
useSyncDatabaseFactorySequence(t,
syncDatabaseFactoryStep{db: source},
syncDatabaseFactoryStep{db: target},
)
preview, err := NewSyncEngine(Reporter{}).Preview(compositeTableConfig(), "orders", 20)
if err != nil {
t.Fatalf("Preview() error = %v", err)
}
if preview.PKColumn != "tenant_id,order_id" || !reflect.DeepEqual(preview.PKColumns, []string{"tenant_id", "order_id"}) || preview.TotalInserts != 1 || preview.TotalUpdates != 1 || preview.TotalDeletes != 1 {
t.Fatalf("composite-key preview = %+v", preview)
}
if len(preview.Inserts) != 1 || preview.Inserts[0].PK != `[2,7]` {
t.Fatalf("preview inserts = %#v", preview.Inserts)
}
if len(preview.Updates) != 1 || preview.Updates[0].PK != `[1,7]` || preview.Updates[0].Source["status"] != "paid" || preview.Updates[0].Target["status"] != "pending" {
t.Fatalf("preview updates = %#v", preview.Updates)
}
if len(preview.Deletes) != 1 || preview.Deletes[0].PK != `[3,7]` {
t.Fatalf("preview deletes = %#v", preview.Deletes)
}
})
}
func TestRunSyncMySQLLikeAppliesCompositePrimaryKeyChanges(t *testing.T) {
source, target := compositeTableDatabases()
useSyncDatabaseFactorySequence(t,
syncDatabaseFactoryStep{db: source},
syncDatabaseFactoryStep{db: target},
)
result := NewSyncEngine(Reporter{}).RunSync(compositeTableConfig())
if !result.Success || result.TablesSynced != 1 || result.RowsInserted != 1 || result.RowsUpdated != 1 || result.RowsDeleted != 1 {
t.Fatalf("RunSync() = %+v", result)
}
if len(target.appliedChanges.Updates) != 1 || !reflect.DeepEqual(target.appliedChanges.Updates[0].Keys, map[string]interface{}{"tenant_id": int64(1), "order_id": int64(7)}) {
t.Fatalf("updates = %#v", target.appliedChanges.Updates)
}
if len(target.appliedChanges.Deletes) != 1 || !reflect.DeepEqual(target.appliedChanges.Deletes[0], map[string]interface{}{"tenant_id": int64(3), "order_id": int64(7)}) {
t.Fatalf("deletes = %#v", target.appliedChanges.Deletes)
}
}
func TestRunSyncPostgresLikeSourceQueryMapsCompositePrimaryKey(t *testing.T) {
const sourceSQL = "SELECT tenant, order_no, status FROM active_orders"
sourceRows, targetRows := compositeKeyRows()
source := &fakeMigrationDB{queryData: map[string][]map[string]interface{}{
sourceSQL: {
{"tenant": sourceRows[0]["tenant_id"], "order_no": sourceRows[0]["order_id"], "status": sourceRows[0]["status"]},
{"tenant": sourceRows[1]["tenant_id"], "order_no": sourceRows[1]["order_id"], "status": sourceRows[1]["status"]},
},
}}
target := &fakeQuerySyncTargetDB{fakeMigrationDB: fakeMigrationDB{
columns: map[string][]connection.ColumnDefinition{"public.orders": compositeKeyColumns()},
queryData: map[string][]map[string]interface{}{
`SELECT * FROM "public"."orders"`: targetRows,
},
}}
useSyncDatabaseFactorySequence(t,
syncDatabaseFactoryStep{db: source},
syncDatabaseFactoryStep{db: target},
)
result := NewSyncEngine(Reporter{}).RunSync(SyncConfig{
SourceConfig: connection.ConnectionConfig{Type: "postgres", Database: "source_db"},
TargetConfig: connection.ConnectionConfig{Type: "postgres", Database: "target_db"},
SourceQuery: sourceSQL,
Content: "data",
Mode: "insert_update",
Mappings: []SyncObjectMapping{{
Source: SyncObjectRef{Name: "active_orders"},
Target: SyncObjectRef{Schema: "public", Name: "orders"},
KeyColumns: []string{"order_no", "tenant"},
Columns: []SyncColumnMapping{
{Source: "tenant", Target: "tenant_id"},
{Source: "order_no", Target: "order_id"},
{Source: "status", Target: "status"},
},
}},
TableOptions: map[string]TableOptions{
"active_orders": {Insert: true, Update: true, Delete: true},
},
})
if !result.Success || result.RowsInserted != 1 || result.RowsUpdated != 1 || result.RowsDeleted != 1 {
t.Fatalf("RunSync() = %+v", result)
}
if target.appliedTable != "public.orders" {
t.Fatalf("applied table = %q", target.appliedTable)
}
if len(target.appliedChanges.Updates) != 1 || len(target.appliedChanges.Updates[0].Keys) != 2 {
t.Fatalf("updates = %#v", target.appliedChanges.Updates)
}
if len(target.appliedChanges.Deletes) != 1 || len(target.appliedChanges.Deletes[0]) != 2 {
t.Fatalf("deletes = %#v", target.appliedChanges.Deletes)
}
}
func TestLoadSourceQueryContextRejectsInvalidCompositeKeyMappings(t *testing.T) {
target := &fakeMigrationDB{columns: map[string][]connection.ColumnDefinition{
"app.orders": compositeKeyColumns(),
}}
baseColumns := []SyncColumnMapping{
{Source: "tenant", Target: "tenant_id"},
{Source: "order_no", Target: "order_id"},
{Source: "status_raw", Target: "status"},
}
cases := []struct {
name string
keyColumns []string
want string
}{
{name: "missing key", keyColumns: []string{"tenant"}, want: "数量一致"},
{name: "duplicate key", keyColumns: []string{"tenant", "tenant"}, want: "字段重复"},
{name: "non primary key", keyColumns: []string{"order_no", "status_raw"}, want: "必须属于目标表主键"},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
config := SyncConfig{
SourceConfig: connection.ConnectionConfig{Type: "mysql", Database: "source_db"},
TargetConfig: connection.ConnectionConfig{Type: "mysql", Database: "app"},
SourceQuery: "SELECT tenant, order_no, status_raw FROM active_orders",
Content: "data",
Mode: "insert_update",
Mappings: []SyncObjectMapping{{
Source: SyncObjectRef{Name: "active_orders"},
Target: SyncObjectRef{Schema: "app", Name: "orders"},
KeyColumns: tc.keyColumns,
Columns: baseColumns,
}},
}
_, err := loadSourceQuerySyncContext(config, &fakeMigrationDB{}, target, false, false, true)
if err == nil || !strings.Contains(err.Error(), tc.want) {
t.Fatalf("loadSourceQuerySyncContext() error = %v, want containing %q", err, tc.want)
}
})
}
}
func TestCompositePrimaryKeySelectionUsesPreviewTuple(t *testing.T) {
rows := []map[string]interface{}{
{"tenant_id": int64(1), "order_id": int64(7)},
{"tenant_id": int64(2), "order_id": int64(7)},
}
selected := filterRowsByKeySelection([]string{"tenant_id", "order_id"}, rows, true, []string{`[2,7]`})
if len(selected) != 1 || selected[0]["tenant_id"] != int64(2) {
t.Fatalf("selected rows = %#v", selected)
}
}

112
internal/sync/preview.go
View File

@@ -20,6 +20,7 @@ type PreviewUpdateRow struct {
type TableDiffPreview struct {
Table string `json:"table"`
PKColumn string `json:"pkColumn"`
PKColumns []string `json:"pkColumns,omitempty"`
ColumnTypes map[string]string `json:"columnTypes,omitempty"`
SchemaSummary string `json:"schemaSummary,omitempty"`
SchemaWarnings []string `json:"schemaWarnings,omitempty"`
@@ -108,26 +109,25 @@ func (s *SyncEngine) Preview(config SyncConfig, tableName string, limit int) (Ta
if len(pkCols) == 0 {
return TableDiffPreview{}, syncTextError("data_sync.backend.error.preview_pk_required", nil)
}
if len(pkCols) > 1 {
return TableDiffPreview{}, syncTextError("data_sync.backend.error.preview_composite_pk_unsupported", map[string]any{
"columns": strings.Join(pkCols, ","),
})
}
sourcePKCol := pkCols[0]
pkCol := sourcePKCol
pkColsForCompare := append([]string(nil), pkCols...)
if hasExplicitSyncMappings(config) {
mappedPK, ok := projection.TargetColumn(sourcePKCol)
if !ok || strings.TrimSpace(mappedPK) == "" {
return TableDiffPreview{}, fmt.Errorf("表 %s 的主键字段 %s 未映射到目标字段，无法生成差异预览", tableName, sourcePKCol)
for index, sourceKey := range pkCols {
mappedPK, ok := projection.TargetColumn(sourceKey)
if !ok || strings.TrimSpace(mappedPK) == "" {
return TableDiffPreview{}, fmt.Errorf("表 %s 的主键字段 %s 未映射到目标字段，无法生成差异预览", tableName, sourceKey)
}
pkColsForCompare[index] = mappedPK
}
pkCol = mappedPK
}
pkCol := pkColsForCompare[0]
sourceType := resolveMigrationDBType(config.SourceConfig)
targetType := resolveMigrationDBType(config.TargetConfig)
out := TableDiffPreview{
Table: tableName,
PKColumn: pkCol,
PKColumn: strings.Join(pkColsForCompare, ","),
PKColumns: append([]string(nil), pkColsForCompare...),
ColumnTypes: make(map[string]string, len(cols)),
SchemaSummary: firstNonEmpty(plan.PlannedAction, "结构预览"),
SchemaWarnings: append([]string(nil), plan.Warnings...),
@@ -191,17 +191,17 @@ func (s *SyncEngine) Preview(config SyncConfig, tableName string, limit int) (Ta
if len(out.Inserts) >= limit {
break
}
pkVal := strings.TrimSpace(fmt.Sprintf("%v", row[pkCol]))
if pkVal == "" || pkVal == "<nil>" {
key, ok := selectionRowKey(row, pkColsForCompare)
if !ok {
continue
}
out.Inserts = append(out.Inserts, PreviewRow{PK: pkVal, Row: row})
out.Inserts = append(out.Inserts, PreviewRow{PK: key, Row: row})
}
return out, nil
}
handled := false
if !hasExplicitSyncMappings(config) {
if !hasExplicitSyncMappings(config) && len(pkCols) == 1 {
handled, _, err = scanTableDiffInPages(sourceDB, targetDB, sourceType, targetType, plan, cols, nil, sourcePKCol, targetColSet, true, func(page pagedDiffPage) error {
out.TotalInserts += len(page.Inserts)
out.TotalUpdates += len(page.Updates)
@@ -271,58 +271,46 @@ func (s *SyncEngine) Preview(config SyncConfig, tableName string, limit int) (Ta
}
}
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 {
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++
inserts, updates, deletes, _ := diffRowsByKeyColumns(pkColsForCompare, sourceRows, targetRows)
out.TotalInserts, out.TotalUpdates, out.TotalDeletes = len(inserts), len(updates), len(deletes)
for _, row := range inserts {
if len(out.Inserts) < limit {
out.Inserts = append(out.Inserts, PreviewRow{PK: pkVal, Row: sRow})
key, _ := selectionRowKey(row, pkColsForCompare)
out.Inserts = append(out.Inserts, PreviewRow{PK: key, Row: row})
}
}
for pkVal, row := range targetMap {
if _, ok := sourcePKSet[pkVal]; ok {
continue
targetRowsByKey := make(map[string]map[string]interface{}, len(targetRows))
for _, row := range targetRows {
if key, ok := syncRowKey(row, pkColsForCompare); ok {
targetRowsByKey[key] = row
}
out.TotalDeletes++
}
sourceRowsByKey := make(map[string]map[string]interface{}, len(sourceRows))
for _, row := range sourceRows {
if key, ok := syncRowKey(row, pkColsForCompare); ok {
sourceRowsByKey[key] = row
}
}
for _, update := range updates {
if len(out.Updates) < limit {
identityKey, _ := syncRowKey(update.Keys, pkColsForCompare)
displayKey, _ := selectionRowKey(update.Keys, pkColsForCompare)
changed := make([]string, 0, len(update.Values))
for column := range update.Values {
changed = append(changed, column)
}
out.Updates = append(out.Updates, PreviewUpdateRow{
PK: displayKey,
ChangedColumns: changed,
Source: sourceRowsByKey[identityKey],
Target: targetRowsByKey[identityKey],
})
}
}
for _, row := range deletes {
if len(out.Deletes) < limit {
out.Deletes = append(out.Deletes, PreviewRow{PK: pkVal, Row: row})
key, _ := selectionRowKey(row, pkColsForCompare)
out.Deletes = append(out.Deletes, PreviewRow{PK: key, Row: row})
}
}

28
internal/sync/preview_i18n_test.go
View File

@@ -34,7 +34,6 @@ func assertNoLegacyPreviewChinese(t *testing.T, text string) {
}
}
func TestPreviewCatalogKeysExist(t *testing.T) {
catalogs, err := i18n.LoadCatalogs()
if err != nil {
@@ -184,33 +183,6 @@ func TestPreviewUsesCurrentLanguageForPreflightErrors(t *testing.T) {
},
wantKey: "data_sync.backend.error.preview_pk_required",
},
{
name: "preview composite primary key unsupported",
run: func(t *testing.T) error {
sourceCols := []connection.ColumnDefinition{
{Name: "id", Type: "bigint", Nullable: "NO", Key: "PRI"},
{Name: "tenant_id", Type: "bigint", Nullable: "NO", Key: "PRI"},
}
useSyncDatabaseFactorySequence(t,
syncDatabaseFactoryStep{db: &fakeMigrationDB{
columns: map[string][]connection.ColumnDefinition{
"shop.users": sourceCols,
},
}},
syncDatabaseFactoryStep{db: &fakeMigrationDB{
columns: map[string][]connection.ColumnDefinition{
"app.users": sourceCols,
},
}},
)
_, err := NewSyncEngine(Reporter{}).Preview(basePreviewI18nConfig(), tableName, 20)
return err
},
wantKey: "data_sync.backend.error.preview_composite_pk_unsupported",
wantParams: map[string]any{
"columns": "id,tenant_id",
},
},
}
for _, tc := range cases {

89
internal/sync/row_diff.go Normal file
View File

@@ -0,0 +1,89 @@
package sync
import (
"GoNavi-Wails/internal/connection"
"encoding/json"
"fmt"
)
// diffRowsByKeyColumns compares rows by the complete, ordered key tuple.
// Updates and deletes retain every key component for the database driver.
func diffRowsByKeyColumns(keyColumns []string, sourceRows, targetRows []map[string]interface{}) ([]map[string]interface{}, []connection.UpdateRow, []map[string]interface{}, int) {
targetMap := make(map[string]map[string]interface{}, len(targetRows))
for _, row := range targetRows {
key, ok := syncRowKey(row, keyColumns)
if !ok {
continue
}
targetMap[key] = row
}
sourceKeySet := make(map[string]struct{}, len(sourceRows))
inserts := make([]map[string]interface{}, 0)
updates := make([]connection.UpdateRow, 0)
same := 0
for _, sourceRow := range sourceRows {
key, ok := syncRowKey(sourceRow, keyColumns)
if !ok {
continue
}
sourceKeySet[key] = struct{}{}
if targetRow, exists := targetMap[key]; exists {
changes := make(map[string]interface{})
for column, value := range sourceRow {
if fmt.Sprintf("%v", value) != fmt.Sprintf("%v", targetRow[column]) {
changes[column] = value
}
}
if len(changes) == 0 {
same++
continue
}
updates = append(updates, connection.UpdateRow{
Keys: syncRowKeys(sourceRow, keyColumns),
Values: changes,
})
continue
}
inserts = append(inserts, sourceRow)
}
deletes := make([]map[string]interface{}, 0)
for key, row := range targetMap {
if _, exists := sourceKeySet[key]; exists {
continue
}
deletes = append(deletes, syncRowKeys(row, keyColumns))
}
return inserts, updates, deletes, same
}
func syncRowKeys(row map[string]interface{}, keyColumns []string) map[string]interface{} {
keys := make(map[string]interface{}, len(keyColumns))
for _, column := range keyColumns {
keys[column] = row[column]
}
return keys
}
func syncRowKey(row map[string]interface{}, keyColumns []string) (string, bool) {
if len(keyColumns) == 0 {
return "", false
}
values := make([]interface{}, 0, len(keyColumns))
for _, column := range keyColumns {
value, ok := row[column]
if !ok || value == nil {
return "", false
}
if bytes, ok := value.([]byte); ok {
value = string(bytes)
}
values = append(values, value)
}
encoded, err := json.Marshal(values)
if err != nil {
return "", false
}
return string(encoded), true
}

49
internal/sync/row_selection.go
View File

@@ -6,25 +6,32 @@ import (
)
func filterRowsByPKSelection(pkCol string, rows []map[string]interface{}, enabled bool, selectedPKs []string) []map[string]interface{} {
return filterRowsByKeySelection([]string{pkCol}, rows, enabled, selectedPKs)
}
func filterRowsByKeySelection(keyColumns []string, rows []map[string]interface{}, enabled bool, selectedKeys []string) []map[string]interface{} {
if !enabled {
return nil
}
if len(rows) == 0 {
return rows
}
if len(selectedPKs) == 0 {
if len(selectedKeys) == 0 {
return rows
}
set := make(map[string]struct{}, len(selectedPKs))
for _, pk := range selectedPKs {
set[pk] = struct{}{}
set := make(map[string]struct{}, len(selectedKeys))
for _, key := range selectedKeys {
set[key] = struct{}{}
}
out := make([]map[string]interface{}, 0, len(rows))
for _, row := range rows {
pkStr := fmt.Sprintf("%v", row[pkCol])
if _, ok := set[pkStr]; ok {
key, ok := selectionRowKey(row, keyColumns)
if !ok {
continue
}
if _, selected := set[key]; selected {
out = append(out, row)
}
}
@@ -32,27 +39,45 @@ func filterRowsByPKSelection(pkCol string, rows []map[string]interface{}, enable
}
func filterUpdatesByPKSelection(pkCol string, updates []connection.UpdateRow, enabled bool, selectedPKs []string) []connection.UpdateRow {
return filterUpdatesByKeySelection([]string{pkCol}, updates, enabled, selectedPKs)
}
func filterUpdatesByKeySelection(keyColumns []string, updates []connection.UpdateRow, enabled bool, selectedKeys []string) []connection.UpdateRow {
if !enabled {
return nil
}
if len(updates) == 0 {
return updates
}
if len(selectedPKs) == 0 {
if len(selectedKeys) == 0 {
return updates
}
set := make(map[string]struct{}, len(selectedPKs))
for _, pk := range selectedPKs {
set[pk] = struct{}{}
set := make(map[string]struct{}, len(selectedKeys))
for _, key := range selectedKeys {
set[key] = struct{}{}
}
out := make([]connection.UpdateRow, 0, len(updates))
for _, u := range updates {
pkStr := fmt.Sprintf("%v", u.Keys[pkCol])
if _, ok := set[pkStr]; ok {
key, ok := selectionRowKey(u.Keys, keyColumns)
if !ok {
continue
}
if _, selected := set[key]; selected {
out = append(out, u)
}
}
return out
}
func selectionRowKey(row map[string]interface{}, keyColumns []string) (string, bool) {
if len(keyColumns) == 1 {
value, ok := row[keyColumns[0]]
if !ok || value == nil {
return "", false
}
return fmt.Sprintf("%v", value), true
}
return syncRowKey(row, keyColumns)
}

3
internal/sync/source_query_paging.go
View File

@@ -12,6 +12,9 @@ func (s *SyncEngine) tryApplySourceQueryInPages(config SyncConfig, res *SyncResu
if hasExplicitSyncMappings(config) {
return false, pagedDiffCounts{}, nil
}
if len(ctx.PKColumns) > 1 {
return false, pagedDiffCounts{}, nil
}
sourceType := resolveMigrationDBType(config.SourceConfig)
if !supportsPagedSourceQuery(sourceType) || !supportsPagedDiffPKLookup(ctx.TargetType) {
return false, pagedDiffCounts{}, nil

145
internal/sync/source_query_sync.go
View File

@@ -16,6 +16,7 @@ type sourceQuerySyncContext struct {
TargetType string
TargetCols []connection.ColumnDefinition
PKColumn string
PKColumns []string
SourceRows []map[string]interface{}
TargetRows []map[string]interface{}
SkippedRows int
@@ -90,7 +91,7 @@ func resolveTargetQueryTable(config SyncConfig, tableName string) (string, strin
return targetType, targetSchema, targetTable, targetQueryTable
}
func resolveSinglePKColumn(cols []connection.ColumnDefinition) (string, error) {
func resolvePKColumns(cols []connection.ColumnDefinition) ([]string, error) {
pkCols := make([]string, 0, 2)
for _, col := range cols {
if col.Key == "PRI" || col.Key == "PK" {
@@ -98,14 +99,9 @@ func resolveSinglePKColumn(cols []connection.ColumnDefinition) (string, error) {
}
}
if len(pkCols) == 0 {
return "", syncTextError("data_sync.backend.error.target_pk_required_for_query_diff", nil)
return nil, syncTextError("data_sync.backend.error.target_pk_required_for_query_diff", nil)
}
if len(pkCols) > 1 {
return "", syncTextError("data_sync.backend.error.target_composite_pk_query_diff_unsupported", map[string]any{
"columns": strings.Join(pkCols, ","),
})
}
return pkCols[0], nil
return pkCols, nil
}
func loadSourceQuerySyncContext(config SyncConfig, sourceDB db.Database, targetDB db.Database, needSourceRows bool, needTargetRows bool, requirePK bool) (sourceQuerySyncContext, error) {
@@ -161,22 +157,41 @@ func loadSourceQuerySyncContextWithContext(runCtx context.Context, config SyncCo
}
if requirePK {
pkColumn, err := resolveSinglePKColumn(targetCols)
pkColumns, err := resolvePKColumns(targetCols)
if err != nil {
return sourceQuerySyncContext{}, err
}
if mapping, mapped, mappingErr := sourceQueryMapping(config); mappingErr != nil {
return sourceQuerySyncContext{}, mappingErr
} else if mapped {
if len(mapping.KeyColumns) != 1 {
return sourceQuerySyncContext{}, fmt.Errorf("SQL 结果 insert_update 当前要求恰好一个稳定 keyColumns")
if len(mapping.KeyColumns) != len(pkColumns) {
return sourceQuerySyncContext{}, fmt.Errorf("SQL 结果 insert_update 的 keyColumns 必须与目标表主键列数量一致")
}
mappedKey, ok := projection.TargetColumn(mapping.KeyColumns[0])
if !ok || !strings.EqualFold(strings.TrimSpace(mappedKey), strings.TrimSpace(pkColumn)) {
return sourceQuerySyncContext{}, fmt.Errorf("SQL 结果映射后的稳定 key %s 必须与目标表主键 %s 一致", mappedKey, pkColumn)
targetPKSet := make(map[string]string, len(pkColumns))
for _, targetKey := range pkColumns {
targetPKSet[strings.ToLower(strings.TrimSpace(targetKey))] = targetKey
}
mappedPKSet := make(map[string]struct{}, len(mapping.KeyColumns))
for _, sourceKey := range mapping.KeyColumns {
mappedKey, ok := projection.TargetColumn(sourceKey)
mappedLower := strings.ToLower(strings.TrimSpace(mappedKey))
if !ok || mappedLower == "" {
return sourceQuerySyncContext{}, fmt.Errorf("SQL 结果映射后的稳定 key %s 必须与目标表主键一致", mappedKey)
}
if _, exists := targetPKSet[mappedLower]; !exists {
if len(pkColumns) == 1 {
return sourceQuerySyncContext{}, fmt.Errorf("SQL 结果映射后的稳定 key %s 必须与目标表主键 %s 一致", mappedKey, pkColumns[0])
}
return sourceQuerySyncContext{}, fmt.Errorf("SQL 结果映射后的稳定 key %s 必须属于目标表主键 (%s)", mappedKey, strings.Join(pkColumns, ","))
}
if _, duplicate := mappedPKSet[mappedLower]; duplicate {
return sourceQuerySyncContext{}, fmt.Errorf("SQL 结果映射后的稳定 key 重复指向目标主键 %s", targetPKSet[mappedLower])
}
mappedPKSet[mappedLower] = struct{}{}
}
}
ctx.PKColumn = pkColumn
ctx.PKColumns = pkColumns
ctx.PKColumn = strings.Join(pkColumns, ",")
}
if needTargetRows {
@@ -201,62 +216,6 @@ func projectionForSourceQuery(config SyncConfig) (*CompiledProjection, error) {
return CompileProjection(mapping)
}
func diffRowsByPK(pkCol string, sourceRows, targetRows []map[string]interface{}) ([]map[string]interface{}, []connection.UpdateRow, []map[string]interface{}, int) {
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))
inserts := make([]map[string]interface{}, 0)
updates := make([]connection.UpdateRow, 0)
same := 0
for _, sourceRow := range sourceRows {
if sourceRow[pkCol] == nil {
continue
}
pkVal := strings.TrimSpace(fmt.Sprintf("%v", sourceRow[pkCol]))
if pkVal == "" || pkVal == "<nil>" {
continue
}
sourcePKSet[pkVal] = struct{}{}
if targetRow, exists := targetMap[pkVal]; exists {
changes := make(map[string]interface{})
for key, value := range sourceRow {
if fmt.Sprintf("%v", value) != fmt.Sprintf("%v", targetRow[key]) {
changes[key] = value
}
}
if len(changes) == 0 {
same++
continue
}
updates = append(updates, connection.UpdateRow{
Keys: map[string]interface{}{pkCol: sourceRow[pkCol]},
Values: changes,
})
continue
}
inserts = append(inserts, sourceRow)
}
deletes := make([]map[string]interface{}, 0)
for pkVal, row := range targetMap {
if _, exists := sourcePKSet[pkVal]; exists {
continue
}
deletes = append(deletes, map[string]interface{}{pkCol: row[pkCol]})
}
return inserts, updates, deletes, same
}
func buildTargetColumnSet(cols []connection.ColumnDefinition) map[string]struct{} {
targetColSet := make(map[string]struct{}, len(cols))
for _, col := range cols {
@@ -328,7 +287,7 @@ func (s *SyncEngine) analyzeSourceQuery(config SyncConfig) SyncAnalyzeResult {
handled := false
counts := pagedDiffCounts{}
var scanErr error
if !hasExplicitSyncMappings(config) {
if !hasExplicitSyncMappings(config) && len(ctx.PKColumns) == 1 {
handled, counts, scanErr = scanSourceQueryDiffInPages(sourceDB, targetDB, sourceType, ctx.TargetType, strings.TrimSpace(config.SourceQuery), ctx.TargetQueryTable, ctx.TargetCols, ctx.PKColumn, true, nil)
}
if handled {
@@ -362,7 +321,7 @@ func (s *SyncEngine) analyzeSourceQuery(config SyncConfig) SyncAnalyzeResult {
return result
}
inserts, updates, deletes, same := diffRowsByPK(ctx.PKColumn, ctx.SourceRows, ctx.TargetRows)
inserts, updates, deletes, same := diffRowsByKeyColumns(ctx.PKColumns, ctx.SourceRows, ctx.TargetRows)
summary.CanSync = true
summary.PKColumn = ctx.PKColumn
summary.Inserts = len(inserts)
@@ -407,6 +366,7 @@ func (s *SyncEngine) previewSourceQuery(config SyncConfig, limit int) (TableDiff
out := TableDiffPreview{
Table: ctx.TableName,
PKColumn: ctx.PKColumn,
PKColumns: append([]string(nil), ctx.PKColumns...),
ColumnTypes: make(map[string]string, len(ctx.TargetCols)),
SchemaSummary: previewSummary,
Inserts: make([]PreviewRow, 0, limit),
@@ -424,7 +384,7 @@ func (s *SyncEngine) previewSourceQuery(config SyncConfig, limit int) (TableDiff
handled := false
var scanErr error
if !hasExplicitSyncMappings(config) {
if !hasExplicitSyncMappings(config) && len(ctx.PKColumns) == 1 {
handled, _, scanErr = scanSourceQueryDiffInPages(sourceDB, targetDB, sourceType, ctx.TargetType, strings.TrimSpace(config.SourceQuery), ctx.TargetQueryTable, ctx.TargetCols, ctx.PKColumn, true, func(page pagedDiffPage) error {
out.TotalInserts += len(page.Inserts)
out.TotalUpdates += len(page.Updates)
@@ -477,10 +437,11 @@ func (s *SyncEngine) previewSourceQuery(config SyncConfig, limit int) (TableDiff
return TableDiffPreview{}, err
}
inserts, updates, deletes, _ := diffRowsByPK(ctx.PKColumn, ctx.SourceRows, ctx.TargetRows)
inserts, updates, deletes, _ := diffRowsByKeyColumns(ctx.PKColumns, ctx.SourceRows, ctx.TargetRows)
out = TableDiffPreview{
Table: ctx.TableName,
PKColumn: ctx.PKColumn,
PKColumns: append([]string(nil), ctx.PKColumns...),
ColumnTypes: make(map[string]string, len(ctx.TargetCols)),
SchemaSummary: previewSummary,
TotalInserts: len(inserts),
@@ -503,24 +464,32 @@ func (s *SyncEngine) previewSourceQuery(config SyncConfig, limit int) (TableDiff
if idx >= limit {
break
}
pk := strings.TrimSpace(fmt.Sprintf("%v", row[ctx.PKColumn]))
out.Inserts = append(out.Inserts, PreviewRow{PK: pk, Row: row})
key, ok := selectionRowKey(row, ctx.PKColumns)
if ok {
out.Inserts = append(out.Inserts, PreviewRow{PK: key, Row: row})
}
}
for idx, update := range updates {
if idx >= limit {
break
}
pk := strings.TrimSpace(fmt.Sprintf("%v", update.Keys[ctx.PKColumn]))
identityKey, ok := syncRowKey(update.Keys, ctx.PKColumns)
if !ok {
continue
}
displayKey, _ := selectionRowKey(update.Keys, ctx.PKColumns)
targetRow := map[string]interface{}{}
for _, row := range ctx.TargetRows {
if fmt.Sprintf("%v", row[ctx.PKColumn]) == fmt.Sprintf("%v", update.Keys[ctx.PKColumn]) {
rowKey, rowOK := syncRowKey(row, ctx.PKColumns)
if rowOK && rowKey == identityKey {
targetRow = row
break
}
}
sourceRow := map[string]interface{}{}
for _, row := range ctx.SourceRows {
if fmt.Sprintf("%v", row[ctx.PKColumn]) == fmt.Sprintf("%v", update.Keys[ctx.PKColumn]) {
rowKey, rowOK := syncRowKey(row, ctx.PKColumns)
if rowOK && rowKey == identityKey {
sourceRow = row
break
}
@@ -530,7 +499,7 @@ func (s *SyncEngine) previewSourceQuery(config SyncConfig, limit int) (TableDiff
changedColumns = append(changedColumns, column)
}
out.Updates = append(out.Updates, PreviewUpdateRow{
PK: pk,
PK: displayKey,
ChangedColumns: changedColumns,
Source: sourceRow,
Target: targetRow,
@@ -540,8 +509,10 @@ func (s *SyncEngine) previewSourceQuery(config SyncConfig, limit int) (TableDiff
if idx >= limit {
break
}
pk := strings.TrimSpace(fmt.Sprintf("%v", row[ctx.PKColumn]))
out.Deletes = append(out.Deletes, PreviewRow{PK: pk, Row: row})
key, ok := selectionRowKey(row, ctx.PKColumns)
if ok {
out.Deletes = append(out.Deletes, PreviewRow{PK: key, Row: row})
}
}
return out, nil
}
@@ -648,10 +619,10 @@ func (s *SyncEngine) runSourceQuerySync(config SyncConfig) SyncResult {
}
result.RowsSkipped += ctx.SkippedRows
if tableMode == "insert_update" {
inserts, updates, deletes, _ = diffRowsByPK(ctx.PKColumn, ctx.SourceRows, ctx.TargetRows)
inserts = filterRowsByPKSelection(ctx.PKColumn, inserts, opts.Insert, opts.SelectedInsertPKs)
updates = filterUpdatesByPKSelection(ctx.PKColumn, updates, opts.Update, opts.SelectedUpdatePKs)
deletes = filterRowsByPKSelection(ctx.PKColumn, deletes, opts.Delete, opts.SelectedDeletePKs)
inserts, updates, deletes, _ = diffRowsByKeyColumns(ctx.PKColumns, ctx.SourceRows, ctx.TargetRows)
inserts = filterRowsByKeySelection(ctx.PKColumns, inserts, opts.Insert, opts.SelectedInsertPKs)
updates = filterUpdatesByKeySelection(ctx.PKColumns, updates, opts.Update, opts.SelectedUpdatePKs)
deletes = filterRowsByKeySelection(ctx.PKColumns, deletes, opts.Delete, opts.SelectedDeletePKs)
} else {
inserts = ctx.SourceRows
if !opts.Insert {

106
internal/sync/source_query_sync_test.go
View File

@@ -11,6 +11,45 @@ import (
"testing"
)
func TestDiffRowsByKeyColumnsUsesCompleteCompositeKey(t *testing.T) {
source := []map[string]interface{}{
{"tenant_id": int64(1), "order_id": int64(7), "status": "paid"},
{"tenant_id": int64(2), "order_id": int64(7), "status": "new"},
}
target := []map[string]interface{}{
{"tenant_id": int64(1), "order_id": int64(7), "status": "new"},
{"tenant_id": int64(3), "order_id": int64(7), "status": "old"},
}
inserts, updates, deletes, same := diffRowsByKeyColumns([]string{"tenant_id", "order_id"}, source, target)
if len(inserts) != 1 || len(updates) != 1 || len(deletes) != 1 || same != 0 {
t.Fatalf("unexpected composite diff: inserts=%#v updates=%#v deletes=%#v same=%d", inserts, updates, deletes, same)
}
if got := updates[0].Keys; len(got) != 2 || got["tenant_id"] != int64(1) || got["order_id"] != int64(7) {
t.Fatalf("update keys = %#v, want complete composite key", got)
}
if got := deletes[0]; len(got) != 2 || got["tenant_id"] != int64(3) || got["order_id"] != int64(7) {
t.Fatalf("delete keys = %#v, want complete composite key", got)
}
}
func TestDiffRowsByKeyColumnsPreservesWhitespaceAndEmptyStrings(t *testing.T) {
source := []map[string]interface{}{
{"tenant_id": "", "order_id": "7", "status": "insert"},
{"tenant_id": " tenant", "order_id": "8", "status": "left-space"},
}
target := []map[string]interface{}{
{"tenant_id": "tenant", "order_id": "8", "status": "no-space"},
}
inserts, updates, deletes, same := diffRowsByKeyColumns([]string{"tenant_id", "order_id"}, source, target)
if len(inserts) != 2 || len(updates) != 0 || len(deletes) != 1 || same != 0 {
t.Fatalf("whitespace-sensitive diff: inserts=%#v updates=%#v deletes=%#v same=%d", inserts, updates, deletes, same)
}
if key, ok := syncRowKey(source[0], []string{"tenant_id", "order_id"}); !ok || key != `["","7"]` {
t.Fatalf("empty-string composite key = %q, %v", key, ok)
}
}
type fakeQuerySyncTargetDB struct {
fakeMigrationDB
appliedTable string
@@ -237,14 +276,14 @@ func TestValidateSourceQuerySyncConfigUsesCurrentLanguageForValidationErrors(t *
}
}
func TestResolveSinglePKColumnUsesCurrentLanguageForQueryDiffErrors(t *testing.T) {
func TestResolvePKColumnsUsesCurrentLanguageForQueryDiffErrors(t *testing.T) {
SetBackendLanguage(i18n.LanguageEnUS)
t.Cleanup(func() {
SetBackendLanguage(i18n.LanguageZhCN)
})
t.Run("target primary key required", func(t *testing.T) {
_, err := resolveSinglePKColumn([]connection.ColumnDefinition{
_, err := resolvePKColumns([]connection.ColumnDefinition{
{Name: "id", Type: "bigint", Nullable: "NO"},
})
if err == nil {
@@ -258,22 +297,18 @@ func TestResolveSinglePKColumnUsesCurrentLanguageForQueryDiffErrors(t *testing.T
assertNoLegacySourceQueryChinese(t, err.Error())
})
t.Run("composite primary key unsupported", func(t *testing.T) {
_, err := resolveSinglePKColumn([]connection.ColumnDefinition{
t.Run("composite primary key preserves column order", func(t *testing.T) {
keys, err := resolvePKColumns([]connection.ColumnDefinition{
{Name: "id", Type: "bigint", Nullable: "NO", Key: "PRI"},
{Name: "tenant_id", Type: "bigint", Nullable: "NO", Key: "PRI"},
})
if err == nil {
t.Fatal("expected composite primary key error")
if err != nil {
t.Fatalf("resolvePKColumns() error = %v", err)
}
want := localizedSyncTestText(t, i18n.LanguageEnUS, "data_sync.backend.error.target_composite_pk_query_diff_unsupported", map[string]any{
"columns": "id,tenant_id",
})
if err.Error() != want {
t.Fatalf("expected localized composite PK message %q, got %q", want, err.Error())
want := []string{"id", "tenant_id"}
if !reflect.DeepEqual(keys, want) {
t.Fatalf("resolvePKColumns() = %#v, want %#v", keys, want)
}
assertNoLegacySourceQueryChinese(t, err.Error())
})
}
@@ -806,6 +841,51 @@ func TestPreviewSourceQueryUsesCurrentLanguageForSchemaSummary(t *testing.T) {
assertNoLegacySourceQueryChinese(t, preview.SchemaSummary)
}
func TestPreviewSourceQueryFallbackKeepsSingleKeyDisplayAndRows(t *testing.T) {
const sourceSQL = "SELECT id, name FROM active_users"
sourceDB := &fakeMigrationDB{queryData: map[string][]map[string]interface{}{
sourceSQL: {{"id": int64(1), "name": "new"}},
}}
targetDB := &fakeQuerySyncTargetDB{fakeMigrationDB: fakeMigrationDB{
columns: map[string][]connection.ColumnDefinition{
"app.users": {
{Name: "id", Type: "bigint", Nullable: "NO", Key: "PRI"},
{Name: "name", Type: "varchar(64)", Nullable: "YES"},
},
},
queryData: map[string][]map[string]interface{}{
"SELECT * FROM `app`.`users`": {{"id": int64(1), "name": "old"}},
},
}}
useSyncDatabaseFactorySequence(t,
syncDatabaseFactoryStep{db: sourceDB},
syncDatabaseFactoryStep{db: targetDB},
)
preview, err := NewSyncEngine(Reporter{}).previewSourceQuery(SyncConfig{
SourceConfig: connection.ConnectionConfig{Type: "mysql", Database: "source_db"},
TargetConfig: connection.ConnectionConfig{Type: "mysql", Database: "app"},
SourceQuery: sourceSQL,
Content: "data",
Mode: "insert_update",
Mappings: []SyncObjectMapping{{
Source: SyncObjectRef{Name: "active_users"},
Target: SyncObjectRef{Schema: "app", Name: "users"},
KeyColumns: []string{"id"},
Columns: []SyncColumnMapping{
{Source: "id", Target: "id"},
{Source: "name", Target: "name"},
},
}},
}, 20)
if err != nil {
t.Fatalf("previewSourceQuery() error = %v", err)
}
if len(preview.Updates) != 1 || preview.Updates[0].PK != "1" || preview.Updates[0].Source["name"] != "new" || preview.Updates[0].Target["name"] != "old" {
t.Fatalf("preview updates = %#v", preview.Updates)
}
}
func TestRunSourceQuerySyncUsesCurrentLanguageForStartProgressAndSourceLog(t *testing.T) {
SetBackendLanguage(i18n.LanguageEnUS)
t.Cleanup(func() {

124
internal/sync/sync_engine.go
View File

@@ -404,7 +404,7 @@ func (s *SyncEngine) runSync(config SyncConfig) SyncResult {
}
requirePK := tableMode == "insert_update" && plan.TargetTableExists
pkCol := ""
targetPKCol := ""
targetPKCols := []string(nil)
if requirePK {
if len(pkCols) == 0 {
message := fmt.Sprintf("表 %s 未找到主键，当前模式需要差异对比，已跳过", tableName)
@@ -412,23 +412,19 @@ func (s *SyncEngine) runSync(config SyncConfig) SyncResult {
markTableFailure(message)
return
}
if len(pkCols) > 1 {
message := fmt.Sprintf("表 %s 为复合主键（%s），当前暂不支持差异同步", tableName, strings.Join(pkCols, ","))
s.appendLog(config.JobID, &result, "warn", message)
markTableFailure(message)
return
}
pkCol = pkCols[0]
targetPKCol = pkCol
targetPKCols = append([]string(nil), pkCols...)
if hasExplicitSyncMappings(config) {
mappedPK, ok := projection.TargetColumn(pkCol)
if !ok || strings.TrimSpace(mappedPK) == "" {
message := fmt.Sprintf("表 %s 的主键字段 %s 未映射到目标字段，无法执行差异同步", tableName, pkCol)
s.appendLog(config.JobID, &result, "warn", message)
markTableFailure(message)
return
for index, sourcePKCol := range pkCols {
mappedPK, ok := projection.TargetColumn(sourcePKCol)
if !ok || strings.TrimSpace(mappedPK) == "" {
message := fmt.Sprintf("表 %s 的主键字段 %s 未映射到目标字段，无法执行差异同步", tableName, sourcePKCol)
s.appendLog(config.JobID, &result, "warn", message)
markTableFailure(message)
return
}
targetPKCols[index] = mappedPK
}
targetPKCol = mappedPK
}
}
@@ -459,33 +455,35 @@ func (s *SyncEngine) runSync(config SyncConfig) SyncResult {
return
}
if handled, counts, err := s.tryApplyDiffInPages(config, &result, i, totalTables, tableName, sourceDB, targetDB, plan, cols, targetCols, opts, sourceType, targetType, applyTableName, pkCol); handled {
result.RowsInserted += counts.Inserts
result.RowsUpdated += counts.Updates
result.RowsDeleted += counts.Deletes
if err != nil {
logger.Error(err, "分页差异同步失败：表=%s", tableName)
message := fmt.Sprintf("分页差异同步失败: %v", err)
s.appendLog(config.JobID, &result, "error", " -> "+message)
markTableFailure(message)
return
}
if counts.Inserts > 0 || counts.Updates > 0 || counts.Deletes > 0 {
s.appendLog(config.JobID, &result, "info", fmt.Sprintf(" -> 分页差异同步完成：插入=%d 更新=%d 删除=%d", counts.Inserts, counts.Updates, counts.Deletes))
} else {
s.appendLog(config.JobID, &result, "info", " -> 数据一致，无需变更.")
}
if schemaChangesAllowed && len(plan.PostDataSQL) > 0 {
s.progress(config.JobID, i, totalTables, tableName, "创建索引")
if err := executeSyncSQLStatementsContext(s.context(), targetDB, plan.PostDataSQL); err != nil {
message := fmt.Sprintf("创建索引失败：表=%s 错误=%v", tableName, err)
s.appendLog(config.JobID, &result, "error", message)
if len(targetPKCols) <= 1 {
if handled, counts, err := s.tryApplyDiffInPages(config, &result, i, totalTables, tableName, sourceDB, targetDB, plan, cols, targetCols, opts, sourceType, targetType, applyTableName, pkCol); handled {
result.RowsInserted += counts.Inserts
result.RowsUpdated += counts.Updates
result.RowsDeleted += counts.Deletes
if err != nil {
logger.Error(err, "分页差异同步失败：表=%s", tableName)
message := fmt.Sprintf("分页差异同步失败: %v", err)
s.appendLog(config.JobID, &result, "error", " -> "+message)
markTableFailure(message)
return
}
if counts.Inserts > 0 || counts.Updates > 0 || counts.Deletes > 0 {
s.appendLog(config.JobID, &result, "info", fmt.Sprintf(" -> 分页差异同步完成：插入=%d 更新=%d 删除=%d", counts.Inserts, counts.Updates, counts.Deletes))
} else {
s.appendLog(config.JobID, &result, "info", " -> 数据一致，无需变更.")
}
if schemaChangesAllowed && len(plan.PostDataSQL) > 0 {
s.progress(config.JobID, i, totalTables, tableName, "创建索引")
if err := executeSyncSQLStatementsContext(s.context(), targetDB, plan.PostDataSQL); err != nil {
message := fmt.Sprintf("创建索引失败：表=%s 错误=%v", tableName, err)
s.appendLog(config.JobID, &result, "error", message)
markTableFailure(message)
return
}
}
tableCompleted = true
return
}
tableCompleted = true
return
}
s.progress(config.JobID, i, totalTables, tableName, "读取源表数据")
@@ -525,52 +523,10 @@ func (s *SyncEngine) runSync(config SyncConfig) SyncResult {
}
s.progress(config.JobID, i, totalTables, tableName, "对比差异")
targetMap := make(map[string]map[string]interface{}, len(targetRows))
for _, row := range targetRows {
if row[targetPKCol] == nil {
continue
}
pkVal := fmt.Sprintf("%v", row[targetPKCol])
if strings.TrimSpace(pkVal) == "" || pkVal == "<nil>" {
continue
}
targetMap[pkVal] = row
}
sourcePKSet := make(map[string]struct{}, len(sourceRows))
for _, sRow := range sourceRows {
if sRow[targetPKCol] == nil {
continue
}
pkVal := fmt.Sprintf("%v", sRow[targetPKCol])
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{}{targetPKCol: sRow[targetPKCol]}, Values: changes})
}
} else {
inserts = append(inserts, sRow)
}
}
if opts.Delete {
for pkStr, row := range targetMap {
if _, ok := sourcePKSet[pkStr]; ok {
continue
}
deletes = append(deletes, map[string]interface{}{targetPKCol: row[targetPKCol]})
}
}
inserts = filterRowsByPKSelection(targetPKCol, inserts, opts.Insert, opts.SelectedInsertPKs)
updates = filterUpdatesByPKSelection(targetPKCol, updates, opts.Update, opts.SelectedUpdatePKs)
deletes = filterRowsByPKSelection(targetPKCol, deletes, opts.Delete, opts.SelectedDeletePKs)
inserts, updates, deletes, _ = diffRowsByKeyColumns(targetPKCols, sourceRows, targetRows)
inserts = filterRowsByKeySelection(targetPKCols, inserts, opts.Insert, opts.SelectedInsertPKs)
updates = filterUpdatesByKeySelection(targetPKCols, updates, opts.Update, opts.SelectedUpdatePKs)
deletes = filterRowsByKeySelection(targetPKCols, deletes, opts.Delete, opts.SelectedDeletePKs)
} else {
inserts = sourceRows
if !opts.Insert {