🐛 fix(sync): 修复 Oracle 同步连接与 MySQL 备份导出异常

- 分离 Oracle/OceanBase Oracle 同步连接 Service Name 与选中 schema - 兼容旧同步请求中 database 被 schema 覆盖的情况 - 规范 MySQL/MariaDB SHOW CREATE TABLE 标识符引用 Refs #549 Refs #518
2026-08-21 16:34:21 +08:00 · 2026-06-11 10:24:48 +08:00
parent 450d1d66b4
commit 74a422a5e2
23 changed files with 388 additions and 55 deletions
--- 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,
--- 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',
    });
  });
--- 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,
--- 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"];
--- 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)
--- 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")
+	}
+}
--- 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
 	}
--- 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
 	}
--- 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)
+			}
+		})
+	}
+}
--- 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)
--- 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 = "使用已有目标表导入"
--- 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 推断与目标方言生成器落地"
--- 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 = "使用已有目标表导入"
--- 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
 	}
--- 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 = "使用已有目标表导入"
--- 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 = "使用已有目标表导入"
--- 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
 	}
--- 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 = "使用已有目标表导入"
--- 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) 获取源表字段
--- 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
 }
--- 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 {
--- 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()

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