GithubBackup/MyGoNavi
1
0
Fork 0
mirror of https://github.com/Syngnat/GoNavi.git synced 2026-08-12 09:34:05 +08:00
Code Issues Packages Projects Releases Wiki Activity

🐛 fix(sync): 禁止仅同步数据时修改表结构 (#878)

Browse Source 
Fixes #878
This commit is contained in:
AutumnNazi
2026-08-08 07:21:07 +08:00
parent 14f24925ab
commit 9e661958fe
8 changed files with 236 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -744,6 +744,12 @@ const DataSyncModal: React.FC<{
createIndexes,
]);
useEffect(() => {
if (syncContent === "data" && autoAddColumns) {
setAutoAddColumns(false);
}
}, [syncContent, autoAddColumns]);
useEffect(() => {
if (sourceDatasetMode !== "query") return;
if (workflowType !== "sync") {
@@ -1982,7 +1988,7 @@ const DataSyncModal: React.FC<{
<Checkbox
checked={autoAddColumns}
onChange={(e) => setAutoAddColumns(e.target.checked)}
disabled={isSourceQueryMode}
disabled={isSourceQueryMode || syncContent === "data"}
>
{isSchemaCompareEntry
? tr("data_sync.compare_entry.option.auto_add_columns")

39
frontend/src/components/dataSyncRequest.test.ts
View File

@@ -305,6 +305,44 @@ describe('validateDataSyncExecutionReadiness', () => {
})).toEqual({ ready: false, reason: 'no_effective_operations' });
});
it('does not treat schema differences as effective work in data-only mode', () => {
expect(validateDataSyncExecutionReadiness({
...baseReadinessInput,
syncContent: 'data',
selectedTables: ['users'],
analyzedTables: [{
table: 'users',
canSync: true,
inserts: 0,
updates: 0,
deletes: 0,
schemaDiffCount: 1,
targetTableExists: true,
}],
tableOptions: {
users: { insert: false, update: false, delete: false },
},
})).toEqual({ ready: false, reason: 'no_effective_operations' });
});
it('does not treat a missing target table as effective work in data-only mode', () => {
expect(validateDataSyncExecutionReadiness({
...baseReadinessInput,
syncContent: 'data',
selectedTables: ['users'],
analyzedTables: [{
table: 'users',
canSync: true,
inserts: 0,
schemaDiffCount: 0,
targetTableExists: false,
}],
tableOptions: {
users: { insert: false, update: false, delete: false },
},
})).toEqual({ ready: false, reason: 'no_effective_operations' });
});
it('allows structure work for both mode and an empty auto-created target', () => {
expect(validateDataSyncExecutionReadiness({
...baseReadinessInput,
@@ -324,6 +362,7 @@ describe('validateDataSyncExecutionReadiness', () => {
expect(validateDataSyncExecutionReadiness({
...baseReadinessInput,
syncContent: 'both',
selectedTables: ['users'],
analyzedTables: [{
table: 'users',

6
frontend/src/components/dataSyncRequest.ts
View File

@@ -290,8 +290,10 @@ export const validateDataSyncExecutionReadiness = ({
};
}
const targetNeedsCreation = analysis.targetTableExists === false;
const hasSchemaWork = syncContent !== 'schema'
const schemaChangesAllowed = syncContent !== 'data';
const targetNeedsCreation = schemaChangesAllowed
&& analysis.targetTableExists === false;
const hasSchemaWork = schemaChangesAllowed
&& normalizedCount(analysis.schemaDiffCount) > 0;
let hasDataWork = false;
if (mode === 'full_overwrite') {

41
internal/sync/direct_import_integrity_test.go
View File

@@ -83,6 +83,7 @@ func TestDirectImportFullOverwriteStopsBeforeClearingWhenAutoAddColumnFails(t *t
target := &directImportIntegrityTargetDB{execErr: errors.New("add column rejected")}
config := SyncConfig{
JobID: "direct-import-auto-add",
Content: "both",
Mode: "full_overwrite",
AutoAddColumns: true,
SourceConfig: connection.ConnectionConfig{Type: "mysql", Host: "source", Database: "src"},
@@ -121,3 +122,43 @@ func TestDirectImportFullOverwriteStopsBeforeClearingWhenAutoAddColumnFails(t *t
t.Fatalf("target table was cleared after auto-add failure: %v", target.execs)
}
}
func TestDirectImportDataOnlyRejectsAutoAddColumnsBeforeClearing(t *testing.T) {
engine := &SyncEngine{}
source := &fakeMigrationDB{}
target := &directImportIntegrityTargetDB{}
config := SyncConfig{
JobID: "direct-import-data-only",
Content: "data",
Mode: "full_overwrite",
AutoAddColumns: true,
SourceConfig: connection.ConnectionConfig{Type: "mysql", Host: "source", Database: "src"},
TargetConfig: connection.ConnectionConfig{Type: "mysql", Host: "target", Database: "dst"},
}
plan := SchemaMigrationPlan{
SourceQueryTable: "src.users",
TargetQueryTable: "dst.users",
TargetSchema: "dst",
TargetTable: "users",
TargetTableExists: true,
}
sourceCols := []connection.ColumnDefinition{
{Name: "id", Type: "bigint", Key: "PRI"},
{Name: "name", Type: "varchar(128)"},
}
targetCols := []connection.ColumnDefinition{{Name: "id", Type: "bigint", Key: "PRI"}}
handled, inserted, err := engine.tryApplyDirectImportInPages(
config, &SyncResult{}, 0, 1, "users", source, target, plan,
sourceCols, targetCols, TableOptions{Insert: true}, "mysql", "mysql", "users",
)
if !handled {
t.Fatal("direct import should handle the request")
}
if err == nil || !strings.Contains(err.Error(), "仅同步数据") {
t.Fatalf("expected data-only auto-add rejection, got %v", err)
}
if inserted != 0 || len(target.execs) != 0 || target.clearedTarget() {
t.Fatalf("data-only direct import modified the target: inserted=%d exec=%v", inserted, target.execs)
}
}

3
internal/sync/direct_import_paging.go
View File

@@ -117,6 +117,9 @@ func (s *SyncEngine) prepareDirectImportTargetColumnSet(config SyncConfig, res *
}
if config.AutoAddColumns && supportsAutoAddColumnsForPair(sourceType, targetType) {
if !syncContentAllowsSchemaChanges(config.Content) {
return nil, fmt.Errorf("目标表缺少字段，仅同步数据模式不允许自动补齐：%s", strings.Join(missing, ", "))
}
s.appendLog(config.JobID, res, "warn", fmt.Sprintf(" -> 目标表缺少字段 %d 个，开始自动补齐: %s", len(missing), strings.Join(missing, ", ")))
added := 0
sourceColsByLower := make(map[string]connection.ColumnDefinition, len(sourceCols))

108
internal/sync/fallback_safety_test.go
View File

@@ -73,6 +73,114 @@ func TestRunSyncFallbackFullOverwriteValidatesColumnsBeforeClear(t *testing.T) {
}
}
func TestRunSyncDataOnlyRejectsPlannedSchemaChanges(t *testing.T) {
columns := []connection.ColumnDefinition{
{Name: "id", Type: "bigint", Nullable: "NO", Key: "PRI"},
{Name: "name", Type: "varchar(255)", Nullable: "YES"},
}
sourceDB := &fakeMigrationDB{
columns: map[string][]connection.ColumnDefinition{"source_db.users": columns},
indexes: map[string][]connection.IndexDefinition{"source_db.users": {
{Name: "idx_users_name", ColumnName: "name", NonUnique: 1, SeqInIndex: 1, IndexType: "BTREE"},
}},
}
targetDB := &recordingExecSyncTargetDB{fakeQuerySyncTargetDB: fakeQuerySyncTargetDB{fakeMigrationDB: fakeMigrationDB{
columns: map[string][]connection.ColumnDefinition{"target_db.users": {columns[0]}},
}}}
useSyncDatabaseFactorySequence(t,
syncDatabaseFactoryStep{db: sourceDB},
syncDatabaseFactoryStep{db: targetDB},
)
result := NewSyncEngine(Reporter{}).RunSync(SyncConfig{
SourceConfig: connection.ConnectionConfig{Type: "mysql", Database: "source_db"},
TargetConfig: connection.ConnectionConfig{Type: "mysql", Database: "target_db"},
SourceDatabase: "source_db",
TargetDatabase: "target_db",
Tables: []string{"users"},
Content: "data",
Mode: "insert_update",
AutoAddColumns: true,
CreateIndexes: true,
TableOptions: map[string]TableOptions{"users": {}},
})
if result.Success || !strings.Contains(result.Message, "仅同步数据") {
t.Fatalf("expected data-only schema-change rejection, got %+v", result)
}
if len(targetDB.execLog) != 0 {
t.Fatalf("data-only sync must not execute schema SQL: %v", targetDB.execLog)
}
}
func TestRunSyncDataOnlyRejectsMissingTargetAutoCreate(t *testing.T) {
columns := []connection.ColumnDefinition{{Name: "id", Type: "bigint", Nullable: "NO", Key: "PRI"}}
sourceDB := &fakeMigrationDB{columns: map[string][]connection.ColumnDefinition{"source_db.users": columns}}
targetDB := &recordingNonBatchSyncTargetDB{}
useSyncDatabaseFactorySequence(t,
syncDatabaseFactoryStep{db: sourceDB},
syncDatabaseFactoryStep{db: targetDB},
)
result := NewSyncEngine(Reporter{}).RunSync(SyncConfig{
SourceConfig: connection.ConnectionConfig{Type: "mysql", Database: "source_db"},
TargetConfig: connection.ConnectionConfig{Type: "mysql", Database: "target_db"},
SourceDatabase: "source_db",
TargetDatabase: "target_db",
Tables: []string{"users"},
Content: "data",
Mode: "insert_update",
TargetTableStrategy: "auto_create_if_missing",
TableOptions: map[string]TableOptions{"users": {}},
})
if result.Success || !strings.Contains(result.Message, "仅同步数据") {
t.Fatalf("expected data-only target-create rejection, got %+v", result)
}
if len(targetDB.execLog) != 0 {
t.Fatalf("data-only sync must not create a missing target: %v", targetDB.execLog)
}
}
func TestRunSyncDataOnlyRejectsPostDataSchemaChanges(t *testing.T) {
columns := []connection.ColumnDefinition{
{Name: "id", Type: "bigint", Nullable: "NO", Key: "PRI"},
{Name: "name", Type: "varchar(255)", Nullable: "YES"},
}
sourceDB := &fakeMigrationDB{
columns: map[string][]connection.ColumnDefinition{"source_db.users": columns},
indexes: map[string][]connection.IndexDefinition{"source_db.users": {
{Name: "idx_users_name", ColumnName: "name", NonUnique: 1, SeqInIndex: 1, IndexType: "BTREE"},
}},
}
targetDB := &recordingNonBatchSyncTargetDB{fakeMigrationDB: fakeMigrationDB{
columns: map[string][]connection.ColumnDefinition{"target_db.users": columns},
}}
useSyncDatabaseFactorySequence(t,
syncDatabaseFactoryStep{db: sourceDB},
syncDatabaseFactoryStep{db: targetDB},
)
result := NewSyncEngine(Reporter{}).RunSync(SyncConfig{
SourceConfig: connection.ConnectionConfig{Type: "mysql", Database: "source_db"},
TargetConfig: connection.ConnectionConfig{Type: "mysql", Database: "target_db"},
SourceDatabase: "source_db",
TargetDatabase: "target_db",
Tables: []string{"users"},
Content: "data",
Mode: "insert_update",
CreateIndexes: true,
TableOptions: map[string]TableOptions{"users": {}},
})
if !result.Success {
t.Fatalf("data-only no-op should remain successful, got %+v", result)
}
if len(targetDB.execLog) != 0 {
t.Fatalf("data-only sync must not create indexes: %v", targetDB.execLog)
}
}
func TestRunSyncFallbackFullOverwriteChecksApplierBeforeClear(t *testing.T) {
columns := []connection.ColumnDefinition{{Name: "id", Type: "bigint", Nullable: "NO", Key: "PRI"}}
sourceDB := &staticRowsSyncSourceDB{

9
internal/sync/migration_runtime_helpers.go
View File

@@ -6,6 +6,15 @@ import (
"strings"
)
func syncContentAllowsSchemaChanges(content string) bool {
switch strings.ToLower(strings.TrimSpace(content)) {
case "schema", "both":
return true
default:
return false
}
}
func supportsAutoAddColumnsForPair(sourceType string, targetType string) bool {
source := normalizeMigrationDBType(sourceType)
target := normalizeMigrationDBType(targetType)

29
internal/sync/sync_engine.go
View File

@@ -116,6 +116,7 @@ func (s *SyncEngine) RunSync(config SyncConfig) SyncResult {
}
defaultMode := normalizeSyncMode(config.Mode)
strategy := normalizeTargetTableStrategy(config.TargetTableStrategy)
schemaChangesAllowed := syncContentAllowsSchemaChanges(config.Content)
contentLabel := "仅同步数据"
if syncSchema && syncData {
@@ -192,6 +193,20 @@ func (s *SyncEngine) RunSync(config SyncConfig) SyncResult {
if strings.TrimSpace(plan.PlannedAction) != "" {
s.appendLog(config.JobID, &result, "info", fmt.Sprintf(" -> %s", plan.PlannedAction))
}
if !schemaChangesAllowed {
if !plan.TargetTableExists && plan.AutoCreate {
message := fmt.Sprintf("表 %s 目标表不存在，仅同步数据模式不允许自动创建目标表", tableName)
s.appendLog(config.JobID, &result, "warn", message)
markTableFailure(message)
return
}
if len(plan.PreDataSQL) > 0 {
message := fmt.Sprintf("表 %s 存在结构差异，仅同步数据模式不允许修改目标表结构", tableName)
s.appendLog(config.JobID, &result, "warn", message)
markTableFailure(message)
return
}
}
if !plan.TargetTableExists && !plan.AutoCreate {
message := fmt.Sprintf("表 %s 目标表不存在，当前策略不允许自动建表，已跳过", tableName)
@@ -281,7 +296,7 @@ func (s *SyncEngine) RunSync(config SyncConfig) SyncResult {
if !hasEffectiveSyncDataOperation(tableMode, opts) {
if tableMode == "insert_update" {
s.appendLog(config.JobID, &result, "info", fmt.Sprintf("表 %s 未选择数据变更，按无变更处理", tableName))
if len(plan.PostDataSQL) > 0 {
if schemaChangesAllowed && len(plan.PostDataSQL) > 0 {
s.progress(config.JobID, i, totalTables, tableName, "创建索引")
if err := executeSQLStatements(targetDB.Exec, plan.PostDataSQL); err != nil {
message := fmt.Sprintf("创建索引失败：表=%s 错误=%v", tableName, err)
@@ -345,7 +360,7 @@ func (s *SyncEngine) RunSync(config SyncConfig) SyncResult {
} else {
s.appendLog(config.JobID, &result, "info", " -> 源表无可导入数据")
}
if len(plan.PostDataSQL) > 0 {
if schemaChangesAllowed && len(plan.PostDataSQL) > 0 {
s.progress(config.JobID, i, totalTables, tableName, "创建索引")
if err := executeSQLStatements(targetDB.Exec, plan.PostDataSQL); err != nil {
message := fmt.Sprintf("创建索引失败：表=%s 错误=%v", tableName, err)
@@ -374,7 +389,7 @@ func (s *SyncEngine) RunSync(config SyncConfig) SyncResult {
} else {
s.appendLog(config.JobID, &result, "info", " -> 数据一致，无需变更.")
}
if len(plan.PostDataSQL) > 0 {
if schemaChangesAllowed && len(plan.PostDataSQL) > 0 {
s.progress(config.JobID, i, totalTables, tableName, "创建索引")
if err := executeSQLStatements(targetDB.Exec, plan.PostDataSQL); err != nil {
message := fmt.Sprintf("创建索引失败：表=%s 错误=%v", tableName, err)
@@ -496,6 +511,12 @@ func (s *SyncEngine) RunSync(config SyncConfig) SyncResult {
}
sort.Strings(missing)
if len(missing) > 0 {
if config.AutoAddColumns && !schemaChangesAllowed {
message := fmt.Sprintf("目标表缺少字段，仅同步数据模式不允许自动补齐：%s", strings.Join(missing, ", "))
s.appendLog(config.JobID, &result, "warn", " -> "+message)
markTableFailure(message)
return
}
if config.AutoAddColumns && supportsAutoAddColumnsForPair(sourceType, targetType) {
s.appendLog(config.JobID, &result, "warn", fmt.Sprintf(" -> 目标表缺少字段 %d 个，开始自动补齐: %s", len(missing), strings.Join(missing, ", ")))
added := 0
@@ -581,7 +602,7 @@ func (s *SyncEngine) RunSync(config SyncConfig) SyncResult {
s.appendLog(config.JobID, &result, "info", " -> 数据一致，无需变更.")
}
if len(plan.PostDataSQL) > 0 {
if schemaChangesAllowed && len(plan.PostDataSQL) > 0 {
s.progress(config.JobID, i, totalTables, tableName, "创建索引")
if err := executeSQLStatements(targetDB.Exec, plan.PostDataSQL); err != nil {
message := fmt.Sprintf("创建索引失败：表=%s 错误=%v", tableName, err)