You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@shardingsphere.apache.org by pa...@apache.org on 2023/11/08 14:37:47 UTC
(shardingsphere) branch master updated: Refactor CreateTableConfiguration (#28986)
This is an automated email from the ASF dual-hosted git repository.
panjuan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shardingsphere.git
The following commit(s) were added to refs/heads/master by this push:
new 45b990823b9 Refactor CreateTableConfiguration (#28986)
45b990823b9 is described below
commit 45b990823b9fa019c4569b5760b8197f22d5152a
Author: Liang Zhang <zh...@apache.org>
AuthorDate: Wed Nov 8 22:37:38 2023 +0800
Refactor CreateTableConfiguration (#28986)
---
.../common/config/CreateTableConfiguration.java | 24 +++++++---------------
.../datasource/AbstractDataSourcePreparer.java | 17 ++++++++-------
.../datasource/PrepareTargetSchemasParameter.java | 4 +++-
.../datasource/PrepareTargetTablesParameter.java | 4 +++-
.../datasource/MySQLDataSourcePreparer.java | 4 ++--
.../datasource/OpenGaussDataSourcePreparer.java | 4 ++--
.../datasource/PostgreSQLDataSourcePreparer.java | 4 ++--
.../migration/api/impl/MigrationJobAPI.java | 17 +++++++--------
.../config/MigrationTaskConfiguration.java | 6 ++++--
.../migration/prepare/MigrationJobPreparer.java | 24 +++++++++++-----------
.../core/fixture/H2DataSourcePreparer.java | 4 ++--
11 files changed, 52 insertions(+), 60 deletions(-)
diff --git a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/common/config/CreateTableConfiguration.java b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/common/config/CreateTableConfiguration.java
index 8cbdca7819e..ede27092550 100644
--- a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/common/config/CreateTableConfiguration.java
+++ b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/common/config/CreateTableConfiguration.java
@@ -23,29 +23,19 @@ import lombok.ToString;
import org.apache.shardingsphere.data.pipeline.api.PipelineDataSourceConfiguration;
import org.apache.shardingsphere.data.pipeline.common.metadata.SchemaTableName;
-import java.util.Collection;
-
/**
* Create table configuration.
*/
@RequiredArgsConstructor
@Getter
-@ToString
+@ToString(exclude = {"sourceDataSourceConfig", "targetDataSourceConfig"})
public final class CreateTableConfiguration {
- private final Collection<CreateTableEntry> createTableEntries;
+ private final PipelineDataSourceConfiguration sourceDataSourceConfig;
+
+ private final SchemaTableName sourceName;
+
+ private final PipelineDataSourceConfiguration targetDataSourceConfig;
- @RequiredArgsConstructor
- @Getter
- @ToString(exclude = {"sourceDataSourceConfig", "targetDataSourceConfig"})
- public static final class CreateTableEntry {
-
- private final PipelineDataSourceConfiguration sourceDataSourceConfig;
-
- private final SchemaTableName sourceName;
-
- private final PipelineDataSourceConfiguration targetDataSourceConfig;
-
- private final SchemaTableName targetName;
- }
+ private final SchemaTableName targetName;
}
diff --git a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/preparer/datasource/AbstractDataSourcePreparer.java b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/preparer/datasource/AbstractDataSourcePreparer.java
index d7b584de272..32c3dc70e92 100644
--- a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/preparer/datasource/AbstractDataSourcePreparer.java
+++ b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/preparer/datasource/AbstractDataSourcePreparer.java
@@ -20,7 +20,6 @@ package org.apache.shardingsphere.data.pipeline.core.preparer.datasource;
import lombok.extern.slf4j.Slf4j;
import org.apache.shardingsphere.data.pipeline.api.PipelineDataSourceConfiguration;
import org.apache.shardingsphere.data.pipeline.common.config.CreateTableConfiguration;
-import org.apache.shardingsphere.data.pipeline.common.config.CreateTableConfiguration.CreateTableEntry;
import org.apache.shardingsphere.data.pipeline.common.datasource.PipelineDataSourceManager;
import org.apache.shardingsphere.data.pipeline.common.datasource.PipelineDataSourceWrapper;
import org.apache.shardingsphere.data.pipeline.common.metadata.generator.PipelineDDLGenerator;
@@ -56,11 +55,11 @@ public abstract class AbstractDataSourcePreparer implements DataSourcePreparer {
if (!dialectDatabaseMetaData.isSchemaAvailable()) {
return;
}
- CreateTableConfiguration createTableConfig = param.getCreateTableConfig();
+ Collection<CreateTableConfiguration> createTableConfigs = param.getCreateTableConfigurations();
String defaultSchema = dialectDatabaseMetaData.getDefaultSchema().orElse(null);
PipelineCommonSQLBuilder pipelineSQLBuilder = new PipelineCommonSQLBuilder(targetDatabaseType);
Collection<String> createdSchemaNames = new HashSet<>();
- for (CreateTableEntry each : createTableConfig.getCreateTableEntries()) {
+ for (CreateTableConfiguration each : createTableConfigs) {
String targetSchemaName = each.getTargetName().getSchemaName().toString();
if (null == targetSchemaName || targetSchemaName.equalsIgnoreCase(defaultSchema) || createdSchemaNames.contains(targetSchemaName)) {
continue;
@@ -113,13 +112,13 @@ public abstract class AbstractDataSourcePreparer implements DataSourcePreparer {
return PATTERN_CREATE_TABLE.matcher(createTableSQL).replaceFirst("CREATE TABLE IF NOT EXISTS ");
}
- protected final String getCreateTargetTableSQL(final CreateTableEntry createTableEntry, final PipelineDataSourceManager dataSourceManager,
+ protected final String getCreateTargetTableSQL(final CreateTableConfiguration createTableConfig, final PipelineDataSourceManager dataSourceManager,
final SQLParserEngine sqlParserEngine) throws SQLException {
- DatabaseType databaseType = createTableEntry.getSourceDataSourceConfig().getDatabaseType();
- DataSource sourceDataSource = dataSourceManager.getDataSource(createTableEntry.getSourceDataSourceConfig());
- String schemaName = createTableEntry.getSourceName().getSchemaName().toString();
- String sourceTableName = createTableEntry.getSourceName().getTableName().toString();
- String targetTableName = createTableEntry.getTargetName().getTableName().toString();
+ DatabaseType databaseType = createTableConfig.getSourceDataSourceConfig().getDatabaseType();
+ DataSource sourceDataSource = dataSourceManager.getDataSource(createTableConfig.getSourceDataSourceConfig());
+ String schemaName = createTableConfig.getSourceName().getSchemaName().toString();
+ String sourceTableName = createTableConfig.getSourceName().getTableName().toString();
+ String targetTableName = createTableConfig.getTargetName().getTableName().toString();
PipelineDDLGenerator generator = new PipelineDDLGenerator();
return generator.generateLogicDDL(databaseType, sourceDataSource, schemaName, sourceTableName, targetTableName, sqlParserEngine);
}
diff --git a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/preparer/datasource/PrepareTargetSchemasParameter.java b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/preparer/datasource/PrepareTargetSchemasParameter.java
index b2ce2ea0d84..c65cbcf854a 100644
--- a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/preparer/datasource/PrepareTargetSchemasParameter.java
+++ b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/preparer/datasource/PrepareTargetSchemasParameter.java
@@ -23,6 +23,8 @@ import org.apache.shardingsphere.data.pipeline.common.config.CreateTableConfigur
import org.apache.shardingsphere.data.pipeline.common.datasource.PipelineDataSourceManager;
import org.apache.shardingsphere.infra.database.core.type.DatabaseType;
+import java.util.Collection;
+
/**
* Prepare target schemas parameter.
*/
@@ -32,7 +34,7 @@ public final class PrepareTargetSchemasParameter {
private final DatabaseType targetDatabaseType;
- private final CreateTableConfiguration createTableConfig;
+ private final Collection<CreateTableConfiguration> createTableConfigurations;
private final PipelineDataSourceManager dataSourceManager;
}
diff --git a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/preparer/datasource/PrepareTargetTablesParameter.java b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/preparer/datasource/PrepareTargetTablesParameter.java
index e1ef95e9563..8702f24d428 100644
--- a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/preparer/datasource/PrepareTargetTablesParameter.java
+++ b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/preparer/datasource/PrepareTargetTablesParameter.java
@@ -23,6 +23,8 @@ import org.apache.shardingsphere.data.pipeline.common.config.CreateTableConfigur
import org.apache.shardingsphere.data.pipeline.common.datasource.PipelineDataSourceManager;
import org.apache.shardingsphere.infra.parser.SQLParserEngine;
+import java.util.Collection;
+
/**
* Prepare target tables parameter.
*/
@@ -30,7 +32,7 @@ import org.apache.shardingsphere.infra.parser.SQLParserEngine;
@Getter
public final class PrepareTargetTablesParameter {
- private final CreateTableConfiguration createTableConfig;
+ private final Collection<CreateTableConfiguration> createTableConfigurations;
private final PipelineDataSourceManager dataSourceManager;
diff --git a/kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/prepare/datasource/MySQLDataSourcePreparer.java b/kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/prepare/datasource/MySQLDataSourcePreparer.java
index 24f967a3302..240d76ae6ec 100644
--- a/kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/prepare/datasource/MySQLDataSourcePreparer.java
+++ b/kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/prepare/datasource/MySQLDataSourcePreparer.java
@@ -17,7 +17,7 @@
package org.apache.shardingsphere.data.pipeline.mysql.prepare.datasource;
-import org.apache.shardingsphere.data.pipeline.common.config.CreateTableConfiguration.CreateTableEntry;
+import org.apache.shardingsphere.data.pipeline.common.config.CreateTableConfiguration;
import org.apache.shardingsphere.data.pipeline.common.datasource.PipelineDataSourceManager;
import org.apache.shardingsphere.data.pipeline.core.preparer.datasource.AbstractDataSourcePreparer;
import org.apache.shardingsphere.data.pipeline.core.preparer.datasource.PrepareTargetTablesParameter;
@@ -33,7 +33,7 @@ public final class MySQLDataSourcePreparer extends AbstractDataSourcePreparer {
@Override
public void prepareTargetTables(final PrepareTargetTablesParameter param) throws SQLException {
PipelineDataSourceManager dataSourceManager = param.getDataSourceManager();
- for (CreateTableEntry each : param.getCreateTableConfig().getCreateTableEntries()) {
+ for (CreateTableConfiguration each : param.getCreateTableConfigurations()) {
String createTargetTableSQL = getCreateTargetTableSQL(each, dataSourceManager, param.getSqlParserEngine());
try (Connection targetConnection = getCachedDataSource(dataSourceManager, each.getTargetDataSourceConfig()).getConnection()) {
executeTargetTableSQL(targetConnection, addIfNotExistsForCreateTableSQL(createTargetTableSQL));
diff --git a/kernel/data-pipeline/dialect/opengauss/src/main/java/org/apache/shardingsphere/data/pipeline/opengauss/prepare/datasource/OpenGaussDataSourcePreparer.java b/kernel/data-pipeline/dialect/opengauss/src/main/java/org/apache/shardingsphere/data/pipeline/opengauss/prepare/datasource/OpenGaussDataSourcePreparer.java
index ba76a9b7213..3eadb66e501 100644
--- a/kernel/data-pipeline/dialect/opengauss/src/main/java/org/apache/shardingsphere/data/pipeline/opengauss/prepare/datasource/OpenGaussDataSourcePreparer.java
+++ b/kernel/data-pipeline/dialect/opengauss/src/main/java/org/apache/shardingsphere/data/pipeline/opengauss/prepare/datasource/OpenGaussDataSourcePreparer.java
@@ -19,7 +19,7 @@ package org.apache.shardingsphere.data.pipeline.opengauss.prepare.datasource;
import com.google.common.base.Splitter;
import lombok.extern.slf4j.Slf4j;
-import org.apache.shardingsphere.data.pipeline.common.config.CreateTableConfiguration.CreateTableEntry;
+import org.apache.shardingsphere.data.pipeline.common.config.CreateTableConfiguration;
import org.apache.shardingsphere.data.pipeline.common.datasource.PipelineDataSourceManager;
import org.apache.shardingsphere.data.pipeline.core.preparer.datasource.AbstractDataSourcePreparer;
import org.apache.shardingsphere.data.pipeline.core.preparer.datasource.PrepareTargetSchemasParameter;
@@ -50,7 +50,7 @@ public final class OpenGaussDataSourcePreparer extends AbstractDataSourcePrepare
@Override
public void prepareTargetTables(final PrepareTargetTablesParameter param) throws SQLException {
PipelineDataSourceManager dataSourceManager = param.getDataSourceManager();
- for (CreateTableEntry each : param.getCreateTableConfig().getCreateTableEntries()) {
+ for (CreateTableConfiguration each : param.getCreateTableConfigurations()) {
String createTargetTableSQL = getCreateTargetTableSQL(each, dataSourceManager, param.getSqlParserEngine());
try (Connection targetConnection = getCachedDataSource(dataSourceManager, each.getTargetDataSourceConfig()).getConnection()) {
for (String sql : Splitter.on(";").trimResults().omitEmptyStrings().splitToList(createTargetTableSQL)) {
diff --git a/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/prepare/datasource/PostgreSQLDataSourcePreparer.java b/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/prepare/datasource/PostgreSQLDataSourcePreparer.java
index 10c0c05dda6..c0d1398036b 100644
--- a/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/prepare/datasource/PostgreSQLDataSourcePreparer.java
+++ b/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/prepare/datasource/PostgreSQLDataSourcePreparer.java
@@ -18,7 +18,7 @@
package org.apache.shardingsphere.data.pipeline.postgresql.prepare.datasource;
import com.google.common.base.Splitter;
-import org.apache.shardingsphere.data.pipeline.common.config.CreateTableConfiguration.CreateTableEntry;
+import org.apache.shardingsphere.data.pipeline.common.config.CreateTableConfiguration;
import org.apache.shardingsphere.data.pipeline.common.datasource.PipelineDataSourceManager;
import org.apache.shardingsphere.data.pipeline.core.preparer.datasource.AbstractDataSourcePreparer;
import org.apache.shardingsphere.data.pipeline.core.preparer.datasource.PrepareTargetTablesParameter;
@@ -34,7 +34,7 @@ public final class PostgreSQLDataSourcePreparer extends AbstractDataSourcePrepar
@Override
public void prepareTargetTables(final PrepareTargetTablesParameter param) throws SQLException {
PipelineDataSourceManager dataSourceManager = param.getDataSourceManager();
- for (CreateTableEntry each : param.getCreateTableConfig().getCreateTableEntries()) {
+ for (CreateTableConfiguration each : param.getCreateTableConfigurations()) {
String createTargetTableSQL = getCreateTargetTableSQL(each, dataSourceManager, param.getSqlParserEngine());
try (Connection targetConnection = getCachedDataSource(dataSourceManager, each.getTargetDataSourceConfig()).getConnection()) {
for (String sql : Splitter.on(";").trimResults().omitEmptyStrings().splitToList(createTargetTableSQL)) {
diff --git a/kernel/data-pipeline/scenario/migration/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/api/impl/MigrationJobAPI.java b/kernel/data-pipeline/scenario/migration/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/api/impl/MigrationJobAPI.java
index f58330485e6..bd5853ac5bd 100644
--- a/kernel/data-pipeline/scenario/migration/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/api/impl/MigrationJobAPI.java
+++ b/kernel/data-pipeline/scenario/migration/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/api/impl/MigrationJobAPI.java
@@ -23,7 +23,6 @@ import org.apache.shardingsphere.data.pipeline.api.PipelineDataSourceConfigurati
import org.apache.shardingsphere.data.pipeline.api.type.ShardingSpherePipelineDataSourceConfiguration;
import org.apache.shardingsphere.data.pipeline.api.type.StandardPipelineDataSourceConfiguration;
import org.apache.shardingsphere.data.pipeline.common.config.CreateTableConfiguration;
-import org.apache.shardingsphere.data.pipeline.common.config.CreateTableConfiguration.CreateTableEntry;
import org.apache.shardingsphere.data.pipeline.common.config.ImporterConfiguration;
import org.apache.shardingsphere.data.pipeline.common.config.job.PipelineJobConfiguration;
import org.apache.shardingsphere.data.pipeline.common.config.job.yaml.YamlPipelineJobConfiguration;
@@ -264,34 +263,32 @@ public final class MigrationJobAPI extends AbstractInventoryIncrementalJobAPIImp
MigrationJobConfiguration jobConfig = (MigrationJobConfiguration) pipelineJobConfig;
IncrementalDumperContext incrementalDumperContext = new MigrationIncrementalDumperContextCreator(
jobConfig).createDumperContext(jobConfig.getJobShardingDataNodes().get(jobShardingItem));
- CreateTableConfiguration createTableConfig = buildCreateTableConfiguration(jobConfig, incrementalDumperContext.getCommonContext().getTableAndSchemaNameMapper());
+ Collection<CreateTableConfiguration> createTableConfigs = buildCreateTableConfigurations(jobConfig, incrementalDumperContext.getCommonContext().getTableAndSchemaNameMapper());
Set<CaseInsensitiveIdentifier> targetTableNames = jobConfig.getTargetTableNames().stream().map(CaseInsensitiveIdentifier::new).collect(Collectors.toSet());
Map<CaseInsensitiveIdentifier, Set<String>> shardingColumnsMap = new ShardingColumnsExtractor().getShardingColumnsMap(
((ShardingSpherePipelineDataSourceConfiguration) jobConfig.getTarget()).getRootConfig().getRules(), targetTableNames);
ImporterConfiguration importerConfig = buildImporterConfiguration(
jobConfig, pipelineProcessConfig, shardingColumnsMap, incrementalDumperContext.getCommonContext().getTableAndSchemaNameMapper());
MigrationTaskConfiguration result = new MigrationTaskConfiguration(
- incrementalDumperContext.getCommonContext().getDataSourceName(), createTableConfig, incrementalDumperContext, importerConfig);
+ incrementalDumperContext.getCommonContext().getDataSourceName(), createTableConfigs, incrementalDumperContext, importerConfig);
log.info("buildTaskConfiguration, result={}", result);
return result;
}
- private CreateTableConfiguration buildCreateTableConfiguration(final MigrationJobConfiguration jobConfig,
- final TableAndSchemaNameMapper tableAndSchemaNameMapper) {
- Collection<CreateTableEntry> createTableEntries = new LinkedList<>();
+ private Collection<CreateTableConfiguration> buildCreateTableConfigurations(final MigrationJobConfiguration jobConfig, final TableAndSchemaNameMapper tableAndSchemaNameMapper) {
+ Collection<CreateTableConfiguration> result = new LinkedList<>();
for (JobDataNodeEntry each : jobConfig.getTablesFirstDataNodes().getEntries()) {
String sourceSchemaName = tableAndSchemaNameMapper.getSchemaName(each.getLogicTableName());
DialectDatabaseMetaData dialectDatabaseMetaData = new DatabaseTypeRegistry(jobConfig.getTargetDatabaseType()).getDialectDatabaseMetaData();
String targetSchemaName = dialectDatabaseMetaData.isSchemaAvailable() ? sourceSchemaName : null;
DataNode dataNode = each.getDataNodes().get(0);
PipelineDataSourceConfiguration sourceDataSourceConfig = jobConfig.getSources().get(dataNode.getDataSourceName());
- CreateTableEntry createTableEntry = new CreateTableEntry(
+ CreateTableConfiguration createTableConfig = new CreateTableConfiguration(
sourceDataSourceConfig, new SchemaTableName(sourceSchemaName, dataNode.getTableName()),
jobConfig.getTarget(), new SchemaTableName(targetSchemaName, each.getLogicTableName()));
- createTableEntries.add(createTableEntry);
+ result.add(createTableConfig);
}
- CreateTableConfiguration result = new CreateTableConfiguration(createTableEntries);
- log.info("getCreateTableConfiguration, result={}", result);
+ log.info("buildCreateTableConfigurations, result={}", result);
return result;
}
diff --git a/kernel/data-pipeline/scenario/migration/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/config/MigrationTaskConfiguration.java b/kernel/data-pipeline/scenario/migration/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/config/MigrationTaskConfiguration.java
index 9b273a62756..ac72f5ed12b 100644
--- a/kernel/data-pipeline/scenario/migration/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/config/MigrationTaskConfiguration.java
+++ b/kernel/data-pipeline/scenario/migration/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/config/MigrationTaskConfiguration.java
@@ -20,10 +20,12 @@ package org.apache.shardingsphere.data.pipeline.scenario.migration.config;
import lombok.Getter;
import lombok.RequiredArgsConstructor;
import lombok.ToString;
-import org.apache.shardingsphere.data.pipeline.core.ingest.dumper.context.IncrementalDumperContext;
import org.apache.shardingsphere.data.pipeline.common.config.CreateTableConfiguration;
import org.apache.shardingsphere.data.pipeline.common.config.ImporterConfiguration;
import org.apache.shardingsphere.data.pipeline.common.config.PipelineTaskConfiguration;
+import org.apache.shardingsphere.data.pipeline.core.ingest.dumper.context.IncrementalDumperContext;
+
+import java.util.Collection;
/**
* Migration task configuration.
@@ -35,7 +37,7 @@ public final class MigrationTaskConfiguration implements PipelineTaskConfigurati
private final String dataSourceName;
- private final CreateTableConfiguration createTableConfig;
+ private final Collection<CreateTableConfiguration> createTableConfigurations;
private final IncrementalDumperContext dumperContext;
diff --git a/kernel/data-pipeline/scenario/migration/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/prepare/MigrationJobPreparer.java b/kernel/data-pipeline/scenario/migration/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/prepare/MigrationJobPreparer.java
index 9b0a3a66ec1..e6ef884220a 100644
--- a/kernel/data-pipeline/scenario/migration/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/prepare/MigrationJobPreparer.java
+++ b/kernel/data-pipeline/scenario/migration/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/prepare/MigrationJobPreparer.java
@@ -18,13 +18,8 @@
package org.apache.shardingsphere.data.pipeline.scenario.migration.prepare;
import lombok.extern.slf4j.Slf4j;
-import org.apache.shardingsphere.data.pipeline.core.ingest.dumper.context.IncrementalDumperContext;
-import org.apache.shardingsphere.data.pipeline.core.ingest.dumper.context.InventoryDumperContext;
import org.apache.shardingsphere.data.pipeline.api.PipelineDataSourceConfiguration;
import org.apache.shardingsphere.data.pipeline.api.type.StandardPipelineDataSourceConfiguration;
-import org.apache.shardingsphere.data.pipeline.core.ingest.channel.PipelineChannel;
-import org.apache.shardingsphere.data.pipeline.core.ingest.dumper.Dumper;
-import org.apache.shardingsphere.data.pipeline.common.metadata.loader.PipelineTableMetaDataLoader;
import org.apache.shardingsphere.data.pipeline.common.config.CreateTableConfiguration;
import org.apache.shardingsphere.data.pipeline.common.config.ImporterConfiguration;
import org.apache.shardingsphere.data.pipeline.common.context.PipelineContextManager;
@@ -37,11 +32,17 @@ import org.apache.shardingsphere.data.pipeline.common.job.progress.InventoryIncr
import org.apache.shardingsphere.data.pipeline.common.job.progress.JobItemIncrementalTasksProgress;
import org.apache.shardingsphere.data.pipeline.common.job.progress.JobOffsetInfo;
import org.apache.shardingsphere.data.pipeline.common.job.progress.listener.PipelineJobProgressListener;
+import org.apache.shardingsphere.data.pipeline.common.metadata.loader.PipelineTableMetaDataLoader;
+import org.apache.shardingsphere.data.pipeline.common.spi.ingest.dumper.IncrementalDumperCreator;
import org.apache.shardingsphere.data.pipeline.common.task.progress.IncrementalTaskProgress;
import org.apache.shardingsphere.data.pipeline.core.exception.job.PrepareJobWithGetBinlogPositionException;
import org.apache.shardingsphere.data.pipeline.core.importer.Importer;
import org.apache.shardingsphere.data.pipeline.core.importer.SingleChannelConsumerImporter;
import org.apache.shardingsphere.data.pipeline.core.importer.sink.PipelineSink;
+import org.apache.shardingsphere.data.pipeline.core.ingest.channel.PipelineChannel;
+import org.apache.shardingsphere.data.pipeline.core.ingest.dumper.Dumper;
+import org.apache.shardingsphere.data.pipeline.core.ingest.dumper.context.IncrementalDumperContext;
+import org.apache.shardingsphere.data.pipeline.core.ingest.dumper.context.InventoryDumperContext;
import org.apache.shardingsphere.data.pipeline.core.job.PipelineJobCenter;
import org.apache.shardingsphere.data.pipeline.core.job.PipelineJobIdUtils;
import org.apache.shardingsphere.data.pipeline.core.preparer.InventoryTaskSplitter;
@@ -55,15 +56,14 @@ import org.apache.shardingsphere.data.pipeline.scenario.migration.api.impl.Migra
import org.apache.shardingsphere.data.pipeline.scenario.migration.config.MigrationJobConfiguration;
import org.apache.shardingsphere.data.pipeline.scenario.migration.config.MigrationTaskConfiguration;
import org.apache.shardingsphere.data.pipeline.scenario.migration.context.MigrationJobItemContext;
-import org.apache.shardingsphere.data.pipeline.common.spi.ingest.dumper.IncrementalDumperCreator;
+import org.apache.shardingsphere.infra.database.core.spi.DatabaseTypedSPILoader;
+import org.apache.shardingsphere.infra.exception.core.ShardingSpherePreconditions;
+import org.apache.shardingsphere.infra.exception.core.external.sql.type.generic.UnsupportedSQLOperationException;
import org.apache.shardingsphere.infra.lock.GlobalLockNames;
import org.apache.shardingsphere.infra.lock.LockContext;
import org.apache.shardingsphere.infra.lock.LockDefinition;
import org.apache.shardingsphere.infra.metadata.ShardingSphereMetaData;
import org.apache.shardingsphere.infra.parser.SQLParserEngine;
-import org.apache.shardingsphere.infra.database.core.spi.DatabaseTypedSPILoader;
-import org.apache.shardingsphere.infra.exception.core.ShardingSpherePreconditions;
-import org.apache.shardingsphere.infra.exception.core.external.sql.type.generic.UnsupportedSQLOperationException;
import org.apache.shardingsphere.mode.lock.GlobalLockDefinition;
import java.sql.SQLException;
@@ -157,13 +157,13 @@ public final class MigrationJobPreparer {
private void prepareTarget(final MigrationJobItemContext jobItemContext) throws SQLException {
MigrationJobConfiguration jobConfig = jobItemContext.getJobConfig();
- CreateTableConfiguration createTableConfig = jobItemContext.getTaskConfig().getCreateTableConfig();
+ Collection<CreateTableConfiguration> createTableConfigs = jobItemContext.getTaskConfig().getCreateTableConfigurations();
PipelineDataSourceManager dataSourceManager = jobItemContext.getDataSourceManager();
- PrepareTargetSchemasParameter prepareTargetSchemasParam = new PrepareTargetSchemasParameter(jobItemContext.getJobConfig().getTargetDatabaseType(), createTableConfig, dataSourceManager);
+ PrepareTargetSchemasParameter prepareTargetSchemasParam = new PrepareTargetSchemasParameter(jobItemContext.getJobConfig().getTargetDatabaseType(), createTableConfigs, dataSourceManager);
PipelineJobPreparerUtils.prepareTargetSchema(jobItemContext.getJobConfig().getTargetDatabaseType(), prepareTargetSchemasParam);
ShardingSphereMetaData metaData = PipelineContextManager.getContext(PipelineJobIdUtils.parseContextKey(jobConfig.getJobId())).getContextManager().getMetaDataContexts().getMetaData();
SQLParserEngine sqlParserEngine = PipelineJobPreparerUtils.getSQLParserEngine(metaData, jobConfig.getTargetDatabaseName());
- PipelineJobPreparerUtils.prepareTargetTables(jobItemContext.getJobConfig().getTargetDatabaseType(), new PrepareTargetTablesParameter(createTableConfig, dataSourceManager, sqlParserEngine));
+ PipelineJobPreparerUtils.prepareTargetTables(jobItemContext.getJobConfig().getTargetDatabaseType(), new PrepareTargetTablesParameter(createTableConfigs, dataSourceManager, sqlParserEngine));
}
private void prepareIncremental(final MigrationJobItemContext jobItemContext) {
diff --git a/test/it/pipeline/src/test/java/org/apache/shardingsphere/test/it/data/pipeline/core/fixture/H2DataSourcePreparer.java b/test/it/pipeline/src/test/java/org/apache/shardingsphere/test/it/data/pipeline/core/fixture/H2DataSourcePreparer.java
index 820803b38b0..2503c95f555 100644
--- a/test/it/pipeline/src/test/java/org/apache/shardingsphere/test/it/data/pipeline/core/fixture/H2DataSourcePreparer.java
+++ b/test/it/pipeline/src/test/java/org/apache/shardingsphere/test/it/data/pipeline/core/fixture/H2DataSourcePreparer.java
@@ -17,7 +17,7 @@
package org.apache.shardingsphere.test.it.data.pipeline.core.fixture;
-import org.apache.shardingsphere.data.pipeline.common.config.CreateTableConfiguration.CreateTableEntry;
+import org.apache.shardingsphere.data.pipeline.common.config.CreateTableConfiguration;
import org.apache.shardingsphere.data.pipeline.common.datasource.PipelineDataSourceManager;
import org.apache.shardingsphere.data.pipeline.core.preparer.datasource.AbstractDataSourcePreparer;
import org.apache.shardingsphere.data.pipeline.core.preparer.datasource.PrepareTargetTablesParameter;
@@ -33,7 +33,7 @@ public final class H2DataSourcePreparer extends AbstractDataSourcePreparer {
@Override
public void prepareTargetTables(final PrepareTargetTablesParameter param) throws SQLException {
PipelineDataSourceManager dataSourceManager = param.getDataSourceManager();
- for (CreateTableEntry each : param.getCreateTableConfig().getCreateTableEntries()) {
+ for (CreateTableConfiguration each : param.getCreateTableConfigurations()) {
String createTargetTableSQL = getCreateTargetTableSQL(each, dataSourceManager, param.getSqlParserEngine());
try (Connection targetConnection = getCachedDataSource(dataSourceManager, each.getTargetDataSourceConfig()).getConnection()) {
executeTargetTableSQL(targetConnection, addIfNotExistsForCreateTableSQL(createTargetTableSQL));