fix(dlm): don't compare the table structure if syncTableStructure is …

…off (#3014)
oceanbase · Jul 25, 2024 · 743c034 · 743c034
1 parent c586b89
commit 743c034
Show file tree

Hide file tree

Showing 4 changed files with 18 additions and 7 deletions.
diff --git a/...ce/src/main/java/com/oceanbase/odc/service/schedule/flowtask/DataArchivePreprocessor.java b/...ce/src/main/java/com/oceanbase/odc/service/schedule/flowtask/DataArchivePreprocessor.java
@@ -86,10 +86,9 @@ public void process(CreateFlowInstanceReq req) {
             // permission to access it.
             Database sourceDb = databaseService.detail(dataArchiveParameters.getSourceDatabaseId());
             Database targetDb = databaseService.detail(dataArchiveParameters.getTargetDataBaseId());
-            if (!dataArchiveParameters.getSyncTableStructure().isEmpty()
-                    && sourceDb.getDataSource().getDialectType() != targetDb.getDataSource().getDialectType()) {
-                throw new UnsupportedException(
-                        "Different types of databases do not support structural synchronization.");
+            if (!dataArchiveParameters.getSyncTableStructure().isEmpty()) {
+                supportSyncTableStructure(sourceDb.getDataSource().getDialectType(), targetDb.getDataSource()
+                        .getDialectType());
             }
             dataArchiveParameters.setSourceDatabaseName(sourceDb.getName());
             dataArchiveParameters.setTargetDatabaseName(targetDb.getName());
@@ -188,4 +187,15 @@ private void initDefaultConfig(DataArchiveParameters parameters) {
         parameters.setQueryTimeout(dlmConfiguration.getTaskConnectionQueryTimeout());
         parameters.setShardingStrategy(dlmConfiguration.getShardingStrategy());
     }
+
+    private void supportSyncTableStructure(DialectType srcDbType, DialectType tgtDbType) {
+        if (srcDbType != tgtDbType) {
+            throw new UnsupportedException(
+                    "Different types of databases do not support table structure synchronization.");
+        }
+        if (!srcDbType.isMysql()) {
+            throw new UnsupportedException(
+                    String.format("The database does not support table structure synchronization,type=%s", srcDbType));
+        }
+    }
 }
diff --git a/server/odc-service/src/main/java/com/oceanbase/odc/service/schedule/job/AbstractDlmJob.java b/server/odc-service/src/main/java/com/oceanbase/odc/service/schedule/job/AbstractDlmJob.java
@@ -114,7 +114,8 @@ public void executeTask(Long taskId, List<DlmTableUnit> dlmTableUnits, Long time
                         dlmTableUnit.getTableName());
                 continue;
             }
-            if (dlmTableUnit.getType() == JobType.MIGRATE) {
+            if (dlmTableUnit.getType() == JobType.MIGRATE
+                    && !dlmTableUnit.getParameters().getSyncDBObjectType().isEmpty()) {
                 try {
                     DLMTableStructureSynchronizer.sync(dlmTableUnit.getSourceDatasourceInfo(),
                             dlmTableUnit.getTargetDatasourceInfo(), dlmTableUnit.getTableName(),

diff --git a/...c-service/src/main/java/com/oceanbase/odc/service/task/executor/task/DataArchiveTask.java b/...c-service/src/main/java/com/oceanbase/odc/service/task/executor/task/DataArchiveTask.java
@@ -88,7 +88,7 @@ protected boolean doStart(JobContext context) throws Exception {
                 log.info("The table had been completed,tableName={}", dlmTableUnit.getTableName());
                 continue;
             }
-            if (parameters.getJobType() == JobType.MIGRATE) {
+            if (parameters.getJobType() == JobType.MIGRATE && !parameters.getSyncTableStructure().isEmpty()) {
                 try {
                     DLMTableStructureSynchronizer.sync(
                             DataSourceInfoMapper.toConnectionConfig(parameters.getSourceDs()),

diff --git a/...ce/src/main/java/com/oceanbase/odc/service/task/listener/DefaultJobTerminateListener.java b/...ce/src/main/java/com/oceanbase/odc/service/task/listener/DefaultJobTerminateListener.java
@@ -77,7 +77,7 @@ public void onEvent(JobTerminateEvent event) {
                 scheduleService.refreshScheduleStatus(Long.parseLong(o.getJobName()));
                 // Trigger the data-delete job if necessary after the data-archive task is completed.
                 if (parameters.getJobType() == com.oceanbase.tools.migrator.common.enums.JobType.MIGRATE
-                        && parameters.isDeleteAfterMigration()) {
+                        && parameters.isDeleteAfterMigration() && taskStatus == TaskStatus.DONE) {
                     scheduleService.dataArchiveDelete(Long.parseLong(o.getJobName()), o.getId());
                     log.info("Trigger delete job succeed.");
                 }