You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@shardingsphere.apache.org by zh...@apache.org on 2022/08/18 04:48:15 UTC
[shardingsphere] branch master updated: Improve DDL genereate method at migration prepare (#20245)
This is an automated email from the ASF dual-hosted git repository.
zhonghongsheng 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 3b15d531953 Improve DDL genereate method at migration prepare (#20245)
3b15d531953 is described below
commit 3b15d53195316122e703b98c0476a4863326732f
Author: Xinze Guo <10...@users.noreply.github.com>
AuthorDate: Thu Aug 18 12:48:04 2022 +0800
Improve DDL genereate method at migration prepare (#20245)
* Revise #20207
* Improve parameter name
* Fix codestyle
---
.../metadata/generator/PipelineDDLGenerator.java | 61 +++++++++++-----------
.../datasource/AbstractDataSourcePreparer.java | 33 +++++++-----
.../datasource/PrepareTargetTablesParameter.java | 26 ++++++---
.../scenario/migration/MigrationJobPreparer.java | 32 +++++++++++-
.../datasource/MySQLDataSourcePreparer.java | 31 +----------
.../datasource/OpenGaussDataSourcePreparer.java | 26 +--------
.../datasource/PostgreSQLDataSourcePreparer.java | 27 +---------
7 files changed, 106 insertions(+), 130 deletions(-)
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/metadata/generator/PipelineDDLGenerator.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/metadata/generator/PipelineDDLGenerator.java
index ca07db51c9a..812dc1bbced 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/metadata/generator/PipelineDDLGenerator.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/metadata/generator/PipelineDDLGenerator.java
@@ -17,10 +17,10 @@
package org.apache.shardingsphere.data.pipeline.core.metadata.generator;
-import javax.sql.DataSource;
import lombok.RequiredArgsConstructor;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
+import org.apache.shardingsphere.data.pipeline.spi.ddlgenerator.CreateTableSQLGenerator;
import org.apache.shardingsphere.data.pipeline.spi.ddlgenerator.CreateTableSQLGeneratorFactory;
import org.apache.shardingsphere.infra.binder.LogicSQL;
import org.apache.shardingsphere.infra.binder.SQLStatementContextFactory;
@@ -33,11 +33,11 @@ import org.apache.shardingsphere.infra.binder.type.ConstraintAvailable;
import org.apache.shardingsphere.infra.binder.type.IndexAvailable;
import org.apache.shardingsphere.infra.binder.type.TableAvailable;
import org.apache.shardingsphere.infra.database.type.DatabaseType;
-import org.apache.shardingsphere.infra.exception.ShardingSphereException;
import org.apache.shardingsphere.infra.metadata.database.ShardingSphereDatabase;
import org.apache.shardingsphere.infra.metadata.database.schema.util.IndexMetaDataUtil;
import org.apache.shardingsphere.infra.parser.ShardingSphereSQLParserEngine;
import org.apache.shardingsphere.infra.rule.identifier.type.DataNodeContainedRule;
+import org.apache.shardingsphere.infra.util.spi.exception.ServiceProviderNotFoundException;
import org.apache.shardingsphere.sql.parser.sql.common.segment.SQLSegment;
import org.apache.shardingsphere.sql.parser.sql.common.segment.ddl.constraint.ConstraintSegment;
import org.apache.shardingsphere.sql.parser.sql.common.segment.ddl.index.IndexSegment;
@@ -45,6 +45,8 @@ import org.apache.shardingsphere.sql.parser.sql.common.segment.generic.table.Sim
import org.apache.shardingsphere.sql.parser.sql.common.segment.generic.table.TableNameSegment;
import org.apache.shardingsphere.sql.parser.sql.common.statement.SQLStatement;
+import javax.sql.DataSource;
+import java.sql.Connection;
import java.sql.SQLException;
import java.util.Collection;
import java.util.Collections;
@@ -69,8 +71,8 @@ public final class PipelineDDLGenerator {
/**
* Generate logic ddl sql.
*
- * @param database database
- * @param dataSourceName data source name
+ * @param sourceDataSource source data source
+ * @param databaseType database type
* @param schemaName schema name
* @param logicTableName table name
* @param actualTableName actual table name
@@ -78,16 +80,13 @@ public final class PipelineDDLGenerator {
* @return ddl SQL
*/
@SneakyThrows
- public String generateLogicDDLSQL(final ShardingSphereDatabase database, final String dataSourceName, final String schemaName,
- final String logicTableName, final String actualTableName,
+ public String generateLogicDDLSQL(final DataSource sourceDataSource, final DatabaseType databaseType, final String schemaName, final String logicTableName, final String actualTableName,
final ShardingSphereSQLParserEngine parserEngine) {
- DatabaseType databaseType = database.getProtocolType();
- log.info("generateLogicDDLSQL, databaseType={}, databaseName={}, schemaName={}, tableName={}, dataSourceNames={}",
- databaseType.getType(), database.getName(), schemaName, logicTableName, database.getResource().getDataSources().keySet());
- Collection<String> multiSQL = generateActualDDLSQL(databaseType, schemaName, actualTableName, database.getResource().getDataSources().get(dataSourceName));
+ log.info("generateLogicDDLSQL, databaseType={}, schemaName={}, tableName={}", databaseType.getType(), schemaName, logicTableName);
+ Collection<String> multiSQL = generateActualDDLSQL(databaseType, schemaName, actualTableName, sourceDataSource);
StringBuilder result = new StringBuilder();
for (String each : multiSQL) {
- Optional<String> logicSQL = decorate(databaseType, database.getName(), schemaName, database, each, parserEngine);
+ Optional<String> logicSQL = decorate(databaseType, schemaName, sourceDataSource, each, logicTableName, parserEngine);
logicSQL.ifPresent(ddlSQL -> result.append(ddlSQL).append(DELIMITER).append(System.lineSeparator()));
}
return result.toString();
@@ -118,12 +117,16 @@ public final class PipelineDDLGenerator {
return sql;
}
- private Optional<String> decorate(final DatabaseType databaseType, final String databaseName, final String schemaName, final ShardingSphereDatabase database, final String sql,
- final ShardingSphereSQLParserEngine parserEngine) {
+ private Optional<String> decorate(final DatabaseType databaseType, final String schemaName, final DataSource dataSource, final String sql, final String logicTableName,
+ final ShardingSphereSQLParserEngine parserEngine) throws SQLException {
if (sql.trim().isEmpty()) {
return Optional.empty();
}
- String result = decorateActualSQL(sql.trim(), database, databaseName, parserEngine);
+ String databaseName;
+ try (Connection connection = dataSource.getConnection()) {
+ databaseName = connection.getCatalog();
+ }
+ String result = decorateActualSQL(sql.trim(), logicTableName, databaseName, parserEngine);
// TODO remove it after set search_path is supported.
if ("openGauss".equals(databaseType.getType())) {
return decorateOpenGauss(databaseName, schemaName, result, parserEngine);
@@ -132,39 +135,38 @@ public final class PipelineDDLGenerator {
}
private Collection<String> generateActualDDLSQL(final DatabaseType databaseType, final String schemaName, final String actualTable, final DataSource dataSource) throws SQLException {
- return CreateTableSQLGeneratorFactory.findInstance(databaseType).orElseThrow(() -> new ShardingSphereException("Failed to get dialect ddl sql generator"))
+ return CreateTableSQLGeneratorFactory.findInstance(databaseType).orElseThrow(() -> new ServiceProviderNotFoundException(CreateTableSQLGenerator.class, databaseType.getType()))
.generate(actualTable, schemaName, dataSource);
}
- private String decorateActualSQL(final String sql, final ShardingSphereDatabase database, final String databaseName, final ShardingSphereSQLParserEngine parserEngine) {
+ private String decorateActualSQL(final String sql, final String logicTableName, final String databaseName, final ShardingSphereSQLParserEngine parserEngine) {
LogicSQL logicSQL = getLogicSQL(sql, databaseName, parserEngine);
SQLStatementContext<?> sqlStatementContext = logicSQL.getSqlStatementContext();
Map<SQLSegment, String> replaceMap = new TreeMap<>(Comparator.comparing(SQLSegment::getStartIndex));
if (sqlStatementContext instanceof CreateTableStatementContext) {
- appendFromIndexAndConstraint(replaceMap, database, sqlStatementContext);
- appendFromTable(replaceMap, database, (TableAvailable) sqlStatementContext);
+ appendFromIndexAndConstraint(replaceMap, logicTableName, sqlStatementContext);
+ appendFromTable(replaceMap, logicTableName, (TableAvailable) sqlStatementContext);
}
if (sqlStatementContext instanceof CommentStatementContext) {
- appendFromTable(replaceMap, database, (TableAvailable) sqlStatementContext);
+ appendFromTable(replaceMap, logicTableName, (TableAvailable) sqlStatementContext);
}
if (sqlStatementContext instanceof CreateIndexStatementContext) {
- appendFromTable(replaceMap, database, (TableAvailable) sqlStatementContext);
- appendFromIndexAndConstraint(replaceMap, database, sqlStatementContext);
+ appendFromTable(replaceMap, logicTableName, (TableAvailable) sqlStatementContext);
+ appendFromIndexAndConstraint(replaceMap, logicTableName, sqlStatementContext);
}
if (sqlStatementContext instanceof AlterTableStatementContext) {
- appendFromIndexAndConstraint(replaceMap, database, sqlStatementContext);
- appendFromTable(replaceMap, database, (TableAvailable) sqlStatementContext);
+ appendFromIndexAndConstraint(replaceMap, logicTableName, sqlStatementContext);
+ appendFromTable(replaceMap, logicTableName, (TableAvailable) sqlStatementContext);
}
return doDecorateActualTable(replaceMap, sql);
}
- private void appendFromIndexAndConstraint(final Map<SQLSegment, String> replaceMap, final ShardingSphereDatabase database, final SQLStatementContext<?> sqlStatementContext) {
+ private void appendFromIndexAndConstraint(final Map<SQLSegment, String> replaceMap, final String logicTableName, final SQLStatementContext<?> sqlStatementContext) {
if (!(sqlStatementContext instanceof TableAvailable) || ((TableAvailable) sqlStatementContext).getTablesContext().getTables().isEmpty()) {
return;
}
TableNameSegment tableNameSegment = ((TableAvailable) sqlStatementContext).getTablesContext().getTables().iterator().next().getTableName();
- String logicTable = findLogicTable(tableNameSegment, database);
- if (!tableNameSegment.getIdentifier().getValue().equals(logicTable)) {
+ if (!tableNameSegment.getIdentifier().getValue().equals(logicTableName)) {
if (sqlStatementContext instanceof IndexAvailable) {
for (IndexSegment each : ((IndexAvailable) sqlStatementContext).getIndexes()) {
String logicIndexName = IndexMetaDataUtil.getLogicIndexName(each.getIndexName().getIdentifier().getValue(), tableNameSegment.getIdentifier().getValue());
@@ -180,11 +182,10 @@ public final class PipelineDDLGenerator {
}
}
- private void appendFromTable(final Map<SQLSegment, String> replaceMap, final ShardingSphereDatabase database, final TableAvailable sqlStatementContext) {
+ private void appendFromTable(final Map<SQLSegment, String> replaceMap, final String logicTableName, final TableAvailable sqlStatementContext) {
for (SimpleTableSegment each : sqlStatementContext.getAllTables()) {
- String logicTable = findLogicTable(each.getTableName(), database);
- if (!logicTable.equals(each.getTableName().getIdentifier().getValue())) {
- replaceMap.put(each.getTableName(), logicTable);
+ if (!logicTableName.equals(each.getTableName().getIdentifier().getValue())) {
+ replaceMap.put(each.getTableName(), logicTableName);
}
}
}
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/prepare/datasource/AbstractDataSourcePreparer.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/prepare/datasource/AbstractDataSourcePreparer.java
index 15cf5495073..191f9a8bc3f 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/prepare/datasource/AbstractDataSourcePreparer.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/prepare/datasource/AbstractDataSourcePreparer.java
@@ -17,27 +17,30 @@
package org.apache.shardingsphere.data.pipeline.core.prepare.datasource;
-import java.util.Optional;
import lombok.extern.slf4j.Slf4j;
import org.apache.shardingsphere.data.pipeline.api.config.job.MigrationJobConfiguration;
+import org.apache.shardingsphere.data.pipeline.api.datanode.JobDataNodeEntry;
import org.apache.shardingsphere.data.pipeline.api.datasource.PipelineDataSourceManager;
import org.apache.shardingsphere.data.pipeline.api.datasource.PipelineDataSourceWrapper;
import org.apache.shardingsphere.data.pipeline.api.datasource.config.PipelineDataSourceConfiguration;
import org.apache.shardingsphere.data.pipeline.api.datasource.config.PipelineDataSourceConfigurationFactory;
import org.apache.shardingsphere.data.pipeline.core.exception.PipelineJobPrepareFailedException;
+import org.apache.shardingsphere.data.pipeline.core.metadata.generator.PipelineDDLGenerator;
import org.apache.shardingsphere.data.pipeline.core.sqlbuilder.PipelineSQLBuilderFactory;
import org.apache.shardingsphere.data.pipeline.spi.sqlbuilder.PipelineSQLBuilder;
+import org.apache.shardingsphere.infra.database.type.DatabaseType;
import org.apache.shardingsphere.infra.database.type.DatabaseTypeEngine;
+import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.SQLException;
import java.sql.Statement;
+import java.util.Collections;
import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.List;
import java.util.Set;
import java.util.regex.Pattern;
-import org.apache.shardingsphere.infra.datanode.DataNode;
-import org.apache.shardingsphere.infra.datanode.DataNodes;
-import org.apache.shardingsphere.infra.metadata.database.ShardingSphereDatabase;
/**
* Abstract data source preparer.
@@ -57,7 +60,7 @@ public abstract class AbstractDataSourcePreparer implements DataSourcePreparer {
String defaultSchema = DatabaseTypeEngine.getDefaultSchemaName(parameter.getTargetDatabaseType(), parameter.getDatabaseName());
log.info("prepareTargetSchemas, schemaNames={}, defaultSchema={}", schemaNames, defaultSchema);
PipelineSQLBuilder pipelineSQLBuilder = PipelineSQLBuilderFactory.getInstance(parameter.getTargetDatabaseType().getType());
- try (Connection targetConnection = getTargetCachedDataSource(parameter.getDataSourceConfig(), parameter.getDataSourceManager()).getConnection()) {
+ try (Connection targetConnection = getCachedDataSource(parameter.getDataSourceConfig(), parameter.getDataSourceManager()).getConnection()) {
for (String each : schemaNames) {
if (each.equalsIgnoreCase(defaultSchema)) {
continue;
@@ -91,7 +94,7 @@ public abstract class AbstractDataSourcePreparer implements DataSourcePreparer {
return dataSourceManager.getDataSource(PipelineDataSourceConfigurationFactory.newInstance(jobConfig.getSource().getType(), jobConfig.getSource().getParameter()));
}
- protected final PipelineDataSourceWrapper getTargetCachedDataSource(final PipelineDataSourceConfiguration dataSourceConfig, final PipelineDataSourceManager dataSourceManager) {
+ protected final PipelineDataSourceWrapper getCachedDataSource(final PipelineDataSourceConfiguration dataSourceConfig, final PipelineDataSourceManager dataSourceManager) {
return dataSourceManager.getDataSource(dataSourceConfig);
}
@@ -116,11 +119,17 @@ public abstract class AbstractDataSourcePreparer implements DataSourcePreparer {
return PATTERN_CREATE_TABLE.matcher(createTableSQL).replaceFirst("CREATE TABLE IF NOT EXISTS ");
}
- protected String getActualTable(final ShardingSphereDatabase database, final String tableName) {
- DataNodes dataNodes = new DataNodes(database.getRuleMetaData().getRules());
- Optional<DataNode> filteredDataNode = dataNodes.getDataNodes(tableName).stream()
- .filter(each -> database.getResource().getDataSources().containsKey(each.getDataSourceName().contains(".") ? each.getDataSourceName().split("\\.")[0] : each.getDataSourceName()))
- .findFirst();
- return filteredDataNode.map(DataNode::getTableName).orElse(tableName);
+ protected List<String> listCreateLogicalTableSQL(final PrepareTargetTablesParameter parameter) {
+ PipelineDDLGenerator generator = new PipelineDDLGenerator();
+ List<String> result = new LinkedList<>();
+ for (JobDataNodeEntry each : parameter.getTablesFirstDataNodes().getEntries()) {
+ String schemaName = parameter.getTableNameSchemaNameMapping().getSchemaName(each.getLogicTableName());
+ String dataSourceName = each.getDataNodes().get(0).getDataSourceName();
+ DataSource dataSource = parameter.getSourceDataSourceMap().get(dataSourceName);
+ DatabaseType databaseType = DatabaseTypeEngine.getDatabaseType(Collections.singletonList(dataSource));
+ String actualTableName = parameter.getTableNameMap().get(each.getLogicTableName());
+ result.add(generator.generateLogicDDLSQL(dataSource, databaseType, schemaName, each.getLogicTableName(), actualTableName, parameter.getSqlParserEngine()));
+ }
+ return result;
}
}
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/prepare/datasource/PrepareTargetTablesParameter.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/prepare/datasource/PrepareTargetTablesParameter.java
index 9aaf034ba3a..ffda71fcb52 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/prepare/datasource/PrepareTargetTablesParameter.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/prepare/datasource/PrepareTargetTablesParameter.java
@@ -23,6 +23,10 @@ import org.apache.shardingsphere.data.pipeline.api.config.TableNameSchemaNameMap
import org.apache.shardingsphere.data.pipeline.api.datanode.JobDataNodeLine;
import org.apache.shardingsphere.data.pipeline.api.datasource.PipelineDataSourceManager;
import org.apache.shardingsphere.data.pipeline.api.datasource.config.PipelineDataSourceConfiguration;
+import org.apache.shardingsphere.infra.parser.ShardingSphereSQLParserEngine;
+
+import javax.sql.DataSource;
+import java.util.Map;
/**
* Prepare target tables parameter.
@@ -34,19 +38,29 @@ public final class PrepareTargetTablesParameter {
private final JobDataNodeLine tablesFirstDataNodes;
- private final PipelineDataSourceConfiguration dataSourceConfig;
+ private final PipelineDataSourceConfiguration targetDataSourceConfig;
+
+ private final Map<String, DataSource> sourceDataSourceMap;
private final PipelineDataSourceManager dataSourceManager;
+ private final Map<String, String> tableNameMap;
+
private final TableNameSchemaNameMapping tableNameSchemaNameMapping;
- public PrepareTargetTablesParameter(@NonNull final String databaseName, @NonNull final PipelineDataSourceConfiguration dataSourceConfig,
- @NonNull final PipelineDataSourceManager dataSourceManager,
- @NonNull final String tablesFirstDataNodes, final TableNameSchemaNameMapping tableNameSchemaNameMapping) {
+ private final ShardingSphereSQLParserEngine sqlParserEngine;
+
+ public PrepareTargetTablesParameter(@NonNull final String databaseName, @NonNull final PipelineDataSourceConfiguration targetDataSourceConfig,
+ @NonNull final Map<String, DataSource> sourceDataSourceMap, @NonNull final PipelineDataSourceManager dataSourceManager,
+ @NonNull final JobDataNodeLine tablesFirstDataNodes, final Map<String, String> tableNameMap, final TableNameSchemaNameMapping tableNameSchemaNameMapping,
+ @NonNull final ShardingSphereSQLParserEngine sqlParserEngine) {
this.databaseName = databaseName;
- this.dataSourceConfig = dataSourceConfig;
- this.tablesFirstDataNodes = JobDataNodeLine.unmarshal(tablesFirstDataNodes);
+ this.targetDataSourceConfig = targetDataSourceConfig;
+ this.sourceDataSourceMap = sourceDataSourceMap;
+ this.tablesFirstDataNodes = tablesFirstDataNodes;
this.dataSourceManager = dataSourceManager;
+ this.tableNameMap = tableNameMap;
this.tableNameSchemaNameMapping = tableNameSchemaNameMapping;
+ this.sqlParserEngine = sqlParserEngine;
}
}
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/MigrationJobPreparer.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/MigrationJobPreparer.java
index e1fbed498b5..384f9fdd1a0 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/MigrationJobPreparer.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/MigrationJobPreparer.java
@@ -21,6 +21,8 @@ import lombok.extern.slf4j.Slf4j;
import org.apache.shardingsphere.data.pipeline.api.config.TableNameSchemaNameMapping;
import org.apache.shardingsphere.data.pipeline.api.config.TaskConfiguration;
import org.apache.shardingsphere.data.pipeline.api.config.job.MigrationJobConfiguration;
+import org.apache.shardingsphere.data.pipeline.api.datanode.JobDataNodeEntry;
+import org.apache.shardingsphere.data.pipeline.api.datanode.JobDataNodeLine;
import org.apache.shardingsphere.data.pipeline.api.datasource.PipelineDataSourceManager;
import org.apache.shardingsphere.data.pipeline.api.datasource.PipelineDataSourceWrapper;
import org.apache.shardingsphere.data.pipeline.api.job.JobStatus;
@@ -40,12 +42,21 @@ import org.apache.shardingsphere.data.pipeline.core.task.IncrementalTask;
import org.apache.shardingsphere.data.pipeline.spi.ingest.channel.PipelineChannelCreator;
import org.apache.shardingsphere.infra.database.type.DatabaseType;
import org.apache.shardingsphere.infra.database.type.DatabaseTypeFactory;
+import org.apache.shardingsphere.infra.datanode.DataNode;
+import org.apache.shardingsphere.infra.datanode.DataNodes;
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.metadata.database.ShardingSphereDatabase;
+import org.apache.shardingsphere.infra.parser.ShardingSphereSQLParserEngine;
import org.apache.shardingsphere.mode.lock.ExclusiveLockDefinition;
+import org.apache.shardingsphere.parser.rule.SQLParserRule;
import java.sql.SQLException;
import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
/**
* Migration job preparer.
@@ -133,9 +144,18 @@ public final class MigrationJobPreparer {
jobItemContext.getTaskConfig().getImporterConfig().getDataSourceConfig(), jobItemContext.getDataSourceManager(), tableNameSchemaNameMapping);
PipelineJobPreparerUtils.prepareTargetSchema(targetDatabaseType, prepareTargetSchemasParameter);
}
+ ShardingSphereMetaData metaData = PipelineContext.getContextManager().getMetaDataContexts().getMetaData();
+ ShardingSphereDatabase sphereDatabase = metaData.getDatabases().get(jobConfig.getDatabaseName());
+ ShardingSphereSQLParserEngine sqlParserEngine = metaData.getGlobalRuleMetaData().getSingleRule(SQLParserRule.class).getSQLParserEngine(sphereDatabase.getProtocolType().getType());
+ JobDataNodeLine jobDataNodeLine = JobDataNodeLine.unmarshal(jobConfig.getTablesFirstDataNodes());
+ Map<String, String> tableNameMap = new HashMap<>();
+ for (JobDataNodeEntry each : jobDataNodeLine.getEntries()) {
+ String actualTableName = getActualTable(sphereDatabase, each.getLogicTableName());
+ tableNameMap.put(each.getLogicTableName(), actualTableName);
+ }
PrepareTargetTablesParameter prepareTargetTablesParameter = new PrepareTargetTablesParameter(jobConfig.getDatabaseName(),
- jobItemContext.getTaskConfig().getImporterConfig().getDataSourceConfig(),
- jobItemContext.getDataSourceManager(), jobConfig.getTablesFirstDataNodes(), tableNameSchemaNameMapping);
+ jobItemContext.getTaskConfig().getImporterConfig().getDataSourceConfig(), sphereDatabase.getResource().getDataSources(), jobItemContext.getDataSourceManager(),
+ jobDataNodeLine, tableNameMap, tableNameSchemaNameMapping, sqlParserEngine);
PipelineJobPreparerUtils.prepareTargetTables(targetDatabaseType, prepareTargetTablesParameter);
}
@@ -149,6 +169,14 @@ public final class MigrationJobPreparer {
return true;
}
+ private String getActualTable(final ShardingSphereDatabase database, final String tableName) {
+ DataNodes dataNodes = new DataNodes(database.getRuleMetaData().getRules());
+ Optional<DataNode> filteredDataNode = dataNodes.getDataNodes(tableName).stream()
+ .filter(each -> database.getResource().getDataSources().containsKey(each.getDataSourceName().contains(".") ? each.getDataSourceName().split("\\.")[0] : each.getDataSourceName()))
+ .findFirst();
+ return filteredDataNode.map(DataNode::getTableName).orElse(tableName);
+ }
+
private void initInventoryTasks(final MigrationJobItemContext jobItemContext) {
InventoryTaskSplitter inventoryTaskSplitter = new InventoryTaskSplitter(jobItemContext.getSourceMetaDataLoader(), jobItemContext.getDataSourceManager(),
jobItemContext.getJobProcessContext().getImporterExecuteEngine(), jobItemContext.getSourceDataSource(), jobItemContext.getTaskConfig(), jobItemContext.getInitProgress());
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/prepare/datasource/MySQLDataSourcePreparer.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/prepare/datasource/MySQLDataSourcePreparer.java
index 1f4c38b90a7..ceccccf1742 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/prepare/datasource/MySQLDataSourcePreparer.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/prepare/datasource/MySQLDataSourcePreparer.java
@@ -18,22 +18,13 @@
package org.apache.shardingsphere.data.pipeline.mysql.prepare.datasource;
import lombok.extern.slf4j.Slf4j;
-import org.apache.shardingsphere.data.pipeline.api.datanode.JobDataNodeEntry;
import org.apache.shardingsphere.data.pipeline.api.datasource.PipelineDataSourceManager;
-import org.apache.shardingsphere.data.pipeline.core.context.PipelineContext;
import org.apache.shardingsphere.data.pipeline.core.exception.PipelineJobPrepareFailedException;
-import org.apache.shardingsphere.data.pipeline.core.metadata.generator.PipelineDDLGenerator;
import org.apache.shardingsphere.data.pipeline.core.prepare.datasource.AbstractDataSourcePreparer;
import org.apache.shardingsphere.data.pipeline.core.prepare.datasource.PrepareTargetTablesParameter;
import java.sql.Connection;
import java.sql.SQLException;
-import java.util.LinkedList;
-import java.util.List;
-import org.apache.shardingsphere.infra.metadata.ShardingSphereMetaData;
-import org.apache.shardingsphere.infra.metadata.database.ShardingSphereDatabase;
-import org.apache.shardingsphere.infra.parser.ShardingSphereSQLParserEngine;
-import org.apache.shardingsphere.parser.rule.SQLParserRule;
/**
* Data source preparer for MySQL.
@@ -44,33 +35,15 @@ public final class MySQLDataSourcePreparer extends AbstractDataSourcePreparer {
@Override
public void prepareTargetTables(final PrepareTargetTablesParameter parameter) {
PipelineDataSourceManager dataSourceManager = parameter.getDataSourceManager();
- try (Connection targetConnection = getTargetCachedDataSource(parameter.getDataSourceConfig(), dataSourceManager).getConnection()) {
- for (String each : getCreateTableSQL(parameter)) {
+ try (Connection targetConnection = getCachedDataSource(parameter.getTargetDataSourceConfig(), dataSourceManager).getConnection()) {
+ for (String each : listCreateLogicalTableSQL(parameter)) {
executeTargetTableSQL(targetConnection, addIfNotExistsForCreateTableSQL(each));
- log.info("create target table '{}' success", each);
}
} catch (final SQLException ex) {
throw new PipelineJobPrepareFailedException("prepare target tables failed.", ex);
}
}
- private List<String> getCreateTableSQL(final PrepareTargetTablesParameter parameter) {
- PipelineDDLGenerator generator = new PipelineDDLGenerator();
- ShardingSphereMetaData metaData = PipelineContext.getContextManager().getMetaDataContexts().getMetaData();
- ShardingSphereDatabase sphereDatabase = metaData.getDatabases().get(parameter.getDatabaseName());
- ShardingSphereSQLParserEngine sqlParserEngine =
- metaData.getGlobalRuleMetaData().getSingleRule(SQLParserRule.class)
- .getSQLParserEngine(sphereDatabase.getProtocolType().getType());
- List<String> result = new LinkedList<>();
- for (JobDataNodeEntry each : parameter.getTablesFirstDataNodes().getEntries()) {
- String schemaName = parameter.getTableNameSchemaNameMapping().getSchemaName(each.getLogicTableName());
- String dataSourceName = each.getDataNodes().get(0).getDataSourceName();
- result.add(generator.generateLogicDDLSQL(sphereDatabase, dataSourceName, schemaName, each.getLogicTableName(),
- getActualTable(sphereDatabase, each.getLogicTableName()), sqlParserEngine));
- }
- return result;
- }
-
@Override
public String getType() {
return "MySQL";
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-opengauss/src/main/java/org/apache/shardingsphere/data/pipeline/opengauss/prepare/datasource/OpenGaussDataSourcePreparer.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-opengauss/src/main/java/org/apache/shardingsphere/data/pipeline/opengauss/prepare/datasource/OpenGaussDataSourcePreparer.java
index de5dcbd0a53..f42546f4471 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-opengauss/src/main/java/org/apache/shardingsphere/data/pipeline/opengauss/prepare/datasource/OpenGaussDataSourcePreparer.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-opengauss/src/main/java/org/apache/shardingsphere/data/pipeline/opengauss/prepare/datasource/OpenGaussDataSourcePreparer.java
@@ -20,22 +20,14 @@ package org.apache.shardingsphere.data.pipeline.opengauss.prepare.datasource;
import com.google.common.base.Splitter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
-import org.apache.shardingsphere.data.pipeline.api.datanode.JobDataNodeEntry;
-import org.apache.shardingsphere.data.pipeline.core.context.PipelineContext;
import org.apache.shardingsphere.data.pipeline.core.exception.PipelineJobPrepareFailedException;
-import org.apache.shardingsphere.data.pipeline.core.metadata.generator.PipelineDDLGenerator;
import org.apache.shardingsphere.data.pipeline.core.prepare.datasource.AbstractDataSourcePreparer;
import org.apache.shardingsphere.data.pipeline.core.prepare.datasource.PrepareTargetTablesParameter;
import java.sql.Connection;
import java.sql.SQLException;
-import java.util.LinkedList;
import java.util.List;
import java.util.stream.Collectors;
-import org.apache.shardingsphere.infra.metadata.ShardingSphereMetaData;
-import org.apache.shardingsphere.infra.metadata.database.ShardingSphereDatabase;
-import org.apache.shardingsphere.infra.parser.ShardingSphereSQLParserEngine;
-import org.apache.shardingsphere.parser.rule.SQLParserRule;
/**
* Data source preparer for openGauss.
@@ -46,7 +38,7 @@ public final class OpenGaussDataSourcePreparer extends AbstractDataSourcePrepare
@Override
public void prepareTargetTables(final PrepareTargetTablesParameter parameter) {
List<String> createLogicalTableSQLs = listCreateLogicalTableSQL(parameter);
- try (Connection targetConnection = getTargetCachedDataSource(parameter.getDataSourceConfig(), parameter.getDataSourceManager()).getConnection()) {
+ try (Connection targetConnection = getCachedDataSource(parameter.getTargetDataSourceConfig(), parameter.getDataSourceManager()).getConnection()) {
for (String createLogicalTableSQL : createLogicalTableSQLs) {
for (String each : Splitter.on(";").splitToList(createLogicalTableSQL).stream().filter(StringUtils::isNotBlank).collect(Collectors.toList())) {
executeTargetTableSQL(targetConnection, addIfNotExistsForCreateTableSQL(each));
@@ -57,22 +49,6 @@ public final class OpenGaussDataSourcePreparer extends AbstractDataSourcePrepare
}
}
- private List<String> listCreateLogicalTableSQL(final PrepareTargetTablesParameter parameter) {
- PipelineDDLGenerator generator = new PipelineDDLGenerator();
- ShardingSphereMetaData metaData = PipelineContext.getContextManager().getMetaDataContexts().getMetaData();
- ShardingSphereDatabase sphereDatabase = metaData.getDatabases().get(parameter.getDatabaseName());
- ShardingSphereSQLParserEngine sqlParserEngine = metaData.getGlobalRuleMetaData().getSingleRule(SQLParserRule.class)
- .getSQLParserEngine(sphereDatabase.getProtocolType().getType());
- List<String> result = new LinkedList<>();
- for (JobDataNodeEntry each : parameter.getTablesFirstDataNodes().getEntries()) {
- String schemaName = parameter.getTableNameSchemaNameMapping().getSchemaName(each.getLogicTableName());
- String dataSourceName = each.getDataNodes().get(0).getDataSourceName();
- result.add(generator.generateLogicDDLSQL(sphereDatabase, dataSourceName, schemaName, each.getLogicTableName(),
- getActualTable(sphereDatabase, each.getLogicTableName()), sqlParserEngine));
- }
- return result;
- }
-
@Override
public String getType() {
return "openGauss";
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/prepare/datasource/PostgreSQLDataSourcePreparer.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/prepare/datasource/PostgreSQLDataSource [...]
index 61287d90bbb..589b1860065 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/prepare/datasource/PostgreSQLDataSourcePreparer.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/prepare/datasource/PostgreSQLDataSourcePreparer.java
@@ -20,22 +20,14 @@ package org.apache.shardingsphere.data.pipeline.postgresql.prepare.datasource;
import com.google.common.base.Splitter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
-import org.apache.shardingsphere.data.pipeline.api.datanode.JobDataNodeEntry;
-import org.apache.shardingsphere.data.pipeline.core.context.PipelineContext;
import org.apache.shardingsphere.data.pipeline.core.exception.PipelineJobPrepareFailedException;
-import org.apache.shardingsphere.data.pipeline.core.metadata.generator.PipelineDDLGenerator;
import org.apache.shardingsphere.data.pipeline.core.prepare.datasource.AbstractDataSourcePreparer;
import org.apache.shardingsphere.data.pipeline.core.prepare.datasource.PrepareTargetTablesParameter;
import java.sql.Connection;
import java.sql.SQLException;
-import java.util.LinkedList;
import java.util.List;
import java.util.stream.Collectors;
-import org.apache.shardingsphere.infra.metadata.ShardingSphereMetaData;
-import org.apache.shardingsphere.infra.metadata.database.ShardingSphereDatabase;
-import org.apache.shardingsphere.infra.parser.ShardingSphereSQLParserEngine;
-import org.apache.shardingsphere.parser.rule.SQLParserRule;
/**
* Data source preparer for PostgresSQL.
@@ -46,7 +38,7 @@ public final class PostgreSQLDataSourcePreparer extends AbstractDataSourcePrepar
@Override
public void prepareTargetTables(final PrepareTargetTablesParameter parameter) {
List<String> createLogicalTableSQLs = listCreateLogicalTableSQL(parameter);
- try (Connection targetConnection = getTargetCachedDataSource(parameter.getDataSourceConfig(), parameter.getDataSourceManager()).getConnection()) {
+ try (Connection targetConnection = getCachedDataSource(parameter.getTargetDataSourceConfig(), parameter.getDataSourceManager()).getConnection()) {
for (String createLogicalTableSQL : createLogicalTableSQLs) {
for (String each : Splitter.on(";").splitToList(createLogicalTableSQL).stream().filter(StringUtils::isNotBlank).collect(Collectors.toList())) {
executeTargetTableSQL(targetConnection, each);
@@ -57,23 +49,6 @@ public final class PostgreSQLDataSourcePreparer extends AbstractDataSourcePrepar
}
}
- private List<String> listCreateLogicalTableSQL(final PrepareTargetTablesParameter parameter) {
- PipelineDDLGenerator generator = new PipelineDDLGenerator();
- ShardingSphereMetaData metaData = PipelineContext.getContextManager().getMetaDataContexts().getMetaData();
- ShardingSphereDatabase sphereDatabase = metaData.getDatabases().get(parameter.getDatabaseName());
- ShardingSphereSQLParserEngine sqlParserEngine =
- metaData.getGlobalRuleMetaData().getSingleRule(SQLParserRule.class)
- .getSQLParserEngine(sphereDatabase.getProtocolType().getType());
- List<String> result = new LinkedList<>();
- for (JobDataNodeEntry each : parameter.getTablesFirstDataNodes().getEntries()) {
- String schemaName = parameter.getTableNameSchemaNameMapping().getSchemaName(each.getLogicTableName());
- String dataSourceName = each.getDataNodes().get(0).getDataSourceName();
- result.add(generator.generateLogicDDLSQL(sphereDatabase, dataSourceName, schemaName, each.getLogicTableName(),
- getActualTable(sphereDatabase, each.getLogicTableName()), sqlParserEngine));
- }
- return result;
- }
-
@Override
public String getType() {
return "PostgreSQL";