diff --git a/frontend/src/components/DataSyncModal.tsx b/frontend/src/components/DataSyncModal.tsx index 292e0d6c..9c233ee5 100644 --- a/frontend/src/components/DataSyncModal.tsx +++ b/frontend/src/components/DataSyncModal.tsx @@ -112,6 +112,17 @@ const resolveRedisDbIndex = (raw?: string): number => { return Number.isInteger(value) && value >= 0 && value <= 15 ? value : 0; }; +const isServiceNameBackedSyncConnection = (conn?: SavedConnection): boolean => { + const type = String(conn?.config?.type || '').trim().toLowerCase(); + if (type === 'oracle') return true; + if (type !== 'oceanbase') return false; + const explicitProtocol = String((conn?.config as any)?.oceanBaseProtocol || '').trim().toLowerCase(); + if (explicitProtocol === 'oracle') return true; + const params = new URLSearchParams(String(conn?.config?.connectionParams || '')); + const protocol = String(params.get('protocol') || params.get('tenantMode') || '').trim().toLowerCase(); + return protocol === 'oracle'; +}; + const buildSqlPreview = ( previewData: any, tableName: string, @@ -253,7 +264,7 @@ const DataSyncModal: React.FC<{ open: boolean; onClose: () => void }> = ({ open, const normalizeConnConfig = (conn: SavedConnection, database?: string) => ( buildRpcConnectionConfig(conn.config, { - database: typeof database === 'string' ? database : (conn.config.database || ''), + database: typeof database === 'string' && !isServiceNameBackedSyncConnection(conn) ? database : (conn.config.database || ''), }) ); @@ -473,6 +484,8 @@ const DataSyncModal: React.FC<{ open: boolean; onClose: () => void }> = ({ open, const config = buildDataSyncRequest({ sourceConfig: normalizeConnConfig(sConn, sourceDb), targetConfig: normalizeConnConfig(tConn, targetDb), + sourceDatabase: sourceDb, + targetDatabase: targetDb, selectedTables, sourceDatasetMode, sourceQuery, @@ -528,6 +541,8 @@ const DataSyncModal: React.FC<{ open: boolean; onClose: () => void }> = ({ open, const config = buildDataSyncRequest({ sourceConfig: normalizeConnConfig(sConn, sourceDb), targetConfig: normalizeConnConfig(tConn, targetDb), + sourceDatabase: sourceDb, + targetDatabase: targetDb, selectedTables, sourceDatasetMode, sourceQuery, @@ -600,6 +615,8 @@ const DataSyncModal: React.FC<{ open: boolean; onClose: () => void }> = ({ open, const config = buildDataSyncRequest({ sourceConfig: normalizeConnConfig(sConn, sourceDb), targetConfig: normalizeConnConfig(tConn, targetDb), + sourceDatabase: sourceDb, + targetDatabase: targetDb, selectedTables, sourceDatasetMode, sourceQuery, diff --git a/frontend/src/components/dataSyncRequest.test.ts b/frontend/src/components/dataSyncRequest.test.ts index 3c6c6e1d..94c90bb9 100644 --- a/frontend/src/components/dataSyncRequest.test.ts +++ b/frontend/src/components/dataSyncRequest.test.ts @@ -40,6 +40,8 @@ describe('buildDataSyncRequest', () => { const payload = buildDataSyncRequest({ sourceConfig: { type: 'mysql' }, targetConfig: { type: 'mysql' }, + sourceDatabase: ' app ', + targetDatabase: ' warehouse ', selectedTables: ['users'], sourceDatasetMode: 'query', sourceQuery: ' SELECT id, name FROM active_users ', @@ -61,6 +63,8 @@ describe('buildDataSyncRequest', () => { autoAddColumns: false, targetTableStrategy: 'existing_only', createIndexes: false, + sourceDatabase: 'app', + targetDatabase: 'warehouse', jobId: 'job-1', }); }); diff --git a/frontend/src/components/dataSyncRequest.ts b/frontend/src/components/dataSyncRequest.ts index 6b5cca45..ed953541 100644 --- a/frontend/src/components/dataSyncRequest.ts +++ b/frontend/src/components/dataSyncRequest.ts @@ -6,6 +6,8 @@ type TargetTableStrategy = 'existing_only' | 'auto_create_if_missing' | 'smart'; type BuildDataSyncRequestParams = { sourceConfig: any; targetConfig: any; + sourceDatabase?: string; + targetDatabase?: string; selectedTables: string[]; sourceDatasetMode: SourceDatasetMode; sourceQuery: string; @@ -54,6 +56,8 @@ export const validateDataSyncSelection = ({ export const buildDataSyncRequest = ({ sourceConfig, targetConfig, + sourceDatabase, + targetDatabase, selectedTables, sourceDatasetMode, sourceQuery, @@ -71,6 +75,8 @@ export const buildDataSyncRequest = ({ return { sourceConfig, targetConfig, + sourceDatabase: String(sourceDatabase || '').trim(), + targetDatabase: String(targetDatabase || '').trim(), tables: selectedTables, sourceQuery: isQueryMode ? String(sourceQuery || '').trim() : undefined, content: isQueryMode ? 'data' : syncContent, diff --git a/frontend/wailsjs/go/models.ts b/frontend/wailsjs/go/models.ts index bf9129fb..400de7d8 100755 --- a/frontend/wailsjs/go/models.ts +++ b/frontend/wailsjs/go/models.ts @@ -1257,6 +1257,8 @@ export namespace sync { export class SyncConfig { sourceConfig: connection.ConnectionConfig; targetConfig: connection.ConnectionConfig; + sourceDatabase?: string; + targetDatabase?: string; tables: string[]; sourceQuery?: string; content?: string; @@ -1276,6 +1278,8 @@ export namespace sync { if ('string' === typeof source) source = JSON.parse(source); this.sourceConfig = this.convertValues(source["sourceConfig"], connection.ConnectionConfig); this.targetConfig = this.convertValues(source["targetConfig"], connection.ConnectionConfig); + this.sourceDatabase = source["sourceDatabase"]; + this.targetDatabase = source["targetDatabase"]; this.tables = source["tables"]; this.sourceQuery = source["sourceQuery"]; this.content = source["content"]; diff --git a/internal/app/methods_sync.go b/internal/app/methods_sync.go index e98a9fe6..a6c388ef 100644 --- a/internal/app/methods_sync.go +++ b/internal/app/methods_sync.go @@ -13,19 +13,51 @@ import ( func (a *App) resolveDataSyncConfigSecrets(config sync.SyncConfig) (sync.SyncConfig, error) { resolved := config - sourceConfig, err := a.resolveConnectionSecrets(config.SourceConfig) + sourceConfig, sourceDatabase, err := a.resolveDataSyncEndpointConfig(config.SourceConfig, config.SourceDatabase) if err != nil { return resolved, fmt.Errorf("恢复源数据库连接密文失败: %w", err) } - targetConfig, err := a.resolveConnectionSecrets(config.TargetConfig) + targetConfig, targetDatabase, err := a.resolveDataSyncEndpointConfig(config.TargetConfig, config.TargetDatabase) if err != nil { return resolved, fmt.Errorf("恢复目标数据库连接密文失败: %w", err) } resolved.SourceConfig = sourceConfig resolved.TargetConfig = targetConfig + resolved.SourceDatabase = sourceDatabase + resolved.TargetDatabase = targetDatabase return resolved, nil } +func (a *App) resolveDataSyncEndpointConfig(raw connection.ConnectionConfig, selectedDatabase string) (connection.ConnectionConfig, string, error) { + resolved, err := a.resolveConnectionSecrets(raw) + if err != nil { + return resolved, selectedDatabase, err + } + + if !strings.EqualFold(strings.TrimSpace(raw.Type), "oracle") || strings.TrimSpace(raw.ID) == "" { + return resolved, strings.TrimSpace(selectedDatabase), nil + } + + repo := newSavedConnectionRepository(a.configDir, a.secretStore) + view, findErr := repo.Find(raw.ID) + if findErr != nil { + return resolved, strings.TrimSpace(selectedDatabase), nil + } + + savedServiceName := strings.TrimSpace(view.Config.Database) + if savedServiceName == "" { + return resolved, strings.TrimSpace(selectedDatabase), nil + } + + selected := strings.TrimSpace(selectedDatabase) + incomingDatabase := strings.TrimSpace(raw.Database) + if selected == "" && incomingDatabase != "" && !strings.EqualFold(incomingDatabase, savedServiceName) { + selected = incomingDatabase + } + resolved.Database = savedServiceName + return resolved, selected, nil +} + // DataSync executes a data synchronization task func (a *App) DataSync(config sync.SyncConfig) sync.SyncResult { jobID := strings.TrimSpace(config.JobID) diff --git a/internal/app/methods_sync_test.go b/internal/app/methods_sync_test.go index 6257d758..8eed2810 100644 --- a/internal/app/methods_sync_test.go +++ b/internal/app/methods_sync_test.go @@ -76,3 +76,73 @@ func TestResolveDataSyncConfigSecretsRestoresSavedSourceAndTargetPasswords(t *te t.Fatalf("expected selected databases to be preserved, got source=%q target=%q", resolved.SourceConfig.Database, resolved.TargetConfig.Database) } } + +func TestResolveDataSyncConfigSecretsRestoresOracleServiceNameFromSavedConnection(t *testing.T) { + app := NewAppWithSecretStore(newFakeAppSecretStore()) + app.configDir = t.TempDir() + + _, err := app.SaveConnection(connection.SavedConnectionInput{ + ID: "source-oracle", + Name: "Source Oracle", + Config: connection.ConnectionConfig{ + ID: "source-oracle", + Type: "oracle", + Host: "oracle.local", + Port: 1521, + User: "scott", + Password: "source-secret", + Database: "ORCLPDB1", + }, + }) + if err != nil { + t.Fatalf("SaveConnection source returned error: %v", err) + } + _, err = app.SaveConnection(connection.SavedConnectionInput{ + ID: "target-mysql", + Name: "Target MySQL", + Config: connection.ConnectionConfig{ + ID: "target-mysql", + Type: "mysql", + Host: "mysql.local", + Port: 3306, + User: "root", + Password: "target-secret", + Database: "warehouse", + }, + }) + if err != nil { + t.Fatalf("SaveConnection target returned error: %v", err) + } + + resolved, err := app.resolveDataSyncConfigSecrets(datasync.SyncConfig{ + SourceConfig: connection.ConnectionConfig{ + ID: "source-oracle", + Type: "oracle", + Host: "oracle.local", + Port: 1521, + User: "scott", + Database: "APP_SCHEMA", + }, + TargetConfig: connection.ConnectionConfig{ + ID: "target-mysql", + Type: "mysql", + Host: "mysql.local", + Port: 3306, + User: "root", + Database: "warehouse", + }, + Tables: []string{"APP_SCHEMA.ORDERS"}, + }) + if err != nil { + t.Fatalf("resolveDataSyncConfigSecrets returned error: %v", err) + } + if resolved.SourceConfig.Database != "ORCLPDB1" { + t.Fatalf("expected Oracle service name to be restored, got %q", resolved.SourceConfig.Database) + } + if resolved.SourceDatabase != "APP_SCHEMA" { + t.Fatalf("expected legacy selected schema to move into SourceDatabase, got %q", resolved.SourceDatabase) + } + if resolved.SourceConfig.Password != "source-secret" || resolved.TargetConfig.Password != "target-secret" { + t.Fatalf("expected source and target passwords to be restored") + } +} diff --git a/internal/db/mariadb_impl.go b/internal/db/mariadb_impl.go index 410cbbe2..06f44650 100644 --- a/internal/db/mariadb_impl.go +++ b/internal/db/mariadb_impl.go @@ -214,12 +214,7 @@ func (m *MariaDB) GetTables(dbName string) ([]string, error) { } func (m *MariaDB) GetCreateStatement(dbName, tableName string) (string, error) { - query := fmt.Sprintf("SHOW CREATE TABLE `%s`.`%s`", dbName, tableName) - if dbName == "" { - query = fmt.Sprintf("SHOW CREATE TABLE `%s`", tableName) - } - - data, _, err := m.Query(query) + data, _, err := m.Query(buildMySQLShowCreateTableQuery(dbName, tableName)) if err != nil { return "", err } diff --git a/internal/db/mysql_impl.go b/internal/db/mysql_impl.go index 61e1e1f1..5b715e7a 100644 --- a/internal/db/mysql_impl.go +++ b/internal/db/mysql_impl.go @@ -964,13 +964,56 @@ func (m *MySQLDB) GetTables(dbName string) ([]string, error) { return resolveShardingSphereLogicalTables(tables, m.Query), nil } -func (m *MySQLDB) GetCreateStatement(dbName, tableName string) (string, error) { - query := fmt.Sprintf("SHOW CREATE TABLE `%s`.`%s`", dbName, tableName) - if dbName == "" { - query = fmt.Sprintf("SHOW CREATE TABLE `%s`", tableName) +func normalizeMySQLIdentifierPart(ident string) string { + value := strings.TrimSpace(ident) + for i := 0; i < 4; i++ { + next := normalizeSQLIdentPartCommon(value) + if next == value { + break + } + value = next + } + if len(value) >= 2 { + first := value[0] + last := value[len(value)-1] + switch { + case first == '\'' && last == '\'': + return strings.TrimSpace(strings.ReplaceAll(value[1:len(value)-1], `''`, `'`)) + case (first == '\'' && last == '"') || (first == '"' && last == '\''): + return strings.TrimSpace(value[1 : len(value)-1]) + } + } + return strings.TrimSpace(value) +} + +func quoteMySQLIdentifier(ident string) string { + return "`" + strings.ReplaceAll(normalizeMySQLIdentifierPart(ident), "`", "``") + "`" +} + +func mysqlQualifiedTableIdentifier(dbName, tableName string) string { + schema := normalizeMySQLIdentifierPart(dbName) + table := strings.TrimSpace(tableName) + if parsedSchema, parsedTable := SplitSQLQualifiedName(table); parsedTable != "" { + if parsedSchema != "" { + schema = normalizeMySQLIdentifierPart(parsedSchema) + } + table = normalizeMySQLIdentifierPart(parsedTable) + } else { + table = normalizeMySQLIdentifierPart(table) } - data, _, err := m.Query(query) + if schema != "" { + return quoteMySQLIdentifier(schema) + "." + quoteMySQLIdentifier(table) + } + return quoteMySQLIdentifier(table) +} + +func buildMySQLShowCreateTableQuery(dbName, tableName string) string { + return "SHOW CREATE TABLE " + mysqlQualifiedTableIdentifier(dbName, tableName) +} + +func (m *MySQLDB) GetCreateStatement(dbName, tableName string) (string, error) { + data, _, err := m.Query(buildMySQLShowCreateTableQuery(dbName, tableName)) if err != nil { return "", err } diff --git a/internal/db/mysql_metadata_test.go b/internal/db/mysql_metadata_test.go index 45df52aa..b3855ffc 100644 --- a/internal/db/mysql_metadata_test.go +++ b/internal/db/mysql_metadata_test.go @@ -82,3 +82,53 @@ func TestCollectMySQLDatabaseNames_ReturnsOriginalErrorWhenNoDatabaseResolved(t t.Fatalf("错误不符合预期: %v", err) } } + +func TestBuildMySQLShowCreateTableQueryNormalizesQuotedIdentifiers(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + dbName string + tableName string + want string + }{ + { + name: "plain db and quoted table", + dbName: "app", + tableName: `"activate_record"`, + want: "SHOW CREATE TABLE `app`.`activate_record`", + }, + { + name: "escaped quoted qualified table overrides db", + dbName: "ignored", + tableName: `\"crm\".\"activate_record\"`, + want: "SHOW CREATE TABLE `crm`.`activate_record`", + }, + { + name: "backtick escaping", + dbName: "app`prod", + tableName: "`audit``log`", + want: "SHOW CREATE TABLE `app``prod`.`audit``log`", + }, + { + name: "quoted table containing dot is not split", + dbName: "app", + tableName: `"activate.record"`, + want: "SHOW CREATE TABLE `app`.`activate.record`", + }, + { + name: "mixed quote artifact from UI row value", + dbName: "app", + tableName: `'activate_record"`, + want: "SHOW CREATE TABLE `app`.`activate_record`", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := buildMySQLShowCreateTableQuery(tt.dbName, tt.tableName); got != tt.want { + t.Fatalf("buildMySQLShowCreateTableQuery(%q,%q)=%q,want=%q", tt.dbName, tt.tableName, got, tt.want) + } + }) + } +} diff --git a/internal/sync/analyze.go b/internal/sync/analyze.go index acb9c3a7..9e61a2a5 100644 --- a/internal/sync/analyze.go +++ b/internal/sync/analyze.go @@ -32,6 +32,7 @@ type SyncAnalyzeResult struct { } func (s *SyncEngine) Analyze(config SyncConfig) SyncAnalyzeResult { + config = normalizeSyncConnectionDatabases(config) result := SyncAnalyzeResult{Success: true, Tables: []TableDiffSummary{}} if isRedisToMongoKeyspacePair(config) { return s.analyzeRedisToMongo(config) diff --git a/internal/sync/migration_clickhouse.go b/internal/sync/migration_clickhouse.go index 1c5f6658..a472aeaf 100644 --- a/internal/sync/migration_clickhouse.go +++ b/internal/sync/migration_clickhouse.go @@ -10,8 +10,8 @@ import ( func buildMySQLToClickHousePlan(config SyncConfig, tableName string, sourceDB db.Database, targetDB db.Database) (SchemaMigrationPlan, []connection.ColumnDefinition, []connection.ColumnDefinition, error) { plan := SchemaMigrationPlan{} - plan.SourceSchema, plan.SourceTable = normalizeSchemaAndTable(config.SourceConfig.Type, config.SourceConfig.Database, tableName) - plan.TargetSchema, plan.TargetTable = normalizeSchemaAndTable(config.TargetConfig.Type, config.TargetConfig.Database, tableName) + plan.SourceSchema, plan.SourceTable = normalizeSchemaAndTable(config.SourceConfig.Type, selectedSyncSourceDatabase(config), tableName) + plan.TargetSchema, plan.TargetTable = normalizeSchemaAndTable(config.TargetConfig.Type, selectedSyncTargetDatabase(config), tableName) plan.SourceQueryTable = qualifiedNameForQuery(config.SourceConfig.Type, plan.SourceSchema, plan.SourceTable, tableName) plan.TargetQueryTable = qualifiedNameForQuery(config.TargetConfig.Type, plan.TargetSchema, plan.TargetTable, tableName) plan.PlannedAction = "使用已有目标表导入" @@ -70,8 +70,8 @@ func buildPGLikeToClickHousePlan(config SyncConfig, tableName string, sourceDB d plan := SchemaMigrationPlan{} sourceType := resolveMigrationDBType(config.SourceConfig) targetType := resolveMigrationDBType(config.TargetConfig) - plan.SourceSchema, plan.SourceTable = normalizeSchemaAndTable(sourceType, config.SourceConfig.Database, tableName) - plan.TargetSchema, plan.TargetTable = normalizeSchemaAndTable(targetType, config.TargetConfig.Database, tableName) + plan.SourceSchema, plan.SourceTable = normalizeSchemaAndTable(sourceType, selectedSyncSourceDatabase(config), tableName) + plan.TargetSchema, plan.TargetTable = normalizeSchemaAndTable(targetType, selectedSyncTargetDatabase(config), tableName) plan.SourceQueryTable = qualifiedNameForQuery(sourceType, plan.SourceSchema, plan.SourceTable, tableName) plan.TargetQueryTable = qualifiedNameForQuery(targetType, plan.TargetSchema, plan.TargetTable, tableName) plan.PlannedAction = "使用已有目标表导入" @@ -128,8 +128,8 @@ func buildPGLikeToClickHousePlan(config SyncConfig, tableName string, sourceDB d func buildClickHouseToMySQLPlan(config SyncConfig, tableName string, sourceDB db.Database, targetDB db.Database) (SchemaMigrationPlan, []connection.ColumnDefinition, []connection.ColumnDefinition, error) { plan := SchemaMigrationPlan{} - plan.SourceSchema, plan.SourceTable = normalizeSchemaAndTable(config.SourceConfig.Type, config.SourceConfig.Database, tableName) - plan.TargetSchema, plan.TargetTable = normalizeSchemaAndTable(config.TargetConfig.Type, config.TargetConfig.Database, tableName) + plan.SourceSchema, plan.SourceTable = normalizeSchemaAndTable(config.SourceConfig.Type, selectedSyncSourceDatabase(config), tableName) + plan.TargetSchema, plan.TargetTable = normalizeSchemaAndTable(config.TargetConfig.Type, selectedSyncTargetDatabase(config), tableName) plan.SourceQueryTable = qualifiedNameForQuery(config.SourceConfig.Type, plan.SourceSchema, plan.SourceTable, tableName) plan.TargetQueryTable = qualifiedNameForQuery(config.TargetConfig.Type, plan.TargetSchema, plan.TargetTable, tableName) plan.PlannedAction = "使用已有目标表导入" @@ -187,8 +187,8 @@ func buildClickHouseToPGLikePlan(config SyncConfig, tableName string, sourceDB d plan := SchemaMigrationPlan{} sourceType := resolveMigrationDBType(config.SourceConfig) targetType := resolveMigrationDBType(config.TargetConfig) - plan.SourceSchema, plan.SourceTable = normalizeSchemaAndTable(sourceType, config.SourceConfig.Database, tableName) - plan.TargetSchema, plan.TargetTable = normalizeSchemaAndTable(targetType, config.TargetConfig.Database, tableName) + plan.SourceSchema, plan.SourceTable = normalizeSchemaAndTable(sourceType, selectedSyncSourceDatabase(config), tableName) + plan.TargetSchema, plan.TargetTable = normalizeSchemaAndTable(targetType, selectedSyncTargetDatabase(config), tableName) plan.SourceQueryTable = qualifiedNameForQuery(sourceType, plan.SourceSchema, plan.SourceTable, tableName) plan.TargetQueryTable = qualifiedNameForQuery(targetType, plan.TargetSchema, plan.TargetTable, tableName) plan.PlannedAction = "使用已有目标表导入" @@ -247,8 +247,8 @@ func buildClickHouseToClickHousePlan(config SyncConfig, tableName string, source plan := SchemaMigrationPlan{} sourceType := resolveMigrationDBType(config.SourceConfig) targetType := resolveMigrationDBType(config.TargetConfig) - plan.SourceSchema, plan.SourceTable = normalizeSchemaAndTable(sourceType, config.SourceConfig.Database, tableName) - plan.TargetSchema, plan.TargetTable = normalizeSchemaAndTable(targetType, config.TargetConfig.Database, tableName) + plan.SourceSchema, plan.SourceTable = normalizeSchemaAndTable(sourceType, selectedSyncSourceDatabase(config), tableName) + plan.TargetSchema, plan.TargetTable = normalizeSchemaAndTable(targetType, selectedSyncTargetDatabase(config), tableName) plan.SourceQueryTable = qualifiedNameForQuery(sourceType, plan.SourceSchema, plan.SourceTable, tableName) plan.TargetQueryTable = qualifiedNameForQuery(targetType, plan.TargetSchema, plan.TargetTable, tableName) plan.PlannedAction = "使用已有目标表导入" diff --git a/internal/sync/migration_kernel_router.go b/internal/sync/migration_kernel_router.go index aff59737..fde0728e 100644 --- a/internal/sync/migration_kernel_router.go +++ b/internal/sync/migration_kernel_router.go @@ -434,8 +434,8 @@ func (mongoToRelationalPlanner) BuildPlan(ctx MigrationBuildContext) (SchemaMigr return SchemaMigrationPlan{}, nil, nil, err } plan := SchemaMigrationPlan{} - plan.SourceSchema, plan.SourceTable = normalizeSchemaAndTable(sourceType, ctx.Config.SourceConfig.Database, ctx.TableName) - plan.TargetSchema, plan.TargetTable = normalizeSchemaAndTable(targetType, ctx.Config.TargetConfig.Database, ctx.TableName) + plan.SourceSchema, plan.SourceTable = normalizeSchemaAndTable(sourceType, selectedSyncSourceDatabase(ctx.Config), ctx.TableName) + plan.TargetSchema, plan.TargetTable = normalizeSchemaAndTable(targetType, selectedSyncTargetDatabase(ctx.Config), ctx.TableName) plan.SourceQueryTable = qualifiedNameForQuery(sourceType, plan.SourceSchema, plan.SourceTable, ctx.TableName) plan.TargetQueryTable = qualifiedNameForQuery(targetType, plan.TargetSchema, plan.TargetTable, ctx.TableName) plan.PlannedAction = "当前库对已进入迁移内核规划阶段，等待 schema 推断与目标方言生成器落地" diff --git a/internal/sync/migration_mongodb.go b/internal/sync/migration_mongodb.go index 8cb1422c..d30411c5 100644 --- a/internal/sync/migration_mongodb.go +++ b/internal/sync/migration_mongodb.go @@ -30,8 +30,8 @@ func buildTabularToMongoPlan(config SyncConfig, tableName string, sourceDB db.Da plan := SchemaMigrationPlan{} sourceType := resolveMigrationDBType(config.SourceConfig) targetType := resolveMigrationDBType(config.TargetConfig) - plan.SourceSchema, plan.SourceTable = normalizeSchemaAndTable(sourceType, config.SourceConfig.Database, tableName) - plan.TargetSchema, plan.TargetTable = normalizeSchemaAndTable(targetType, config.TargetConfig.Database, tableName) + plan.SourceSchema, plan.SourceTable = normalizeSchemaAndTable(sourceType, selectedSyncSourceDatabase(config), tableName) + plan.TargetSchema, plan.TargetTable = normalizeSchemaAndTable(targetType, selectedSyncTargetDatabase(config), tableName) plan.SourceQueryTable = qualifiedNameForQuery(sourceType, plan.SourceSchema, plan.SourceTable, tableName) plan.TargetQueryTable = qualifiedNameForQuery(targetType, plan.TargetSchema, plan.TargetTable, tableName) plan.PlannedAction = "使用已有目标集合导入" @@ -91,8 +91,8 @@ func buildMongoToMongoPlan(config SyncConfig, tableName string, sourceDB db.Data plan := SchemaMigrationPlan{} sourceType := resolveMigrationDBType(config.SourceConfig) targetType := resolveMigrationDBType(config.TargetConfig) - plan.SourceSchema, plan.SourceTable = normalizeSchemaAndTable(sourceType, config.SourceConfig.Database, tableName) - plan.TargetSchema, plan.TargetTable = normalizeSchemaAndTable(targetType, config.TargetConfig.Database, tableName) + plan.SourceSchema, plan.SourceTable = normalizeSchemaAndTable(sourceType, selectedSyncSourceDatabase(config), tableName) + plan.TargetSchema, plan.TargetTable = normalizeSchemaAndTable(targetType, selectedSyncTargetDatabase(config), tableName) plan.SourceQueryTable = qualifiedNameForQuery(sourceType, plan.SourceSchema, plan.SourceTable, tableName) plan.TargetQueryTable = qualifiedNameForQuery(targetType, plan.TargetSchema, plan.TargetTable, tableName) plan.PlannedAction = "使用已有目标集合导入" @@ -154,8 +154,8 @@ func buildMongoToMongoPlan(config SyncConfig, tableName string, sourceDB db.Data func buildMongoToMySQLPlan(config SyncConfig, tableName string, sourceDB db.Database, targetDB db.Database) (SchemaMigrationPlan, []connection.ColumnDefinition, []connection.ColumnDefinition, error) { plan := SchemaMigrationPlan{} - plan.SourceSchema, plan.SourceTable = normalizeSchemaAndTable(config.SourceConfig.Type, config.SourceConfig.Database, tableName) - plan.TargetSchema, plan.TargetTable = normalizeSchemaAndTable(config.TargetConfig.Type, config.TargetConfig.Database, tableName) + plan.SourceSchema, plan.SourceTable = normalizeSchemaAndTable(config.SourceConfig.Type, selectedSyncSourceDatabase(config), tableName) + plan.TargetSchema, plan.TargetTable = normalizeSchemaAndTable(config.TargetConfig.Type, selectedSyncTargetDatabase(config), tableName) plan.SourceQueryTable = qualifiedNameForQuery(config.SourceConfig.Type, plan.SourceSchema, plan.SourceTable, tableName) plan.TargetQueryTable = qualifiedNameForQuery(config.TargetConfig.Type, plan.TargetSchema, plan.TargetTable, tableName) plan.PlannedAction = "使用已有目标表导入" @@ -497,8 +497,8 @@ func moveStringToFront(items []string, target string) []string { func buildMongoToPGLikePlan(config SyncConfig, tableName string, sourceDB db.Database, targetDB db.Database) (SchemaMigrationPlan, []connection.ColumnDefinition, []connection.ColumnDefinition, error) { plan := SchemaMigrationPlan{} targetType := strings.ToLower(strings.TrimSpace(config.TargetConfig.Type)) - plan.SourceSchema, plan.SourceTable = normalizeSchemaAndTable(config.SourceConfig.Type, config.SourceConfig.Database, tableName) - plan.TargetSchema, plan.TargetTable = normalizeSchemaAndTable(config.TargetConfig.Type, config.TargetConfig.Database, tableName) + plan.SourceSchema, plan.SourceTable = normalizeSchemaAndTable(config.SourceConfig.Type, selectedSyncSourceDatabase(config), tableName) + plan.TargetSchema, plan.TargetTable = normalizeSchemaAndTable(config.TargetConfig.Type, selectedSyncTargetDatabase(config), tableName) plan.SourceQueryTable = qualifiedNameForQuery(config.SourceConfig.Type, plan.SourceSchema, plan.SourceTable, tableName) plan.TargetQueryTable = qualifiedNameForQuery(config.TargetConfig.Type, plan.TargetSchema, plan.TargetTable, tableName) plan.PlannedAction = "使用已有目标表导入" diff --git a/internal/sync/migration_redis.go b/internal/sync/migration_redis.go index 84f159f5..83fea67b 100644 --- a/internal/sync/migration_redis.go +++ b/internal/sync/migration_redis.go @@ -73,7 +73,7 @@ func buildRedisToMongoPlan(config SyncConfig, keyName string, targetDB db.Databa SourceSchema: strconv.Itoa(resolveRedisDBIndex(config.SourceConfig)), SourceTable: keyName, SourceQueryTable: keyName, - TargetSchema: strings.TrimSpace(config.TargetConfig.Database), + TargetSchema: strings.TrimSpace(selectedSyncTargetDatabase(config)), TargetTable: collection, TargetQueryTable: collection, PlannedAction: "按 Redis Key 生成 MongoDB 文档导入", @@ -557,7 +557,7 @@ func listMongoRedisCollections(sourceDB db.Database, config SyncConfig) ([]strin if len(config.Tables) > 0 { return dedupeStrings(config.Tables), nil } - tables, err := sourceDB.GetTables(strings.TrimSpace(config.SourceConfig.Database)) + tables, err := sourceDB.GetTables(strings.TrimSpace(selectedSyncSourceDatabase(config))) if err == nil && len(tables) > 0 { return dedupeStrings(tables), nil } diff --git a/internal/sync/migration_tdengine.go b/internal/sync/migration_tdengine.go index 7e45e643..3a6d690d 100644 --- a/internal/sync/migration_tdengine.go +++ b/internal/sync/migration_tdengine.go @@ -12,8 +12,8 @@ func buildTDengineToMySQLPlan(config SyncConfig, tableName string, sourceDB db.D plan := SchemaMigrationPlan{} sourceType := resolveMigrationDBType(config.SourceConfig) targetType := resolveMigrationDBType(config.TargetConfig) - plan.SourceSchema, plan.SourceTable = normalizeSchemaAndTable(sourceType, config.SourceConfig.Database, tableName) - plan.TargetSchema, plan.TargetTable = normalizeSchemaAndTable(targetType, config.TargetConfig.Database, tableName) + plan.SourceSchema, plan.SourceTable = normalizeSchemaAndTable(sourceType, selectedSyncSourceDatabase(config), tableName) + plan.TargetSchema, plan.TargetTable = normalizeSchemaAndTable(targetType, selectedSyncTargetDatabase(config), tableName) plan.SourceQueryTable = qualifiedNameForQuery(sourceType, plan.SourceSchema, plan.SourceTable, tableName) plan.TargetQueryTable = qualifiedNameForQuery(targetType, plan.TargetSchema, plan.TargetTable, tableName) plan.PlannedAction = "使用已有目标表导入" @@ -67,8 +67,8 @@ func buildTDengineToPGLikePlan(config SyncConfig, tableName string, sourceDB db. plan := SchemaMigrationPlan{} sourceType := resolveMigrationDBType(config.SourceConfig) targetType := resolveMigrationDBType(config.TargetConfig) - plan.SourceSchema, plan.SourceTable = normalizeSchemaAndTable(sourceType, config.SourceConfig.Database, tableName) - plan.TargetSchema, plan.TargetTable = normalizeSchemaAndTable(targetType, config.TargetConfig.Database, tableName) + plan.SourceSchema, plan.SourceTable = normalizeSchemaAndTable(sourceType, selectedSyncSourceDatabase(config), tableName) + plan.TargetSchema, plan.TargetTable = normalizeSchemaAndTable(targetType, selectedSyncTargetDatabase(config), tableName) plan.SourceQueryTable = qualifiedNameForQuery(sourceType, plan.SourceSchema, plan.SourceTable, tableName) plan.TargetQueryTable = qualifiedNameForQuery(targetType, plan.TargetSchema, plan.TargetTable, tableName) plan.PlannedAction = "使用已有目标表导入" diff --git a/internal/sync/migration_tdengine_target.go b/internal/sync/migration_tdengine_target.go index 50a18395..5ebc9dbb 100644 --- a/internal/sync/migration_tdengine_target.go +++ b/internal/sync/migration_tdengine_target.go @@ -100,8 +100,8 @@ func buildSourceToTDenginePlan(config SyncConfig, tableName string, sourceDB db. plan := SchemaMigrationPlan{} sourceType := resolveMigrationDBType(config.SourceConfig) targetType := resolveMigrationDBType(config.TargetConfig) - plan.SourceSchema, plan.SourceTable = normalizeSchemaAndTable(sourceType, config.SourceConfig.Database, tableName) - plan.TargetSchema, plan.TargetTable = normalizeSchemaAndTable(targetType, config.TargetConfig.Database, tableName) + plan.SourceSchema, plan.SourceTable = normalizeSchemaAndTable(sourceType, selectedSyncSourceDatabase(config), tableName) + plan.TargetSchema, plan.TargetTable = normalizeSchemaAndTable(targetType, selectedSyncTargetDatabase(config), tableName) plan.SourceQueryTable = qualifiedNameForQuery(sourceType, plan.SourceSchema, plan.SourceTable, tableName) plan.TargetQueryTable = qualifiedNameForQuery(targetType, plan.TargetSchema, plan.TargetTable, tableName) plan.PlannedAction = "使用已有目标表导入" diff --git a/internal/sync/preview.go b/internal/sync/preview.go index 7d4a4321..366bbd34 100644 --- a/internal/sync/preview.go +++ b/internal/sync/preview.go @@ -34,6 +34,7 @@ type TableDiffPreview struct { } func (s *SyncEngine) Preview(config SyncConfig, tableName string, limit int) (TableDiffPreview, error) { + config = normalizeSyncConnectionDatabases(config) if limit <= 0 { limit = 200 } diff --git a/internal/sync/schema_migration.go b/internal/sync/schema_migration.go index 9cde70f9..1e1adc38 100644 --- a/internal/sync/schema_migration.go +++ b/internal/sync/schema_migration.go @@ -98,8 +98,8 @@ func buildSchemaMigrationPlanLegacy(config SyncConfig, tableName string, sourceD plan := SchemaMigrationPlan{} sourceType := resolveMigrationDBType(config.SourceConfig) targetType := resolveMigrationDBType(config.TargetConfig) - plan.SourceSchema, plan.SourceTable = normalizeSchemaAndTable(sourceType, config.SourceConfig.Database, tableName) - plan.TargetSchema, plan.TargetTable = normalizeSchemaAndTable(targetType, config.TargetConfig.Database, tableName) + plan.SourceSchema, plan.SourceTable = normalizeSchemaAndTable(sourceType, selectedSyncSourceDatabase(config), tableName) + plan.TargetSchema, plan.TargetTable = normalizeSchemaAndTable(targetType, selectedSyncTargetDatabase(config), tableName) plan.SourceQueryTable = qualifiedNameForQuery(sourceType, plan.SourceSchema, plan.SourceTable, tableName) plan.TargetQueryTable = qualifiedNameForQuery(targetType, plan.TargetSchema, plan.TargetTable, tableName) plan.PlannedAction = "使用已有目标表导入" @@ -593,8 +593,8 @@ func buildMySQLToMySQLPlan(config SyncConfig, tableName string, sourceDB db.Data plan := SchemaMigrationPlan{} sourceType := resolveMigrationDBType(config.SourceConfig) targetType := resolveMigrationDBType(config.TargetConfig) - plan.SourceSchema, plan.SourceTable = normalizeSchemaAndTable(sourceType, config.SourceConfig.Database, tableName) - plan.TargetSchema, plan.TargetTable = normalizeSchemaAndTable(targetType, config.TargetConfig.Database, tableName) + plan.SourceSchema, plan.SourceTable = normalizeSchemaAndTable(sourceType, selectedSyncSourceDatabase(config), tableName) + plan.TargetSchema, plan.TargetTable = normalizeSchemaAndTable(targetType, selectedSyncTargetDatabase(config), tableName) plan.SourceQueryTable = qualifiedNameForQuery(sourceType, plan.SourceSchema, plan.SourceTable, tableName) plan.TargetQueryTable = qualifiedNameForQuery(targetType, plan.TargetSchema, plan.TargetTable, tableName) plan.PlannedAction = "使用已有目标表导入" @@ -832,8 +832,8 @@ func buildPGLikeToPGLikePlan(config SyncConfig, tableName string, sourceDB db.Da plan := SchemaMigrationPlan{} sourceType := resolveMigrationDBType(config.SourceConfig) targetType := resolveMigrationDBType(config.TargetConfig) - plan.SourceSchema, plan.SourceTable = normalizeSchemaAndTable(sourceType, config.SourceConfig.Database, tableName) - plan.TargetSchema, plan.TargetTable = normalizeSchemaAndTable(targetType, config.TargetConfig.Database, tableName) + plan.SourceSchema, plan.SourceTable = normalizeSchemaAndTable(sourceType, selectedSyncSourceDatabase(config), tableName) + plan.TargetSchema, plan.TargetTable = normalizeSchemaAndTable(targetType, selectedSyncTargetDatabase(config), tableName) plan.SourceQueryTable = qualifiedNameForQuery(sourceType, plan.SourceSchema, plan.SourceTable, tableName) plan.TargetQueryTable = qualifiedNameForQuery(targetType, plan.TargetSchema, plan.TargetTable, tableName) plan.PlannedAction = "使用已有目标表导入" @@ -1078,8 +1078,8 @@ func buildPGLikeToMySQLPlan(config SyncConfig, tableName string, sourceDB db.Dat plan := SchemaMigrationPlan{} sourceType := resolveMigrationDBType(config.SourceConfig) targetType := resolveMigrationDBType(config.TargetConfig) - plan.SourceSchema, plan.SourceTable = normalizeSchemaAndTable(sourceType, config.SourceConfig.Database, tableName) - plan.TargetSchema, plan.TargetTable = normalizeSchemaAndTable(targetType, config.TargetConfig.Database, tableName) + plan.SourceSchema, plan.SourceTable = normalizeSchemaAndTable(sourceType, selectedSyncSourceDatabase(config), tableName) + plan.TargetSchema, plan.TargetTable = normalizeSchemaAndTable(targetType, selectedSyncTargetDatabase(config), tableName) plan.SourceQueryTable = qualifiedNameForQuery(sourceType, plan.SourceSchema, plan.SourceTable, tableName) plan.TargetQueryTable = qualifiedNameForQuery(targetType, plan.TargetSchema, plan.TargetTable, tableName) plan.PlannedAction = "使用已有目标表导入" @@ -1369,8 +1369,8 @@ func buildMySQLToPGLikePlan(config SyncConfig, tableName string, sourceDB db.Dat plan := SchemaMigrationPlan{} sourceType := resolveMigrationDBType(config.SourceConfig) targetType := resolveMigrationDBType(config.TargetConfig) - plan.SourceSchema, plan.SourceTable = normalizeSchemaAndTable(sourceType, config.SourceConfig.Database, tableName) - plan.TargetSchema, plan.TargetTable = normalizeSchemaAndTable(targetType, config.TargetConfig.Database, tableName) + plan.SourceSchema, plan.SourceTable = normalizeSchemaAndTable(sourceType, selectedSyncSourceDatabase(config), tableName) + plan.TargetSchema, plan.TargetTable = normalizeSchemaAndTable(targetType, selectedSyncTargetDatabase(config), tableName) plan.SourceQueryTable = qualifiedNameForQuery(sourceType, plan.SourceSchema, plan.SourceTable, tableName) plan.TargetQueryTable = qualifiedNameForQuery(targetType, plan.TargetSchema, plan.TargetTable, tableName) plan.PlannedAction = "使用已有目标表导入" diff --git a/internal/sync/schema_sync.go b/internal/sync/schema_sync.go index f93abad7..a964d517 100644 --- a/internal/sync/schema_sync.go +++ b/internal/sync/schema_sync.go @@ -14,8 +14,8 @@ func (s *SyncEngine) syncTableSchema(config SyncConfig, res *SyncResult, sourceD } sourceType := resolveMigrationDBType(config.SourceConfig) - sourceSchema, sourceTable := normalizeSchemaAndTable(sourceType, config.SourceConfig.Database, tableName) - targetSchema, targetTable := normalizeSchemaAndTable(targetType, config.TargetConfig.Database, tableName) + sourceSchema, sourceTable := normalizeSchemaAndTable(sourceType, selectedSyncSourceDatabase(config), tableName) + targetSchema, targetTable := normalizeSchemaAndTable(targetType, selectedSyncTargetDatabase(config), tableName) targetQueryTable := qualifiedNameForQuery(targetType, targetSchema, targetTable, tableName) // 1) 获取源表字段 diff --git a/internal/sync/source_query_sync.go b/internal/sync/source_query_sync.go index 4b36cc13..f5901f2f 100644 --- a/internal/sync/source_query_sync.go +++ b/internal/sync/source_query_sync.go @@ -47,7 +47,7 @@ func validateSourceQuerySyncConfig(config SyncConfig) (string, error) { func resolveTargetQueryTable(config SyncConfig, tableName string) (string, string, string, string) { targetType := resolveMigrationDBType(config.TargetConfig) - targetSchema, targetTable := normalizeSchemaAndTable(targetType, config.TargetConfig.Database, tableName) + targetSchema, targetTable := normalizeSchemaAndTable(targetType, selectedSyncTargetDatabase(config), tableName) targetQueryTable := qualifiedNameForQuery(targetType, targetSchema, targetTable, tableName) return targetType, targetSchema, targetTable, targetQueryTable } diff --git a/internal/sync/sql_helpers.go b/internal/sync/sql_helpers.go index 5d5716e5..fd759b3e 100644 --- a/internal/sync/sql_helpers.go +++ b/internal/sync/sql_helpers.go @@ -3,9 +3,73 @@ package sync import ( "strings" + "GoNavi-Wails/internal/connection" "GoNavi-Wails/internal/db" ) +func selectedSyncDatabase(selected string, fallback string) string { + if value := strings.TrimSpace(selected); value != "" { + return value + } + return strings.TrimSpace(fallback) +} + +func selectedSyncSourceDatabase(config SyncConfig) string { + return selectedSyncDatabase(config.SourceDatabase, config.SourceConfig.Database) +} + +func selectedSyncTargetDatabase(config SyncConfig) string { + return selectedSyncDatabase(config.TargetDatabase, config.TargetConfig.Database) +} + +func normalizeSyncConnectionDatabases(config SyncConfig) SyncConfig { + config.SourceConfig = normalizeSyncConnectionDatabase(config.SourceConfig, config.SourceDatabase) + config.TargetConfig = normalizeSyncConnectionDatabase(config.TargetConfig, config.TargetDatabase) + return config +} + +func normalizeSyncConnectionDatabase(config connection.ConnectionConfig, selectedDatabase string) connection.ConnectionConfig { + selected := strings.TrimSpace(selectedDatabase) + if selected == "" { + return config + } + switch resolveMigrationDBType(config) { + case "oracle": + // Oracle 的 ConnectionConfig.Database 是 Service Name，数据同步选择的是 schema/owner。 + return config + case "oceanbase": + if isOceanBaseOracleSyncConnection(config) { + return config + } + default: + config.Database = selected + return config + } + config.Database = selected + return config +} + +func isOceanBaseOracleSyncConnection(config connection.ConnectionConfig) bool { + if !strings.EqualFold(strings.TrimSpace(config.Type), "oceanbase") { + return false + } + if strings.EqualFold(strings.TrimSpace(config.OceanBaseProtocol), "oracle") { + return true + } + for _, part := range strings.FieldsFunc(config.ConnectionParams, func(r rune) bool { return r == '&' || r == ';' }) { + key, value, ok := strings.Cut(part, "=") + if !ok { + continue + } + normalizedKey := strings.ToLower(strings.TrimSpace(key)) + normalizedValue := strings.ToLower(strings.TrimSpace(value)) + if (normalizedKey == "protocol" || normalizedKey == "tenantmode") && normalizedValue == "oracle" { + return true + } + } + return false +} + func normalizeSyncMode(mode string) string { m := strings.ToLower(strings.TrimSpace(mode)) switch m { diff --git a/internal/sync/sql_helpers_test.go b/internal/sync/sql_helpers_test.go index bd7c0aae..82d72452 100644 --- a/internal/sync/sql_helpers_test.go +++ b/internal/sync/sql_helpers_test.go @@ -52,6 +52,49 @@ func TestNormalizeSchemaAndTable_KingbaseNormalizesEscapedQualifiedName(t *testi } } +func TestNormalizeSyncConnectionDatabasesKeepsOracleServiceName(t *testing.T) { + t.Parallel() + + config := SyncConfig{ + SourceConfig: connection.ConnectionConfig{Type: "oracle", Database: "ORCLPDB1"}, + SourceDatabase: "APP_SCHEMA", + TargetConfig: connection.ConnectionConfig{Type: "mysql", Database: "old_target"}, + TargetDatabase: "warehouse", + } + + got := normalizeSyncConnectionDatabases(config) + if got.SourceConfig.Database != "ORCLPDB1" { + t.Fatalf("Oracle 连接 Service Name 不应被 schema 覆盖，got=%q", got.SourceConfig.Database) + } + if selectedSyncSourceDatabase(got) != "APP_SCHEMA" { + t.Fatalf("Oracle 选中 schema 应保留在 SourceDatabase，got=%q", selectedSyncSourceDatabase(got)) + } + if got.TargetConfig.Database != "warehouse" { + t.Fatalf("非 Oracle 目标库应继续写入连接 Database，got=%q", got.TargetConfig.Database) + } +} + +func TestNormalizeSyncConnectionDatabasesKeepsOceanBaseOracleServiceName(t *testing.T) { + t.Parallel() + + config := SyncConfig{ + SourceConfig: connection.ConnectionConfig{ + Type: "oceanbase", + Database: "ORCL", + ConnectionParams: "protocol=oracle", + }, + SourceDatabase: "APP_SCHEMA", + } + + got := normalizeSyncConnectionDatabases(config) + if got.SourceConfig.Database != "ORCL" { + t.Fatalf("OceanBase Oracle 服务名不应被 schema 覆盖，got=%q", got.SourceConfig.Database) + } + if selectedSyncSourceDatabase(got) != "APP_SCHEMA" { + t.Fatalf("OceanBase Oracle 选中 schema 应保留在 SourceDatabase，got=%q", selectedSyncSourceDatabase(got)) + } +} + func TestNormalizeMigrationDBType_KingbaseAliases(t *testing.T) { t.Parallel() diff --git a/internal/sync/sync_engine.go b/internal/sync/sync_engine.go index ff8795f0..6915fc7e 100644 --- a/internal/sync/sync_engine.go +++ b/internal/sync/sync_engine.go @@ -17,6 +17,8 @@ const defaultSyncApplyBatchSize = 1000 type SyncConfig struct { SourceConfig connection.ConnectionConfig `json:"sourceConfig"` TargetConfig connection.ConnectionConfig `json:"targetConfig"` + SourceDatabase string `json:"sourceDatabase,omitempty"` + TargetDatabase string `json:"targetDatabase,omitempty"` Tables []string `json:"tables"` SourceQuery string `json:"sourceQuery,omitempty"` Content string `json:"content,omitempty"` // "data", "schema", "both" @@ -50,6 +52,7 @@ func NewSyncEngine(reporter Reporter) *SyncEngine { // CompareAndSync performs the synchronization func (s *SyncEngine) RunSync(config SyncConfig) SyncResult { + config = normalizeSyncConnectionDatabases(config) result := SyncResult{Success: true, Logs: []string{}} logger.Infof("开始数据同步：源=%s 目标=%s 表数量=%d", formatConnSummaryForSync(config.SourceConfig), formatConnSummaryForSync(config.TargetConfig), len(config.Tables)) if isRedisToMongoKeyspacePair(config) {