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 2022/05/04 03:47:58 UTC
[shardingsphere] branch master updated: Dumper and importer SQL support multi schema (#17306)
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 7515dc4afa8 Dumper and importer SQL support multi schema (#17306)
7515dc4afa8 is described below
commit 7515dc4afa832086b8333ceac09ab3b6e3fd9f86
Author: Hongsheng Zhong <sa...@126.com>
AuthorDate: Wed May 4 11:47:48 2022 +0800
Dumper and importer SQL support multi schema (#17306)
* Refactor buildInventoryDumpSQL
* Refactor quote
* Refactor importerConfig & importer & buildInsertSQL
* Refactor buildUpdateSQL
* Refactor buildDeleteSQL
* Refactor buildTruncateSQL
* Refactor buildCountSQL
* Refactor buildCheckEmptySQL
* Refactor buildSplitByPrimaryKeyRangeSQL
* Refactor buildChunkedQuerySQL
* Refactor buildCRC32SQL
* Unit test
* Append schemaName in PostgreSQL create table SQL
* Add log
---
...hardingRuleAlteredJobConfigurationPreparer.java | 9 +--
.../DataConsistencyCalculateParameter.java | 3 +
.../ingest/InventoryDumperConfiguration.java | 1 +
.../config/rulealtered/ImporterConfiguration.java | 13 +++++
.../spi/check/datasource/DataSourceChecker.java | 6 +-
.../spi/sqlbuilder/PipelineSQLBuilder.java | 37 +++++++++---
.../check/consistency/DataConsistencyChecker.java | 22 ++++---
...RC32MatchDataConsistencyCalculateAlgorithm.java | 6 +-
...DataMatchDataConsistencyCalculateAlgorithm.java | 2 +-
.../datasource/AbstractDataSourceChecker.java | 14 +++--
.../pipeline/core/importer/AbstractImporter.java | 15 +++--
.../ingest/dumper/AbstractInventoryDumper.java | 14 ++---
.../sqlbuilder/AbstractPipelineSQLBuilder.java | 68 ++++++++++++++--------
.../rulealtered/RuleAlteredJobPreparer.java | 4 +-
.../rulealtered/prepare/InventoryTaskSplitter.java | 2 +-
.../job/environment/ScalingEnvironmentManager.java | 10 +++-
...MatchDataConsistencyCalculateAlgorithmTest.java | 5 +-
.../fixture/FixturePipelineSQLBuilder.java | 23 +++++---
.../pipeline/mysql/importer/MySQLImporter.java | 5 ++
.../mysql/sqlbuilder/MySQLPipelineSQLBuilder.java | 11 +++-
.../mysql}/fixture/FixturePipelineSQLBuilder.java | 7 ++-
.../sqlbuilder/MySQLPipelineSQLBuilderTest.java | 6 +-
...data.pipeline.spi.sqlbuilder.PipelineSQLBuilder | 18 ++++++
.../opengauss/importer/OpenGaussImporter.java | 6 ++
.../datasource/OpenGaussDataSourcePreparer.java | 1 +
.../sqlbuilder/OpenGaussPipelineSQLBuilder.java | 9 ++-
.../fixture/FixturePipelineSQLBuilder.java | 7 ++-
...data.pipeline.spi.sqlbuilder.PipelineSQLBuilder | 18 ++++++
.../postgresql/importer/PostgreSQLImporter.java | 6 ++
.../datasource/PostgreSQLDataSourcePreparer.java | 3 +-
.../sqlbuilder/PostgreSQLPipelineSQLBuilder.java | 9 ++-
.../fixture/FixturePipelineSQLBuilder.java | 7 ++-
.../PostgreSQLPipelineSQLBuilderTest.java | 4 +-
...data.pipeline.spi.sqlbuilder.PipelineSQLBuilder | 18 ++++++
.../datasource/AbstractDataSourceCheckerTest.java | 5 +-
.../core/fixture/FixtureDataSourceChecker.java | 3 +-
.../core/fixture/FixturePipelineSQLBuilder.java | 5 ++
.../core/importer/AbstractImporterTest.java | 7 ++-
.../data/pipeline/core/task/InventoryTaskTest.java | 12 +++-
.../spi/sqlbuilder/PipelineSQLBuilderTest.java | 10 ++--
40 files changed, 323 insertions(+), 108 deletions(-)
diff --git a/shardingsphere-features/shardingsphere-sharding/shardingsphere-sharding-core/src/main/java/org/apache/shardingsphere/sharding/schedule/ShardingRuleAlteredJobConfigurationPreparer.java b/shardingsphere-features/shardingsphere-sharding/shardingsphere-sharding-core/src/main/java/org/apache/shardingsphere/sharding/schedule/ShardingRuleAlteredJobConfigurationPreparer.java
index 4954a91f014..bb5123abda4 100644
--- a/shardingsphere-features/shardingsphere-sharding/shardingsphere-sharding-core/src/main/java/org/apache/shardingsphere/sharding/schedule/ShardingRuleAlteredJobConfigurationPreparer.java
+++ b/shardingsphere-features/shardingsphere-sharding/shardingsphere-sharding-core/src/main/java/org/apache/shardingsphere/sharding/schedule/ShardingRuleAlteredJobConfigurationPreparer.java
@@ -152,7 +152,7 @@ public final class ShardingRuleAlteredJobConfigurationPreparer implements RuleAl
Optional<ShardingRuleConfiguration> targetRuleConfigOptional = getTargetRuleConfiguration(jobConfig);
Set<LogicTableName> reShardNeededTables = jobConfig.splitLogicTableNames().stream().map(LogicTableName::new).collect(Collectors.toSet());
Map<LogicTableName, Set<String>> shardingColumnsMap = getShardingColumnsMap(targetRuleConfigOptional.orElse(sourceRuleConfig), reShardNeededTables);
- ImporterConfiguration importerConfig = createImporterConfiguration(jobConfig, onRuleAlteredActionConfig, shardingColumnsMap);
+ ImporterConfiguration importerConfig = createImporterConfiguration(jobConfig, onRuleAlteredActionConfig, shardingColumnsMap, metaData);
TaskConfiguration result = new TaskConfiguration(jobConfig, dumperConfig, importerConfig);
log.info("createTaskConfiguration, dataSourceName={}, result={}", dataSourceName, result);
return result;
@@ -218,12 +218,13 @@ public final class ShardingRuleAlteredJobConfigurationPreparer implements RuleAl
return result;
}
- private static ImporterConfiguration createImporterConfiguration(final RuleAlteredJobConfiguration jobConfig,
- final OnRuleAlteredActionConfiguration onRuleAlteredActionConfig, final Map<LogicTableName, Set<String>> shardingColumnsMap) {
+ private static ImporterConfiguration createImporterConfiguration(final RuleAlteredJobConfiguration jobConfig, final OnRuleAlteredActionConfiguration onRuleAlteredActionConfig,
+ final Map<LogicTableName, Set<String>> shardingColumnsMap, final ShardingSphereMetaData metaData) {
PipelineDataSourceConfiguration dataSourceConfig = PipelineDataSourceConfigurationFactory.newInstance(jobConfig.getTarget().getType(), jobConfig.getTarget().getParameter());
+ TableNameSchemaNameMapping tableNameSchemaNameMapping = new TableNameSchemaNameMapping(TableNameSchemaNameMapping.convert(metaData.getSchemas()));
int batchSize = onRuleAlteredActionConfig.getOutput().getBatchSize();
int retryTimes = jobConfig.getRetryTimes();
- return new ImporterConfiguration(dataSourceConfig, unmodifiable(shardingColumnsMap), batchSize, retryTimes);
+ return new ImporterConfiguration(dataSourceConfig, unmodifiable(shardingColumnsMap), tableNameSchemaNameMapping, batchSize, retryTimes);
}
private static Map<LogicTableName, Set<String>> unmodifiable(final Map<LogicTableName, Set<String>> shardingColumnsMap) {
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/check/consistency/DataConsistencyCalculateParameter.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/check/consistency/DataConsistencyCalculateParameter.java
index 4ba13f73551..59e7b948ef7 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/check/consistency/DataConsistencyCalculateParameter.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/check/consistency/DataConsistencyCalculateParameter.java
@@ -22,6 +22,7 @@ import lombok.Getter;
import lombok.RequiredArgsConstructor;
import lombok.Setter;
import lombok.ToString;
+import org.apache.shardingsphere.data.pipeline.api.config.TableNameSchemaNameMapping;
import org.apache.shardingsphere.data.pipeline.api.datasource.PipelineDataSourceWrapper;
import java.util.Collection;
@@ -41,6 +42,8 @@ public final class DataConsistencyCalculateParameter {
*/
private final PipelineDataSourceWrapper dataSource;
+ private final TableNameSchemaNameMapping tableNameSchemaNameMapping;
+
private final String logicTableName;
private final Collection<String> columnNames;
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/config/ingest/InventoryDumperConfiguration.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/config/ingest/InventoryDumperConfiguration.java
index a09a9aa24e2..9b4ba152b57 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/config/ingest/InventoryDumperConfiguration.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/config/ingest/InventoryDumperConfiguration.java
@@ -35,6 +35,7 @@ public final class InventoryDumperConfiguration extends DumperConfiguration {
private String logicTableName;
+ // TODO rename to uniqueKey
private String primaryKey;
private Integer shardingItem;
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/config/rulealtered/ImporterConfiguration.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/config/rulealtered/ImporterConfiguration.java
index d6a5e45e7c1..af26e11e7cd 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/config/rulealtered/ImporterConfiguration.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/config/rulealtered/ImporterConfiguration.java
@@ -21,6 +21,7 @@ import lombok.Getter;
import lombok.RequiredArgsConstructor;
import lombok.Setter;
import lombok.ToString;
+import org.apache.shardingsphere.data.pipeline.api.config.TableNameSchemaNameMapping;
import org.apache.shardingsphere.data.pipeline.api.datasource.config.PipelineDataSourceConfiguration;
import org.apache.shardingsphere.data.pipeline.api.metadata.LogicTableName;
@@ -45,6 +46,8 @@ public final class ImporterConfiguration {
// TODO columnName case-insensitive?
private final Map<LogicTableName, Set<String>> shardingColumnsMap;
+ private final TableNameSchemaNameMapping tableNameSchemaNameMapping;
+
private final int batchSize;
private final int retryTimes;
@@ -68,4 +71,14 @@ public final class ImporterConfiguration {
public Set<String> getShardingColumns(final String logicTableName) {
return shardingColumnsMap.get(new LogicTableName(logicTableName));
}
+
+ /**
+ * Get schema name.
+ *
+ * @param logicTableName logic table name
+ * @return schema name. nullable
+ */
+ public String getSchemaName(final LogicTableName logicTableName) {
+ return tableNameSchemaNameMapping.getSchemaName(logicTableName);
+ }
}
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/spi/check/datasource/DataSourceChecker.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/spi/check/datasource/DataSourceChecker.java
index 5f77de29d31..337b4c4ca8a 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/spi/check/datasource/DataSourceChecker.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/spi/check/datasource/DataSourceChecker.java
@@ -17,6 +17,7 @@
package org.apache.shardingsphere.data.pipeline.spi.check.datasource;
+import org.apache.shardingsphere.data.pipeline.api.config.TableNameSchemaNameMapping;
import org.apache.shardingsphere.spi.type.typed.StatelessTypedSPI;
import javax.sql.DataSource;
@@ -52,8 +53,9 @@ public interface DataSourceChecker extends StatelessTypedSPI {
* Check table is empty.
*
* @param dataSources data sources
- * @param tableNames table names
+ * @param tableNameSchemaNameMapping mapping
+ * @param logicTableNames logic table names
*/
// TODO rename to common usage name
- void checkTargetTable(Collection<? extends DataSource> dataSources, Collection<String> tableNames);
+ void checkTargetTable(Collection<? extends DataSource> dataSources, TableNameSchemaNameMapping tableNameSchemaNameMapping, Collection<String> logicTableNames);
}
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/spi/sqlbuilder/PipelineSQLBuilder.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/spi/sqlbuilder/PipelineSQLBuilder.java
index 7faaf969e47..d5ba20f357d 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/spi/sqlbuilder/PipelineSQLBuilder.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/spi/sqlbuilder/PipelineSQLBuilder.java
@@ -33,24 +33,36 @@ import java.util.Set;
*/
public interface PipelineSQLBuilder extends StatefulTypedSPI {
+ /**
+ * Build inventory dump SQL.
+ *
+ * @param schemaName schema name
+ * @param tableName table name
+ * @param uniqueKey unique key
+ * @return inventory dump SQL
+ */
+ String buildInventoryDumpSQL(String schemaName, String tableName, String uniqueKey);
+
/**
* Build insert SQL.
*
+ * @param schemaName schema name
* @param dataRecord data record
* @param shardingColumnsMap sharding columns map
* @return insert SQL
*/
- String buildInsertSQL(DataRecord dataRecord, Map<LogicTableName, Set<String>> shardingColumnsMap);
+ String buildInsertSQL(String schemaName, DataRecord dataRecord, Map<LogicTableName, Set<String>> shardingColumnsMap);
/**
* Build update SQL.
*
+ * @param schemaName schema name
* @param dataRecord data record
* @param conditionColumns condition columns
* @param shardingColumnsMap sharding columns map
* @return update SQL
*/
- String buildUpdateSQL(DataRecord dataRecord, Collection<Column> conditionColumns, Map<LogicTableName, Set<String>> shardingColumnsMap);
+ String buildUpdateSQL(String schemaName, DataRecord dataRecord, Collection<Column> conditionColumns, Map<LogicTableName, Set<String>> shardingColumnsMap);
/**
* Extract updated columns.
@@ -64,63 +76,70 @@ public interface PipelineSQLBuilder extends StatefulTypedSPI {
/**
* Build delete SQL.
*
+ * @param schemaName schema name
* @param dataRecord data record
* @param conditionColumns condition columns
* @return delete SQL
*/
- String buildDeleteSQL(DataRecord dataRecord, Collection<Column> conditionColumns);
+ String buildDeleteSQL(String schemaName, DataRecord dataRecord, Collection<Column> conditionColumns);
/**
* Build truncate SQL.
*
+ * @param schemaName schema name
* @param tableName table name
* @return truncate SQL
*/
- String buildTruncateSQL(String tableName);
+ String buildTruncateSQL(String schemaName, String tableName);
/**
* Build count SQL.
*
+ * @param schemaName schema name
* @param tableName table name
* @return count SQL
*/
- String buildCountSQL(String tableName);
+ String buildCountSQL(String schemaName, String tableName);
/**
* Build query SQL.
*
+ * @param schemaName schema name
* @param tableName table name
* @param uniqueKey unique key, it may be primary key, not null
* @param startUniqueValue start unique value, not null
* @return query SQL
*/
- String buildChunkedQuerySQL(String tableName, String uniqueKey, Number startUniqueValue);
+ String buildChunkedQuerySQL(String schemaName, String tableName, String uniqueKey, Number startUniqueValue);
/**
* Build check empty SQL.
*
+ * @param schemaName schema name
* @param tableName table name
* @return check SQL
*/
- String buildCheckEmptySQL(String tableName);
+ String buildCheckEmptySQL(String schemaName, String tableName);
/**
* Build split by primary key range SQL.
*
+ * @param schemaName schema name
* @param tableName table name
* @param primaryKey primary key
* @return split SQL
*/
- String buildSplitByPrimaryKeyRangeSQL(String tableName, String primaryKey);
+ String buildSplitByPrimaryKeyRangeSQL(String schemaName, String tableName, String primaryKey);
/**
* Build CRC32 SQL.
*
+ * @param schemaName schema name
* @param tableName table Name
* @param column column
* @return CRC32 SQL
*/
- default Optional<String> buildCRC32SQL(final String tableName, final String column) {
+ default Optional<String> buildCRC32SQL(final String schemaName, final String tableName, final String column) {
return Optional.empty();
}
}
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/check/consistency/DataConsistencyChecker.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/check/consistency/DataConsistencyChecker.java
index bc34be7fb6f..2aa36f9936f 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/check/consistency/DataConsistencyChecker.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/check/consistency/DataConsistencyChecker.java
@@ -22,6 +22,7 @@ import org.apache.shardingsphere.data.pipeline.api.check.consistency.DataConsist
import org.apache.shardingsphere.data.pipeline.api.check.consistency.DataConsistencyCheckResult;
import org.apache.shardingsphere.data.pipeline.api.check.consistency.DataConsistencyContentCheckResult;
import org.apache.shardingsphere.data.pipeline.api.check.consistency.DataConsistencyCountCheckResult;
+import org.apache.shardingsphere.data.pipeline.api.config.TableNameSchemaNameMapping;
import org.apache.shardingsphere.data.pipeline.api.config.rulealtered.RuleAlteredJobConfiguration;
import org.apache.shardingsphere.data.pipeline.api.datasource.PipelineDataSourceWrapper;
import org.apache.shardingsphere.data.pipeline.api.datasource.config.PipelineDataSourceConfiguration;
@@ -73,9 +74,13 @@ public final class DataConsistencyChecker {
private final Collection<String> logicTableNames;
+ private final TableNameSchemaNameMapping tableNameSchemaNameMapping;
+
public DataConsistencyChecker(final RuleAlteredJobConfiguration jobConfig) {
this.jobConfig = jobConfig;
logicTableNames = jobConfig.splitLogicTableNames();
+ ShardingSphereMetaData metaData = PipelineContext.getContextManager().getMetaDataContexts().getMetaData(jobConfig.getDatabaseName());
+ tableNameSchemaNameMapping = new TableNameSchemaNameMapping(TableNameSchemaNameMapping.convert(metaData.getSchemas()));
}
/**
@@ -135,15 +140,16 @@ public final class DataConsistencyChecker {
return jobId.length() <= 6 ? jobId : jobId.substring(0, 6);
}
- private long count(final DataSource dataSource, final String table, final DatabaseType databaseType) {
+ private long count(final DataSource dataSource, final String tableName, final DatabaseType databaseType) {
+ String sql = PipelineSQLBuilderFactory.newInstance(databaseType.getName()).buildCountSQL(tableNameSchemaNameMapping.getSchemaName(tableName), tableName);
try (
Connection connection = dataSource.getConnection();
- PreparedStatement preparedStatement = connection.prepareStatement(PipelineSQLBuilderFactory.newInstance(databaseType.getName()).buildCountSQL(table));
+ PreparedStatement preparedStatement = connection.prepareStatement(sql);
ResultSet resultSet = preparedStatement.executeQuery()) {
resultSet.next();
return resultSet.getLong(1);
} catch (final SQLException ex) {
- throw new PipelineDataConsistencyCheckFailedException(String.format("Count for table '%s' failed", table), ex);
+ throw new PipelineDataConsistencyCheckFailedException(String.format("Count for table '%s' failed", tableName), ex);
}
}
@@ -170,8 +176,8 @@ public final class DataConsistencyChecker {
TableMetaData tableMetaData = tableMetaDataMap.get(each);
Collection<String> columnNames = tableMetaData.getColumns().keySet();
String uniqueKey = tableMetaData.getPrimaryKeyColumns().get(0);
- DataConsistencyCalculateParameter sourceParameter = buildParameter(sourceDataSource, each, columnNames, sourceDatabaseType, targetDatabaseType, uniqueKey);
- DataConsistencyCalculateParameter targetParameter = buildParameter(targetDataSource, each, columnNames, targetDatabaseType, sourceDatabaseType, uniqueKey);
+ DataConsistencyCalculateParameter sourceParameter = buildParameter(sourceDataSource, tableNameSchemaNameMapping, each, columnNames, sourceDatabaseType, targetDatabaseType, uniqueKey);
+ DataConsistencyCalculateParameter targetParameter = buildParameter(targetDataSource, tableNameSchemaNameMapping, each, columnNames, targetDatabaseType, sourceDatabaseType, uniqueKey);
Iterator<Object> sourceCalculatedResults = calculator.calculate(sourceParameter).iterator();
Iterator<Object> targetCalculatedResults = calculator.calculate(targetParameter).iterator();
boolean contentMatched = true;
@@ -231,8 +237,8 @@ public final class DataConsistencyChecker {
return metaData.getSchemaByName(schema).getTables();
}
- private DataConsistencyCalculateParameter buildParameter(final PipelineDataSourceWrapper sourceDataSource, final String tableName, final Collection<String> columnNames,
- final String sourceDatabaseType, final String targetDatabaseType, final String uniqueKey) {
- return new DataConsistencyCalculateParameter(sourceDataSource, tableName, columnNames, sourceDatabaseType, targetDatabaseType, uniqueKey);
+ private DataConsistencyCalculateParameter buildParameter(final PipelineDataSourceWrapper sourceDataSource, final TableNameSchemaNameMapping tableNameSchemaNameMapping, final String tableName,
+ final Collection<String> columnNames, final String sourceDatabaseType, final String targetDatabaseType, final String uniqueKey) {
+ return new DataConsistencyCalculateParameter(sourceDataSource, tableNameSchemaNameMapping, tableName, columnNames, sourceDatabaseType, targetDatabaseType, uniqueKey);
}
}
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/check/consistency/algorithm/CRC32MatchDataConsistencyCalculateAlgorithm.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/check/consistency/algorithm/CRC32MatchDataConsistencyCalculateAlgorithm.java
index 2bab64f3b3f..f972d01cb4e 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/check/consistency/algorithm/CRC32MatchDataConsistencyCalculateAlgorithm.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/check/consistency/algorithm/CRC32MatchDataConsistencyCalculateAlgorithm.java
@@ -48,7 +48,9 @@ public final class CRC32MatchDataConsistencyCalculateAlgorithm implements DataCo
}
private long calculateCRC32(final PipelineSQLBuilder sqlBuilder, final DataConsistencyCalculateParameter parameter, final String columnName) {
- Optional<String> sql = sqlBuilder.buildCRC32SQL(parameter.getLogicTableName(), columnName);
+ String logicTableName = parameter.getLogicTableName();
+ String schemaName = parameter.getTableNameSchemaNameMapping().getSchemaName(logicTableName);
+ Optional<String> sql = sqlBuilder.buildCRC32SQL(schemaName, logicTableName, columnName);
if (!sql.isPresent()) {
throw new PipelineDataConsistencyCheckFailedException(
String.format("Unsupported CRC32 data consistency calculate algorithm with database type `%s`", parameter.getDatabaseType()));
@@ -56,7 +58,7 @@ public final class CRC32MatchDataConsistencyCalculateAlgorithm implements DataCo
try {
return calculateCRC32(parameter.getDataSource(), sql.get());
} catch (final SQLException ex) {
- throw new PipelineDataConsistencyCheckFailedException(String.format("Table `%s` data check failed.", parameter.getLogicTableName()), ex);
+ throw new PipelineDataConsistencyCheckFailedException(String.format("Table `%s` data check failed.", logicTableName), ex);
}
}
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/check/consistency/algorithm/DataMatchDataConsistencyCalculateAlgorithm.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/check/consistency/algorithm/DataMatchDataConsistencyCalculateAlgorithm.java
index 8d6a4c803aa..f71b177e599 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/check/consistency/algorithm/DataMatchDataConsistencyCalculateAlgorithm.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/check/consistency/algorithm/DataMatchDataConsistencyCalculateAlgorithm.java
@@ -78,7 +78,7 @@ public final class DataMatchDataConsistencyCalculateAlgorithm extends AbstractSt
String uniqueKey = parameter.getUniqueKey();
CalculatedResult previousCalculatedResult = (CalculatedResult) parameter.getPreviousCalculatedResult();
Number startUniqueKeyValue = null != previousCalculatedResult ? previousCalculatedResult.getMaxUniqueKeyValue() : -1;
- String sql = sqlBuilder.buildChunkedQuerySQL(logicTableName, uniqueKey, startUniqueKeyValue);
+ String sql = sqlBuilder.buildChunkedQuerySQL(parameter.getTableNameSchemaNameMapping().getSchemaName(logicTableName), logicTableName, uniqueKey, startUniqueKeyValue);
try {
return query(parameter.getDataSource(), sql, uniqueKey, startUniqueKeyValue, chunkSize);
} catch (final SQLException ex) {
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/check/datasource/AbstractDataSourceChecker.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/check/datasource/AbstractDataSourceChecker.java
index bdb2f718990..678f11f47a9 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/check/datasource/AbstractDataSourceChecker.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/check/datasource/AbstractDataSourceChecker.java
@@ -17,6 +17,8 @@
package org.apache.shardingsphere.data.pipeline.core.check.datasource;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.shardingsphere.data.pipeline.api.config.TableNameSchemaNameMapping;
import org.apache.shardingsphere.data.pipeline.core.exception.PipelineJobPrepareFailedException;
import org.apache.shardingsphere.data.pipeline.core.sqlbuilder.PipelineSQLBuilderFactory;
import org.apache.shardingsphere.data.pipeline.spi.check.datasource.DataSourceChecker;
@@ -32,6 +34,7 @@ import java.util.Collection;
/**
* Abstract data source checker.
*/
+@Slf4j
public abstract class AbstractDataSourceChecker implements DataSourceChecker {
@Override
@@ -46,19 +49,20 @@ public abstract class AbstractDataSourceChecker implements DataSourceChecker {
}
@Override
- public final void checkTargetTable(final Collection<? extends DataSource> dataSources, final Collection<String> tableNames) {
+ public final void checkTargetTable(final Collection<? extends DataSource> dataSources, final TableNameSchemaNameMapping tableNameSchemaNameMapping, final Collection<String> logicTableNames) {
try {
for (DataSource each : dataSources) {
- checkEmpty(each, tableNames);
+ checkEmpty(each, tableNameSchemaNameMapping, logicTableNames);
}
} catch (final SQLException ex) {
throw new PipelineJobPrepareFailedException("Check target table failed.", ex);
}
}
- private void checkEmpty(final DataSource dataSource, final Collection<String> tableNames) throws SQLException {
- for (String each : tableNames) {
- String sql = getSQLBuilder().buildCheckEmptySQL(each);
+ private void checkEmpty(final DataSource dataSource, final TableNameSchemaNameMapping tableNameSchemaNameMapping, final Collection<String> logicTableNames) throws SQLException {
+ for (String each : logicTableNames) {
+ String sql = getSQLBuilder().buildCheckEmptySQL(tableNameSchemaNameMapping.getSchemaName(each), each);
+ log.info("checkEmpty, sql={}", sql);
try (
Connection connection = dataSource.getConnection();
PreparedStatement preparedStatement = connection.prepareStatement(sql);
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/importer/AbstractImporter.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/importer/AbstractImporter.java
index 6f61ea7b6c2..9e157ad1329 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/importer/AbstractImporter.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/importer/AbstractImporter.java
@@ -17,6 +17,8 @@
package org.apache.shardingsphere.data.pipeline.core.importer;
+import lombok.AccessLevel;
+import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.apache.shardingsphere.data.pipeline.api.config.rulealtered.ImporterConfiguration;
import org.apache.shardingsphere.data.pipeline.api.executor.AbstractLifecycleExecutor;
@@ -51,6 +53,7 @@ public abstract class AbstractImporter extends AbstractLifecycleExecutor impleme
private static final DataRecordMerger MERGER = new DataRecordMerger();
+ @Getter(AccessLevel.PROTECTED)
private final ImporterConfiguration importerConfig;
private final PipelineDataSourceManager dataSourceManager;
@@ -150,7 +153,8 @@ public abstract class AbstractImporter extends AbstractLifecycleExecutor impleme
}
private void executeBatchInsert(final Connection connection, final List<DataRecord> dataRecords) throws SQLException {
- String insertSql = pipelineSqlBuilder.buildInsertSQL(dataRecords.get(0), importerConfig.getShardingColumnsMap());
+ DataRecord dataRecord = dataRecords.get(0);
+ String insertSql = pipelineSqlBuilder.buildInsertSQL(getSchemaName(dataRecord.getTableName()), dataRecord, importerConfig.getShardingColumnsMap());
try (PreparedStatement ps = connection.prepareStatement(insertSql)) {
ps.setQueryTimeout(30);
for (DataRecord each : dataRecords) {
@@ -163,6 +167,8 @@ public abstract class AbstractImporter extends AbstractLifecycleExecutor impleme
}
}
+ protected abstract String getSchemaName(String logicTableName);
+
private void executeUpdate(final Connection connection, final List<DataRecord> dataRecords) throws SQLException {
for (DataRecord each : dataRecords) {
executeUpdate(connection, each);
@@ -176,7 +182,7 @@ public abstract class AbstractImporter extends AbstractLifecycleExecutor impleme
}
List<Column> conditionColumns = RecordUtil.extractConditionColumns(record, shardingColumns);
List<Column> updatedColumns = pipelineSqlBuilder.extractUpdatedColumns(record, importerConfig.getShardingColumnsMap());
- String updateSql = pipelineSqlBuilder.buildUpdateSQL(record, conditionColumns, importerConfig.getShardingColumnsMap());
+ String updateSql = pipelineSqlBuilder.buildUpdateSQL(getSchemaName(record.getTableName()), record, conditionColumns, importerConfig.getShardingColumnsMap());
try (PreparedStatement ps = connection.prepareStatement(updateSql)) {
for (int i = 0; i < updatedColumns.size(); i++) {
ps.setObject(i + 1, updatedColumns.get(i).getValue());
@@ -190,8 +196,9 @@ public abstract class AbstractImporter extends AbstractLifecycleExecutor impleme
}
private void executeBatchDelete(final Connection connection, final List<DataRecord> dataRecords) throws SQLException {
- List<Column> conditionColumns = RecordUtil.extractConditionColumns(dataRecords.get(0), importerConfig.getShardingColumns(dataRecords.get(0).getTableName()));
- String deleteSQL = pipelineSqlBuilder.buildDeleteSQL(dataRecords.get(0), conditionColumns);
+ DataRecord dataRecord = dataRecords.get(0);
+ List<Column> conditionColumns = RecordUtil.extractConditionColumns(dataRecord, importerConfig.getShardingColumns(dataRecord.getTableName()));
+ String deleteSQL = pipelineSqlBuilder.buildDeleteSQL(getSchemaName(dataRecord.getTableName()), dataRecord, conditionColumns);
try (PreparedStatement ps = connection.prepareStatement(deleteSQL)) {
ps.setQueryTimeout(30);
for (DataRecord each : dataRecords) {
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/dumper/AbstractInventoryDumper.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/dumper/AbstractInventoryDumper.java
index 61142ae0c5d..c2485fc9779 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/dumper/AbstractInventoryDumper.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/dumper/AbstractInventoryDumper.java
@@ -41,8 +41,10 @@ import org.apache.shardingsphere.data.pipeline.core.ingest.IngestDataChangeType;
import org.apache.shardingsphere.data.pipeline.core.ingest.exception.IngestException;
import org.apache.shardingsphere.data.pipeline.core.metadata.loader.PipelineTableMetaDataLoader;
import org.apache.shardingsphere.data.pipeline.core.metadata.model.PipelineTableMetaData;
+import org.apache.shardingsphere.data.pipeline.core.sqlbuilder.PipelineSQLBuilderFactory;
import org.apache.shardingsphere.data.pipeline.spi.ingest.dumper.InventoryDumper;
import org.apache.shardingsphere.data.pipeline.spi.ratelimit.JobRateLimitAlgorithm;
+import org.apache.shardingsphere.data.pipeline.spi.sqlbuilder.PipelineSQLBuilder;
import javax.sql.DataSource;
import java.sql.Connection;
@@ -63,6 +65,8 @@ public abstract class AbstractInventoryDumper extends AbstractLifecycleExecutor
private final PipelineChannel channel;
+ private final PipelineSQLBuilder pipelineSQLBuilder;
+
private final DataSource dataSource;
private final int batchSize;
@@ -78,6 +82,7 @@ public abstract class AbstractInventoryDumper extends AbstractLifecycleExecutor
}
this.inventoryDumperConfig = inventoryDumperConfig;
this.channel = channel;
+ pipelineSQLBuilder = PipelineSQLBuilderFactory.newInstance(inventoryDumperConfig.getDataSourceConfig().getDatabaseType().getName());
this.dataSource = dataSource;
batchSize = inventoryDumperConfig.getBatchSize();
rateLimitAlgorithm = inventoryDumperConfig.getRateLimitAlgorithm();
@@ -97,7 +102,8 @@ public abstract class AbstractInventoryDumper extends AbstractLifecycleExecutor
}
private void dump() {
- String sql = getDumpSQL();
+ String schemaName = inventoryDumperConfig.getSchemaName(new LogicTableName(inventoryDumperConfig.getLogicTableName()));
+ String sql = pipelineSQLBuilder.buildInventoryDumpSQL(schemaName, inventoryDumperConfig.getActualTableName(), inventoryDumperConfig.getPrimaryKey());
IngestPosition<?> position = inventoryDumperConfig.getPosition();
log.info("inventory dump, sql={}, position={}", sql, position);
try (Connection conn = dataSource.getConnection()) {
@@ -121,12 +127,6 @@ public abstract class AbstractInventoryDumper extends AbstractLifecycleExecutor
}
}
- private String getDumpSQL() {
- String tableName = inventoryDumperConfig.getActualTableName();
- String primaryKey = inventoryDumperConfig.getPrimaryKey();
- return "SELECT * FROM " + tableName + " WHERE " + primaryKey + " > ? AND " + primaryKey + " <= ? ORDER BY " + primaryKey + " ASC LIMIT ?";
- }
-
@SneakyThrows(ConcurrentException.class)
private PipelineTableMetaData getTableMetaData() {
return tableMetaDataLazyInitializer.get();
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/AbstractPipelineSQLBuilder.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/AbstractPipelineSQLBuilder.java
index 411776cf394..ea782aabbdb 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/AbstractPipelineSQLBuilder.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/AbstractPipelineSQLBuilder.java
@@ -18,6 +18,7 @@
package org.apache.shardingsphere.data.pipeline.core.sqlbuilder;
import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
import org.apache.shardingsphere.data.pipeline.api.ingest.record.Column;
import org.apache.shardingsphere.data.pipeline.api.ingest.record.DataRecord;
import org.apache.shardingsphere.data.pipeline.api.metadata.LogicTableName;
@@ -65,20 +66,37 @@ public abstract class AbstractPipelineSQLBuilder implements PipelineSQLBuilder {
* @param item to add quote item
* @return add quote string
*/
- public StringBuilder quote(final String item) {
- return new StringBuilder().append(getLeftIdentifierQuoteString()).append(item).append(getRightIdentifierQuoteString());
+ public String quote(final String item) {
+ return getLeftIdentifierQuoteString() + item + getRightIdentifierQuoteString();
}
@Override
- public String buildInsertSQL(final DataRecord dataRecord, final Map<LogicTableName, Set<String>> shardingColumnsMap) {
+ public String buildInventoryDumpSQL(final String schemaName, final String tableName, final String uniqueKey) {
+ String quotedUniqueKey = quote(uniqueKey);
+ return "SELECT * FROM " + decorate(schemaName, tableName) + " WHERE " + quotedUniqueKey + " > ? AND " + quotedUniqueKey + " <= ? ORDER BY " + quotedUniqueKey + " ASC LIMIT ?";
+ }
+
+ protected String decorate(final String schemaName, final String tableName) {
+ StringBuilder result = new StringBuilder();
+ if (isSchemaEnabled() && !Strings.isNullOrEmpty(schemaName)) {
+ result.append(quote(schemaName)).append(".");
+ }
+ result.append(quote(tableName));
+ return result.toString();
+ }
+
+ protected abstract boolean isSchemaEnabled();
+
+ @Override
+ public String buildInsertSQL(final String schemaName, final DataRecord dataRecord, final Map<LogicTableName, Set<String>> shardingColumnsMap) {
String sqlCacheKey = INSERT_SQL_CACHE_KEY_PREFIX + dataRecord.getTableName();
if (!sqlCacheMap.containsKey(sqlCacheKey)) {
- sqlCacheMap.put(sqlCacheKey, buildInsertSQLInternal(dataRecord.getTableName(), dataRecord.getColumns()));
+ sqlCacheMap.put(sqlCacheKey, buildInsertSQLInternal(schemaName, dataRecord.getTableName(), dataRecord.getColumns()));
}
return sqlCacheMap.get(sqlCacheKey);
}
- private String buildInsertSQLInternal(final String tableName, final List<Column> columns) {
+ private String buildInsertSQLInternal(final String schemaName, final String tableName, final List<Column> columns) {
StringBuilder columnsLiteral = new StringBuilder();
StringBuilder holder = new StringBuilder();
for (Column each : columns) {
@@ -87,7 +105,7 @@ public abstract class AbstractPipelineSQLBuilder implements PipelineSQLBuilder {
}
columnsLiteral.setLength(columnsLiteral.length() - 1);
holder.setLength(holder.length() - 1);
- return String.format("INSERT INTO %s(%s) VALUES(%s)", quote(tableName), columnsLiteral, holder);
+ return String.format("INSERT INTO %s(%s) VALUES(%s)", decorate(schemaName, tableName), columnsLiteral, holder);
}
// TODO seems sharding column could be updated for insert statement on conflict by kernel now
@@ -97,10 +115,10 @@ public abstract class AbstractPipelineSQLBuilder implements PipelineSQLBuilder {
}
@Override
- public String buildUpdateSQL(final DataRecord dataRecord, final Collection<Column> conditionColumns, final Map<LogicTableName, Set<String>> shardingColumnsMap) {
+ public String buildUpdateSQL(final String schemaName, final DataRecord dataRecord, final Collection<Column> conditionColumns, final Map<LogicTableName, Set<String>> shardingColumnsMap) {
String sqlCacheKey = UPDATE_SQL_CACHE_KEY_PREFIX + dataRecord.getTableName();
if (!sqlCacheMap.containsKey(sqlCacheKey)) {
- sqlCacheMap.put(sqlCacheKey, buildUpdateSQLInternal(dataRecord.getTableName(), conditionColumns));
+ sqlCacheMap.put(sqlCacheKey, buildUpdateSQLInternal(schemaName, dataRecord.getTableName(), conditionColumns));
}
StringBuilder updatedColumnString = new StringBuilder();
for (Column each : extractUpdatedColumns(dataRecord, shardingColumnsMap)) {
@@ -110,8 +128,8 @@ public abstract class AbstractPipelineSQLBuilder implements PipelineSQLBuilder {
return String.format(sqlCacheMap.get(sqlCacheKey), updatedColumnString);
}
- private String buildUpdateSQLInternal(final String tableName, final Collection<Column> conditionColumns) {
- return String.format("UPDATE %s SET %%s WHERE %s", quote(tableName), buildWhereSQL(conditionColumns));
+ private String buildUpdateSQLInternal(final String schemaName, final String tableName, final Collection<Column> conditionColumns) {
+ return String.format("UPDATE %s SET %%s WHERE %s", decorate(schemaName, tableName), buildWhereSQL(conditionColumns));
}
@Override
@@ -120,21 +138,21 @@ public abstract class AbstractPipelineSQLBuilder implements PipelineSQLBuilder {
}
@Override
- public String buildDeleteSQL(final DataRecord dataRecord, final Collection<Column> conditionColumns) {
+ public String buildDeleteSQL(final String schemaName, final DataRecord dataRecord, final Collection<Column> conditionColumns) {
String sqlCacheKey = DELETE_SQL_CACHE_KEY_PREFIX + dataRecord.getTableName();
if (!sqlCacheMap.containsKey(sqlCacheKey)) {
- sqlCacheMap.put(sqlCacheKey, buildDeleteSQLInternal(dataRecord.getTableName(), conditionColumns));
+ sqlCacheMap.put(sqlCacheKey, buildDeleteSQLInternal(schemaName, dataRecord.getTableName(), conditionColumns));
}
return sqlCacheMap.get(sqlCacheKey);
}
@Override
- public String buildTruncateSQL(final String tableName) {
- return String.format("TRUNCATE TABLE %s", quote(tableName));
+ public String buildTruncateSQL(final String schemaName, final String tableName) {
+ return String.format("TRUNCATE TABLE %s", decorate(schemaName, tableName));
}
- private String buildDeleteSQLInternal(final String tableName, final Collection<Column> conditionColumns) {
- return String.format("DELETE FROM %s WHERE %s", quote(tableName), buildWhereSQL(conditionColumns));
+ private String buildDeleteSQLInternal(final String schemaName, final String tableName, final Collection<Column> conditionColumns) {
+ return String.format("DELETE FROM %s WHERE %s", decorate(schemaName, tableName), buildWhereSQL(conditionColumns));
}
private String buildWhereSQL(final Collection<Column> conditionColumns) {
@@ -147,25 +165,25 @@ public abstract class AbstractPipelineSQLBuilder implements PipelineSQLBuilder {
}
@Override
- public String buildCountSQL(final String tableName) {
- return String.format("SELECT COUNT(*) FROM %s", quote(tableName));
+ public String buildCountSQL(final String schemaName, final String tableName) {
+ return String.format("SELECT COUNT(*) FROM %s", decorate(schemaName, tableName));
}
@Override
- public String buildChunkedQuerySQL(final String tableName, final String uniqueKey, final Number startUniqueValue) {
+ public String buildChunkedQuerySQL(final String schemaName, final String tableName, final String uniqueKey, final Number startUniqueValue) {
Preconditions.checkNotNull(uniqueKey, "uniqueKey is null");
Preconditions.checkNotNull(startUniqueValue, "startUniqueValue is null");
- return "SELECT * FROM " + quote(tableName) + " WHERE " + quote(uniqueKey) + " > ? ORDER BY " + quote(uniqueKey) + " ASC LIMIT ?";
+ return "SELECT * FROM " + decorate(schemaName, tableName) + " WHERE " + quote(uniqueKey) + " > ? ORDER BY " + quote(uniqueKey) + " ASC LIMIT ?";
}
@Override
- public String buildCheckEmptySQL(final String tableName) {
- return String.format("SELECT * FROM %s LIMIT 1", quote(tableName));
+ public String buildCheckEmptySQL(final String schemaName, final String tableName) {
+ return String.format("SELECT * FROM %s LIMIT 1", decorate(schemaName, tableName));
}
@Override
- public String buildSplitByPrimaryKeyRangeSQL(final String tableName, final String primaryKey) {
- String quotedKey = quote(primaryKey).toString();
- return String.format("SELECT MAX(%s) FROM (SELECT %s FROM %s WHERE %s>=? ORDER BY %s LIMIT ?) t", quotedKey, quotedKey, quote(tableName), quotedKey, quotedKey);
+ public String buildSplitByPrimaryKeyRangeSQL(final String schemaName, final String tableName, final String primaryKey) {
+ String quotedKey = quote(primaryKey);
+ return String.format("SELECT MAX(%s) FROM (SELECT %s FROM %s WHERE %s>=? ORDER BY %s LIMIT ?) t", quotedKey, quotedKey, decorate(schemaName, tableName), quotedKey, quotedKey);
}
}
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/RuleAlteredJobPreparer.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/RuleAlteredJobPreparer.java
index 18b056e3715..5b7da97826e 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/RuleAlteredJobPreparer.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/RuleAlteredJobPreparer.java
@@ -19,6 +19,7 @@ package org.apache.shardingsphere.data.pipeline.scenario.rulealtered;
import lombok.extern.slf4j.Slf4j;
import org.apache.shardingsphere.data.pipeline.api.config.TableNameSchemaNameMapping;
+import org.apache.shardingsphere.data.pipeline.api.config.rulealtered.ImporterConfiguration;
import org.apache.shardingsphere.data.pipeline.api.config.rulealtered.RuleAlteredJobConfiguration;
import org.apache.shardingsphere.data.pipeline.api.config.rulealtered.TaskConfiguration;
import org.apache.shardingsphere.data.pipeline.api.datasource.PipelineDataSourceWrapper;
@@ -147,7 +148,8 @@ public final class RuleAlteredJobPreparer {
DataSourceChecker dataSourceChecker = DataSourceCheckerFactory.newInstance(jobContext.getJobConfig().getTargetDatabaseType());
Collection<PipelineDataSourceWrapper> targetDataSources = Collections.singletonList(targetDataSource);
dataSourceChecker.checkConnection(targetDataSources);
- dataSourceChecker.checkTargetTable(targetDataSources, jobContext.getTaskConfig().getImporterConfig().getLogicTableNames());
+ ImporterConfiguration importerConfig = jobContext.getTaskConfig().getImporterConfig();
+ dataSourceChecker.checkTargetTable(targetDataSources, importerConfig.getTableNameSchemaNameMapping(), importerConfig.getLogicTableNames());
}
private void initInventoryTasks(final RuleAlteredJobContext jobContext) {
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/prepare/InventoryTaskSplitter.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/prepare/InventoryTaskSplitter.java
index 3f137f3e80f..3b6cbb99971 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/prepare/InventoryTaskSplitter.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/prepare/InventoryTaskSplitter.java
@@ -183,7 +183,7 @@ public final class InventoryTaskSplitter {
Collection<IngestPosition<?>> result = new LinkedList<>();
RuleAlteredJobConfiguration jobConfig = jobContext.getJobConfig();
String sql = PipelineSQLBuilderFactory.newInstance(jobConfig.getSourceDatabaseType())
- .buildSplitByPrimaryKeyRangeSQL(dumperConfig.getActualTableName(), dumperConfig.getPrimaryKey());
+ .buildSplitByPrimaryKeyRangeSQL(dumperConfig.getSchemaName(new LogicTableName(dumperConfig.getLogicTableName())), dumperConfig.getActualTableName(), dumperConfig.getPrimaryKey());
try (
Connection connection = dataSource.getConnection();
PreparedStatement ps = connection.prepareStatement(sql)) {
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/scaling/core/job/environment/ScalingEnvironmentManager.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/scaling/core/job/environment/ScalingEnvironmentManager.java
index dccd3b0c979..345a1922817 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/scaling/core/job/environment/ScalingEnvironmentManager.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/scaling/core/job/environment/ScalingEnvironmentManager.java
@@ -18,12 +18,16 @@
package org.apache.shardingsphere.scaling.core.job.environment;
import lombok.extern.slf4j.Slf4j;
+import org.apache.shardingsphere.data.pipeline.api.config.TableNameSchemaNameMapping;
import org.apache.shardingsphere.data.pipeline.api.config.rulealtered.RuleAlteredJobConfiguration;
import org.apache.shardingsphere.data.pipeline.api.datasource.PipelineDataSourceWrapper;
import org.apache.shardingsphere.data.pipeline.api.datasource.config.PipelineDataSourceConfigurationFactory;
import org.apache.shardingsphere.data.pipeline.api.datasource.config.yaml.YamlPipelineDataSourceConfiguration;
+import org.apache.shardingsphere.data.pipeline.core.context.PipelineContext;
import org.apache.shardingsphere.data.pipeline.core.datasource.PipelineDataSourceFactory;
import org.apache.shardingsphere.data.pipeline.core.sqlbuilder.PipelineSQLBuilderFactory;
+import org.apache.shardingsphere.data.pipeline.spi.sqlbuilder.PipelineSQLBuilder;
+import org.apache.shardingsphere.infra.metadata.ShardingSphereMetaData;
import java.sql.Connection;
import java.sql.PreparedStatement;
@@ -47,11 +51,15 @@ public final class ScalingEnvironmentManager {
Collection<String> tables = jobConfig.splitLogicTableNames();
log.info("cleanupTargetTables, tables={}", tables);
YamlPipelineDataSourceConfiguration target = jobConfig.getTarget();
+ PipelineSQLBuilder pipelineSQLBuilder = PipelineSQLBuilderFactory.newInstance(jobConfig.getTargetDatabaseType());
+ ShardingSphereMetaData metaData = PipelineContext.getContextManager().getMetaDataContexts().getMetaData(jobConfig.getDatabaseName());
+ TableNameSchemaNameMapping tableNameSchemaNameMapping = new TableNameSchemaNameMapping(TableNameSchemaNameMapping.convert(metaData.getSchemas()));
try (
PipelineDataSourceWrapper dataSource = PipelineDataSourceFactory.newInstance(PipelineDataSourceConfigurationFactory.newInstance(target.getType(), target.getParameter()));
Connection connection = dataSource.getConnection()) {
for (String each : tables) {
- String sql = PipelineSQLBuilderFactory.newInstance(jobConfig.getTargetDatabaseType()).buildTruncateSQL(each);
+ String sql = pipelineSQLBuilder.buildTruncateSQL(tableNameSchemaNameMapping.getSchemaName(each), each);
+ log.info("cleanupTargetTables, tableName={}, sql={}", each, sql);
try (PreparedStatement preparedStatement = connection.prepareStatement(sql)) {
preparedStatement.execute();
}
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/test/java/org/apache/shardingsphere/data/pipeline/core/check/consistency/algorithm/CRC32MatchDataConsistencyCalculateAlgorithmTest.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/test/java/org/apache/shardingsphere/data/pipeline/core/check/consistency/algorithm/CRC32MatchDataConsistencyCalculateAlgorithmTest.java
index 0fd9da9f68b..e8785785976 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/test/java/org/apache/shardingsphere/data/pipeline/core/check/consistency/algorithm/CRC32MatchDataConsistencyCalculateAlgorithmTest.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/test/java/org/apache/shardingsphere/data/pipeline/core/check/consistency/algorithm/CRC32MatchDataConsistencyCalculateAlgorithmTest.java
@@ -18,6 +18,7 @@
package org.apache.shardingsphere.data.pipeline.core.check.consistency.algorithm;
import org.apache.shardingsphere.data.pipeline.api.check.consistency.DataConsistencyCalculateParameter;
+import org.apache.shardingsphere.data.pipeline.api.config.TableNameSchemaNameMapping;
import org.apache.shardingsphere.data.pipeline.api.datasource.PipelineDataSourceWrapper;
import org.apache.shardingsphere.data.pipeline.core.exception.PipelineDataConsistencyCheckFailedException;
import org.junit.Before;
@@ -32,6 +33,7 @@ import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.Arrays;
+import java.util.Collections;
import java.util.Iterator;
import static org.hamcrest.CoreMatchers.is;
@@ -55,7 +57,8 @@ public final class CRC32MatchDataConsistencyCalculateAlgorithmTest {
@Before
public void setUp() throws SQLException {
- parameter = new DataConsistencyCalculateParameter(pipelineDataSource, "foo_tbl", Arrays.asList("foo_col", "bar_col"), "FIXTURE", "FIXTURE", "1");
+ parameter = new DataConsistencyCalculateParameter(pipelineDataSource, new TableNameSchemaNameMapping(Collections.emptyMap()),
+ "foo_tbl", Arrays.asList("foo_col", "bar_col"), "FIXTURE", "FIXTURE", "1");
when(pipelineDataSource.getConnection()).thenReturn(connection);
}
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/test/java/org/apache/shardingsphere/data/pipeline/core/check/consistency/algorithm/fixture/FixturePipelineSQLBuilder.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/test/java/org/apache/shardingsphere/data/pipeline/core/check/consistency/algorithm/fixture/FixturePipelineSQLBuilder.java
index 4ed46ae94ac..bd473aef981 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/test/java/org/apache/shardingsphere/data/pipeline/core/check/consistency/algorithm/fixture/FixturePipelineSQLBuilder.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/test/java/org/apache/shardingsphere/data/pipeline/core/check/consistency/algorithm/fixture/FixturePipelineSQLBuilder.java
@@ -32,12 +32,17 @@ import java.util.Set;
public final class FixturePipelineSQLBuilder implements PipelineSQLBuilder {
@Override
- public String buildInsertSQL(final DataRecord dataRecord, final Map<LogicTableName, Set<String>> shardingColumnsMap) {
+ public String buildInventoryDumpSQL(final String schemaName, final String tableName, final String uniqueKey) {
return "";
}
@Override
- public String buildUpdateSQL(final DataRecord dataRecord, final Collection<Column> conditionColumns, final Map<LogicTableName, Set<String>> shardingColumnsMap) {
+ public String buildInsertSQL(final String schemaName, final DataRecord dataRecord, final Map<LogicTableName, Set<String>> shardingColumnsMap) {
+ return "";
+ }
+
+ @Override
+ public String buildUpdateSQL(final String schemaName, final DataRecord dataRecord, final Collection<Column> conditionColumns, final Map<LogicTableName, Set<String>> shardingColumnsMap) {
return "";
}
@@ -47,37 +52,37 @@ public final class FixturePipelineSQLBuilder implements PipelineSQLBuilder {
}
@Override
- public String buildDeleteSQL(final DataRecord dataRecord, final Collection<Column> conditionColumns) {
+ public String buildDeleteSQL(final String schemaName, final DataRecord dataRecord, final Collection<Column> conditionColumns) {
return "";
}
@Override
- public String buildTruncateSQL(final String tableName) {
+ public String buildTruncateSQL(final String schemaName, final String tableName) {
return "";
}
@Override
- public String buildCountSQL(final String tableName) {
+ public String buildCountSQL(final String schemaName, final String tableName) {
return "";
}
@Override
- public String buildChunkedQuerySQL(final String tableName, final String uniqueKey, final Number startUniqueValue) {
+ public String buildChunkedQuerySQL(final String schemaName, final String tableName, final String uniqueKey, final Number startUniqueValue) {
return "";
}
@Override
- public String buildCheckEmptySQL(final String tableName) {
+ public String buildCheckEmptySQL(final String schemaName, final String tableName) {
return null;
}
@Override
- public String buildSplitByPrimaryKeyRangeSQL(final String tableName, final String primaryKey) {
+ public String buildSplitByPrimaryKeyRangeSQL(final String schemaName, final String tableName, final String primaryKey) {
return "";
}
@Override
- public Optional<String> buildCRC32SQL(final String tableName, final String column) {
+ public Optional<String> buildCRC32SQL(final String schemaName, final String tableName, final String column) {
return Optional.of(String.format("SELECT CRC32(%s) FROM %s", column, tableName));
}
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/importer/MySQLImporter.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/importer/MySQLImporter.java
index 24185e16a94..c0b295a962e 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/importer/MySQLImporter.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/importer/MySQLImporter.java
@@ -35,4 +35,9 @@ public final class MySQLImporter extends AbstractImporter {
queryProps.setProperty("rewriteBatchedStatements", Boolean.TRUE.toString());
importerConfig.getDataSourceConfig().appendJDBCQueryProperties(queryProps);
}
+
+ @Override
+ protected String getSchemaName(final String logicTableName) {
+ return null;
+ }
}
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/sqlbuilder/MySQLPipelineSQLBuilder.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/sqlbuilder/MySQLPipelineSQLBuilder.java
index 29ba522081f..7d51c66d969 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/sqlbuilder/MySQLPipelineSQLBuilder.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/sqlbuilder/MySQLPipelineSQLBuilder.java
@@ -42,8 +42,13 @@ public final class MySQLPipelineSQLBuilder extends AbstractPipelineSQLBuilder {
}
@Override
- public String buildInsertSQL(final DataRecord dataRecord, final Map<LogicTableName, Set<String>> shardingColumnsMap) {
- return super.buildInsertSQL(dataRecord, shardingColumnsMap) + buildDuplicateUpdateSQL(dataRecord, shardingColumnsMap);
+ protected boolean isSchemaEnabled() {
+ return false;
+ }
+
+ @Override
+ public String buildInsertSQL(final String schemaName, final DataRecord dataRecord, final Map<LogicTableName, Set<String>> shardingColumnsMap) {
+ return super.buildInsertSQL(schemaName, dataRecord, shardingColumnsMap) + buildDuplicateUpdateSQL(dataRecord, shardingColumnsMap);
}
private String buildDuplicateUpdateSQL(final DataRecord dataRecord, final Map<LogicTableName, Set<String>> shardingColumnsMap) {
@@ -60,7 +65,7 @@ public final class MySQLPipelineSQLBuilder extends AbstractPipelineSQLBuilder {
}
@Override
- public Optional<String> buildCRC32SQL(final String tableName, final String column) {
+ public Optional<String> buildCRC32SQL(final String schemaName, final String tableName, final String column) {
return Optional.of(String.format("SELECT BIT_XOR(CAST(CRC32(%s) AS UNSIGNED)) AS checksum FROM %s", quote(column), quote(tableName)));
}
diff --git a/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/fixture/FixturePipelineSQLBuilder.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-mysql/src/test/java/org/apache/shardingsphere/data/pipeline/mysql/fixture/FixturePipelineSQLBuilder.java
similarity index 89%
copy from shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/fixture/FixturePipelineSQLBuilder.java
copy to shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-mysql/src/test/java/org/apache/shardingsphere/data/pipeline/mysql/fixture/FixturePipelineSQLBuilder.java
index fb6f5d92440..4c0e9a033ff 100644
--- a/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/fixture/FixturePipelineSQLBuilder.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-mysql/src/test/java/org/apache/shardingsphere/data/pipeline/mysql/fixture/FixturePipelineSQLBuilder.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.shardingsphere.data.pipeline.core.fixture;
+package org.apache.shardingsphere.data.pipeline.mysql.fixture;
import org.apache.shardingsphere.data.pipeline.core.sqlbuilder.AbstractPipelineSQLBuilder;
@@ -31,6 +31,11 @@ public final class FixturePipelineSQLBuilder extends AbstractPipelineSQLBuilder
return "`";
}
+ @Override
+ protected boolean isSchemaEnabled() {
+ return false;
+ }
+
@Override
public String getType() {
return "H2";
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-mysql/src/test/java/org/apache/shardingsphere/data/pipeline/mysql/sqlbuilder/MySQLPipelineSQLBuilderTest.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-mysql/src/test/java/org/apache/shardingsphere/data/pipeline/mysql/sqlbuilder/MySQLPipelineSQLBuilderTest.java
index e2e41316400..6cd7f59f107 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-mysql/src/test/java/org/apache/shardingsphere/data/pipeline/mysql/sqlbuilder/MySQLPipelineSQLBuilderTest.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-mysql/src/test/java/org/apache/shardingsphere/data/pipeline/mysql/sqlbuilder/MySQLPipelineSQLBuilderTest.java
@@ -41,19 +41,19 @@ public final class MySQLPipelineSQLBuilderTest {
@Test
public void assertBuildInsertSQL() {
- String actual = sqlBuilder.buildInsertSQL(mockDataRecord("t1"), shardingColumnsMap);
+ String actual = sqlBuilder.buildInsertSQL(null, mockDataRecord("t1"), shardingColumnsMap);
assertThat(actual, is("INSERT INTO `t1`(`id`,`sc`,`c1`,`c2`,`c3`) VALUES(?,?,?,?,?) ON DUPLICATE KEY UPDATE `sc`=VALUES(`sc`),`c1`=VALUES(`c1`),`c2`=VALUES(`c2`),`c3`=VALUES(`c3`)"));
}
@Test
public void assertBuildInsertSQLHasShardingColumn() {
- String actual = sqlBuilder.buildInsertSQL(mockDataRecord("t2"), shardingColumnsMap);
+ String actual = sqlBuilder.buildInsertSQL(null, mockDataRecord("t2"), shardingColumnsMap);
assertThat(actual, is("INSERT INTO `t2`(`id`,`sc`,`c1`,`c2`,`c3`) VALUES(?,?,?,?,?) ON DUPLICATE KEY UPDATE `c1`=VALUES(`c1`),`c2`=VALUES(`c2`),`c3`=VALUES(`c3`)"));
}
@Test
public void assertBuildSumCrc32SQL() {
- Optional<String> actual = sqlBuilder.buildCRC32SQL("t2", "id");
+ Optional<String> actual = sqlBuilder.buildCRC32SQL(null, "t2", "id");
assertTrue(actual.isPresent());
assertThat(actual.get(), is("SELECT BIT_XOR(CAST(CRC32(`id`) AS UNSIGNED)) AS checksum FROM `t2`"));
}
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-mysql/src/test/resources/META-INF/services/org.apache.shardingsphere.data.pipeline.spi.sqlbuilder.PipelineSQLBuilder b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-mysql/src/test/resources/META-INF/services/org.apache.shardingsphere.data.pipeline.spi.sqlbuilder.PipelineSQLBuilder
new file mode 100644
index 00000000000..8a51f55760c
--- /dev/null
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-mysql/src/test/resources/META-INF/services/org.apache.shardingsphere.data.pipeline.spi.sqlbuilder.PipelineSQLBuilder
@@ -0,0 +1,18 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+org.apache.shardingsphere.data.pipeline.mysql.fixture.FixturePipelineSQLBuilder
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/importer/OpenGaussImporter.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-opengauss/src/main/java/org/apache/shardingsphere/data/pipeline/opengauss/importer/OpenGaussImporter.java
index 6b86b38ad08..6d5be6db7d1 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-opengauss/src/main/java/org/apache/shardingsphere/data/pipeline/opengauss/importer/OpenGaussImporter.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-opengauss/src/main/java/org/apache/shardingsphere/data/pipeline/opengauss/importer/OpenGaussImporter.java
@@ -19,6 +19,7 @@ package org.apache.shardingsphere.data.pipeline.opengauss.importer;
import org.apache.shardingsphere.data.pipeline.api.config.rulealtered.ImporterConfiguration;
import org.apache.shardingsphere.data.pipeline.api.ingest.channel.PipelineChannel;
+import org.apache.shardingsphere.data.pipeline.api.metadata.LogicTableName;
import org.apache.shardingsphere.data.pipeline.core.datasource.PipelineDataSourceManager;
import org.apache.shardingsphere.data.pipeline.core.importer.AbstractImporter;
@@ -30,4 +31,9 @@ public final class OpenGaussImporter extends AbstractImporter {
public OpenGaussImporter(final ImporterConfiguration importerConfig, final PipelineDataSourceManager dataSourceManager, final PipelineChannel channel) {
super(importerConfig, dataSourceManager, channel);
}
+
+ @Override
+ protected String getSchemaName(final String logicTableName) {
+ return getImporterConfig().getSchemaName(new LogicTableName(logicTableName));
+ }
}
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 b0f90beaca7..7110d84075c 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
@@ -121,6 +121,7 @@ public final class OpenGaussDataSourcePreparer extends AbstractDataSourcePrepare
// TODO replace constraint and index name
case CREATE_TABLE:
sql = addIfNotExistsForCreateTableSQL(sql);
+ // TODO schemaName in create table SQL
sql = replaceActualTableNameToLogicTableName(sql, each.getActualTableName(), each.getLogicTableName());
sql = skipCreateTableExtendSet(sql);
return sql;
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/sqlbuilder/OpenGaussPipelineSQLBuilder.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-opengauss/src/main/java/org/apache/shardingsphere/data/pipeline/opengauss/sqlbuilder/OpenGaussPipelineSQLBuilder.java
index d74b05a53ba..4c1008c02cf 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-opengauss/src/main/java/org/apache/shardingsphere/data/pipeline/opengauss/sqlbuilder/OpenGaussPipelineSQLBuilder.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-opengauss/src/main/java/org/apache/shardingsphere/data/pipeline/opengauss/sqlbuilder/OpenGaussPipelineSQLBuilder.java
@@ -43,8 +43,13 @@ public final class OpenGaussPipelineSQLBuilder extends AbstractPipelineSQLBuilde
}
@Override
- public String buildInsertSQL(final DataRecord dataRecord, final Map<LogicTableName, Set<String>> shardingColumnsMap) {
- return super.buildInsertSQL(dataRecord, shardingColumnsMap) + buildConflictSQL(shardingColumnsMap);
+ protected boolean isSchemaEnabled() {
+ return true;
+ }
+
+ @Override
+ public String buildInsertSQL(final String schemaName, final DataRecord dataRecord, final Map<LogicTableName, Set<String>> shardingColumnsMap) {
+ return super.buildInsertSQL(schemaName, dataRecord, shardingColumnsMap) + buildConflictSQL(shardingColumnsMap);
}
@Override
diff --git a/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/fixture/FixturePipelineSQLBuilder.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-opengauss/src/test/java/org/apache/shardingsphere/data/pipeline/opengauss/fixture/FixturePipelineSQLBuilder.java
similarity index 88%
copy from shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/fixture/FixturePipelineSQLBuilder.java
copy to shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-opengauss/src/test/java/org/apache/shardingsphere/data/pipeline/opengauss/fixture/FixturePipelineSQLBuilder.java
index fb6f5d92440..798a0d60fb4 100644
--- a/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/fixture/FixturePipelineSQLBuilder.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-opengauss/src/test/java/org/apache/shardingsphere/data/pipeline/opengauss/fixture/FixturePipelineSQLBuilder.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.shardingsphere.data.pipeline.core.fixture;
+package org.apache.shardingsphere.data.pipeline.opengauss.fixture;
import org.apache.shardingsphere.data.pipeline.core.sqlbuilder.AbstractPipelineSQLBuilder;
@@ -31,6 +31,11 @@ public final class FixturePipelineSQLBuilder extends AbstractPipelineSQLBuilder
return "`";
}
+ @Override
+ protected boolean isSchemaEnabled() {
+ return false;
+ }
+
@Override
public String getType() {
return "H2";
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-opengauss/src/test/resources/META-INF/services/org.apache.shardingsphere.data.pipeline.spi.sqlbuilder.PipelineSQLBuilder b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-opengauss/src/test/resources/META-INF/services/org.apache.shardingsphere.data.pipeline.spi.sqlbuilder.PipelineSQLBuilder
new file mode 100644
index 00000000000..89defeee5e9
--- /dev/null
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-opengauss/src/test/resources/META-INF/services/org.apache.shardingsphere.data.pipeline.spi.sqlbuilder.PipelineSQLBuilder
@@ -0,0 +1,18 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+org.apache.shardingsphere.data.pipeline.opengauss.fixture.FixturePipelineSQLBuilder
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/importer/PostgreSQLImporter.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/importer/PostgreSQLImporter.java
index f044cba5690..8e96369d70c 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/importer/PostgreSQLImporter.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/importer/PostgreSQLImporter.java
@@ -19,6 +19,7 @@ package org.apache.shardingsphere.data.pipeline.postgresql.importer;
import org.apache.shardingsphere.data.pipeline.api.config.rulealtered.ImporterConfiguration;
import org.apache.shardingsphere.data.pipeline.api.ingest.channel.PipelineChannel;
+import org.apache.shardingsphere.data.pipeline.api.metadata.LogicTableName;
import org.apache.shardingsphere.data.pipeline.core.datasource.PipelineDataSourceManager;
import org.apache.shardingsphere.data.pipeline.core.importer.AbstractImporter;
@@ -30,4 +31,9 @@ public final class PostgreSQLImporter extends AbstractImporter {
public PostgreSQLImporter(final ImporterConfiguration importerConfig, final PipelineDataSourceManager dataSourceManager, final PipelineChannel channel) {
super(importerConfig, dataSourceManager, channel);
}
+
+ @Override
+ protected String getSchemaName(final String logicTableName) {
+ return getImporterConfig().getSchemaName(new LogicTableName(logicTableName));
+ }
}
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 9b8368c3024..9f9051d07a3 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
@@ -82,6 +82,7 @@ public final class PostgreSQLDataSourcePreparer extends AbstractDataSourcePrepar
log.error("failed to get actual table definitions", ex);
throw new PipelineJobPrepareFailedException("get table definitions failed.", ex);
}
+ log.info("prepareTargetTables, actualTableDefinitions={}", actualTableDefinitions);
Map<String, Collection<String>> createLogicTableSQLs = getCreateLogicTableSQLs(actualTableDefinitions);
try (
Connection targetConnection = getTargetCachedDataSource(parameter.getJobConfig(), parameter.getDataSourceManager()).getConnection()) {
@@ -171,7 +172,7 @@ public final class PostgreSQLDataSourcePreparer extends AbstractDataSourcePrepar
}
private String queryCreateTableSql(final Connection sourceConnection, final String schemaName, final String actualTableName, final List<String> pkColumns) throws SQLException {
- final String sql = String.format(FETCH_CREATE_TABLE_TEMPLATE, actualTableName, actualTableName, schemaName);
+ final String sql = String.format(FETCH_CREATE_TABLE_TEMPLATE, schemaName + "." + actualTableName, actualTableName, schemaName);
log.info("queryCreateTableSql, sql={}", sql);
try (Statement statement = sourceConnection.createStatement(); ResultSet resultSet = statement.executeQuery(sql)) {
if (!resultSet.next()) {
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/sqlbuilder/PostgreSQLPipelineSQLBuilder.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/sqlbuilder/PostgreSQLPipelineSQLBuilder.java
index 0b069da9ea4..85d18c9fb47 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/sqlbuilder/PostgreSQLPipelineSQLBuilder.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/sqlbuilder/PostgreSQLPipelineSQLBuilder.java
@@ -42,8 +42,13 @@ public final class PostgreSQLPipelineSQLBuilder extends AbstractPipelineSQLBuild
}
@Override
- public String buildInsertSQL(final DataRecord dataRecord, final Map<LogicTableName, Set<String>> shardingColumnsMap) {
- return super.buildInsertSQL(dataRecord, shardingColumnsMap) + buildConflictSQL(dataRecord, shardingColumnsMap);
+ protected boolean isSchemaEnabled() {
+ return true;
+ }
+
+ @Override
+ public String buildInsertSQL(final String schemaName, final DataRecord dataRecord, final Map<LogicTableName, Set<String>> shardingColumnsMap) {
+ return super.buildInsertSQL(schemaName, dataRecord, shardingColumnsMap) + buildConflictSQL(dataRecord, shardingColumnsMap);
}
// Refer to https://www.postgresql.org/docs/current/sql-insert.html
diff --git a/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/fixture/FixturePipelineSQLBuilder.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-postgresql/src/test/java/org/apache/shardingsphere/data/pipeline/postgresql/fixture/FixturePipelineSQLBuilder.java
similarity index 88%
copy from shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/fixture/FixturePipelineSQLBuilder.java
copy to shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-postgresql/src/test/java/org/apache/shardingsphere/data/pipeline/postgresql/fixture/FixturePipelineSQLBuilder.java
index fb6f5d92440..17721880a65 100644
--- a/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/fixture/FixturePipelineSQLBuilder.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-postgresql/src/test/java/org/apache/shardingsphere/data/pipeline/postgresql/fixture/FixturePipelineSQLBuilder.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.shardingsphere.data.pipeline.core.fixture;
+package org.apache.shardingsphere.data.pipeline.postgresql.fixture;
import org.apache.shardingsphere.data.pipeline.core.sqlbuilder.AbstractPipelineSQLBuilder;
@@ -31,6 +31,11 @@ public final class FixturePipelineSQLBuilder extends AbstractPipelineSQLBuilder
return "`";
}
+ @Override
+ protected boolean isSchemaEnabled() {
+ return false;
+ }
+
@Override
public String getType() {
return "H2";
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-postgresql/src/test/java/org/apache/shardingsphere/data/pipeline/postgresql/sqlbuilder/PostgreSQLPipelineSQLBuilderTest.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-postgresql/src/test/java/org/apache/shardingsphere/data/pipeline/postgresql/sqlbuilder/PostgreSQLPipelineSQLBuilderTest.java
index d6a6b83886d..5088fc67703 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-postgresql/src/test/java/org/apache/shardingsphere/data/pipeline/postgresql/sqlbuilder/PostgreSQLPipelineSQLBuilderTest.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-postgresql/src/test/java/org/apache/shardingsphere/data/pipeline/postgresql/sqlbuilder/PostgreSQLPipelineSQLBuilderTest.java
@@ -42,8 +42,8 @@ public final class PostgreSQLPipelineSQLBuilderTest {
@Test
public void assertBuildInsertSQL() {
- String actual = sqlBuilder.buildInsertSQL(mockDataRecord(), shardingColumnsMap);
- assertThat(actual, is("INSERT INTO \"t_order\"(\"order_id\",\"user_id\",\"status\") VALUES(?,?,?) ON CONFLICT (order_id)"
+ String actual = sqlBuilder.buildInsertSQL("schema1", mockDataRecord(), shardingColumnsMap);
+ assertThat(actual, is("INSERT INTO \"schema1\".\"t_order\"(\"order_id\",\"user_id\",\"status\") VALUES(?,?,?) ON CONFLICT (order_id)"
+ " DO UPDATE SET \"status\"=EXCLUDED.\"status\""));
}
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-postgresql/src/test/resources/META-INF/services/org.apache.shardingsphere.data.pipeline.spi.sqlbuilder.PipelineSQLBuilder b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-postgresql/src/test/resources/META-INF/services/org.apache.shardingsphere.data.pipeline.spi.sqlbuilder.PipelineSQLBuilder
new file mode 100644
index 00000000000..1990a1c68e5
--- /dev/null
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-postgresql/src/test/resources/META-INF/services/org.apache.shardingsphere.data.pipeline.spi.sqlbuilder.PipelineSQLBuilder
@@ -0,0 +1,18 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+org.apache.shardingsphere.data.pipeline.postgresql.fixture.FixturePipelineSQLBuilder
diff --git a/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/check/datasource/AbstractDataSourceCheckerTest.java b/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/check/datasource/AbstractDataSourceCheckerTest.java
index 85412f4fe18..50199382cbc 100644
--- a/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/check/datasource/AbstractDataSourceCheckerTest.java
+++ b/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/check/datasource/AbstractDataSourceCheckerTest.java
@@ -18,6 +18,7 @@
package org.apache.shardingsphere.data.pipeline.core.check.datasource;
import lombok.SneakyThrows;
+import org.apache.shardingsphere.data.pipeline.api.config.TableNameSchemaNameMapping;
import org.apache.shardingsphere.data.pipeline.core.exception.PipelineJobPrepareFailedException;
import org.junit.Before;
import org.junit.Test;
@@ -96,7 +97,7 @@ public final class AbstractDataSourceCheckerTest {
when(dataSource.getConnection()).thenReturn(connection);
when(connection.prepareStatement("SELECT * FROM `t_order` LIMIT 1")).thenReturn(preparedStatement);
when(preparedStatement.executeQuery()).thenReturn(resultSet);
- dataSourceChecker.checkTargetTable(dataSources, Collections.singletonList("t_order"));
+ dataSourceChecker.checkTargetTable(dataSources, new TableNameSchemaNameMapping(Collections.emptyMap()), Collections.singletonList("t_order"));
}
@Test(expected = PipelineJobPrepareFailedException.class)
@@ -105,6 +106,6 @@ public final class AbstractDataSourceCheckerTest {
when(connection.prepareStatement("SELECT * FROM `t_order` LIMIT 1")).thenReturn(preparedStatement);
when(preparedStatement.executeQuery()).thenReturn(resultSet);
when(resultSet.next()).thenReturn(true);
- dataSourceChecker.checkTargetTable(dataSources, Collections.singletonList("t_order"));
+ dataSourceChecker.checkTargetTable(dataSources, new TableNameSchemaNameMapping(Collections.emptyMap()), Collections.singletonList("t_order"));
}
}
diff --git a/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/fixture/FixtureDataSourceChecker.java b/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/fixture/FixtureDataSourceChecker.java
index 30d29b1348b..e349ffce843 100644
--- a/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/fixture/FixtureDataSourceChecker.java
+++ b/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/fixture/FixtureDataSourceChecker.java
@@ -17,6 +17,7 @@
package org.apache.shardingsphere.data.pipeline.core.fixture;
+import org.apache.shardingsphere.data.pipeline.api.config.TableNameSchemaNameMapping;
import org.apache.shardingsphere.data.pipeline.spi.check.datasource.DataSourceChecker;
import javax.sql.DataSource;
@@ -37,7 +38,7 @@ public final class FixtureDataSourceChecker implements DataSourceChecker {
}
@Override
- public void checkTargetTable(final Collection<? extends DataSource> dataSources, final Collection<String> tableNames) {
+ public void checkTargetTable(final Collection<? extends DataSource> dataSources, final TableNameSchemaNameMapping tableNameSchemaNameMapping, final Collection<String> tableNames) {
}
@Override
diff --git a/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/fixture/FixturePipelineSQLBuilder.java b/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/fixture/FixturePipelineSQLBuilder.java
index fb6f5d92440..35c4247f621 100644
--- a/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/fixture/FixturePipelineSQLBuilder.java
+++ b/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/fixture/FixturePipelineSQLBuilder.java
@@ -31,6 +31,11 @@ public final class FixturePipelineSQLBuilder extends AbstractPipelineSQLBuilder
return "`";
}
+ @Override
+ protected boolean isSchemaEnabled() {
+ return false;
+ }
+
@Override
public String getType() {
return "H2";
diff --git a/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/importer/AbstractImporterTest.java b/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/importer/AbstractImporterTest.java
index 6bc6b204342..253c856ae9b 100644
--- a/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/importer/AbstractImporterTest.java
+++ b/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/importer/AbstractImporterTest.java
@@ -17,6 +17,7 @@
package org.apache.shardingsphere.data.pipeline.core.importer;
+import org.apache.shardingsphere.data.pipeline.api.config.TableNameSchemaNameMapping;
import org.apache.shardingsphere.data.pipeline.api.config.rulealtered.ImporterConfiguration;
import org.apache.shardingsphere.data.pipeline.api.datasource.PipelineDataSourceWrapper;
import org.apache.shardingsphere.data.pipeline.api.datasource.config.PipelineDataSourceConfiguration;
@@ -81,6 +82,10 @@ public final class AbstractImporterTest {
@Before
public void setUp() throws SQLException {
jdbcImporter = new AbstractImporter(mockImporterConfiguration(), dataSourceManager, channel) {
+ @Override
+ protected String getSchemaName(final String logicTableName) {
+ return null;
+ }
};
when(dataSourceManager.getDataSource(dataSourceConfig)).thenReturn(dataSource);
when(dataSource.getConnection()).thenReturn(connection);
@@ -170,6 +175,6 @@ public final class AbstractImporterTest {
private ImporterConfiguration mockImporterConfiguration() {
Map<LogicTableName, Set<String>> shardingColumnsMap = Collections.singletonMap(new LogicTableName("test_table"), Collections.singleton("user"));
- return new ImporterConfiguration(dataSourceConfig, shardingColumnsMap, 1000, 3);
+ return new ImporterConfiguration(dataSourceConfig, shardingColumnsMap, new TableNameSchemaNameMapping(Collections.emptyMap()), 1000, 3);
}
}
diff --git a/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/task/InventoryTaskTest.java b/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/task/InventoryTaskTest.java
index 5bd2285a73f..96a45891d01 100644
--- a/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/task/InventoryTaskTest.java
+++ b/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/task/InventoryTaskTest.java
@@ -31,6 +31,7 @@ import org.apache.shardingsphere.data.pipeline.core.util.JobConfigurationBuilder
import org.apache.shardingsphere.data.pipeline.core.util.PipelineContextUtil;
import org.apache.shardingsphere.data.pipeline.scenario.rulealtered.RuleAlteredJobContext;
import org.junit.AfterClass;
+import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
@@ -42,14 +43,13 @@ import static org.junit.Assert.assertFalse;
public final class InventoryTaskTest {
- private static TaskConfiguration taskConfig;
-
private static final PipelineDataSourceManager DATA_SOURCE_MANAGER = new PipelineDataSourceManager();
+ private TaskConfiguration taskConfig;
+
@BeforeClass
public static void beforeClass() {
PipelineContextUtil.mockModeConfigAndContextManager();
- taskConfig = new RuleAlteredJobContext(JobConfigurationBuilder.createJobConfiguration()).getTaskConfig();
}
@AfterClass
@@ -57,6 +57,11 @@ public final class InventoryTaskTest {
DATA_SOURCE_MANAGER.close();
}
+ @Before
+ public void setUp() {
+ taskConfig = new RuleAlteredJobContext(JobConfigurationBuilder.createJobConfiguration()).getTaskConfig();
+ }
+
@Test(expected = IngestException.class)
public void assertStartWithGetEstimatedRowsFailure() {
InventoryDumperConfiguration inventoryDumperConfig = new InventoryDumperConfiguration(taskConfig.getDumperConfig());
@@ -84,6 +89,7 @@ public final class InventoryTaskTest {
// TODO use t_order_0, and also others
inventoryDumperConfig.setActualTableName("t_order");
inventoryDumperConfig.setLogicTableName("t_order");
+ inventoryDumperConfig.setPrimaryKey("order_id");
IngestPosition<?> position = taskConfig.getDumperConfig().getPosition();
if (null == position) {
position = new PrimaryKeyPosition(0, 1000);
diff --git a/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/spi/sqlbuilder/PipelineSQLBuilderTest.java b/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/spi/sqlbuilder/PipelineSQLBuilderTest.java
index 056593aa015..879935e26a3 100644
--- a/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/spi/sqlbuilder/PipelineSQLBuilderTest.java
+++ b/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/spi/sqlbuilder/PipelineSQLBuilderTest.java
@@ -42,33 +42,33 @@ public final class PipelineSQLBuilderTest {
@Test
public void assertBuildInsertSQL() {
- String actual = pipelineSQLBuilder.buildInsertSQL(mockDataRecord("t2"), shardingColumnsMap);
+ String actual = pipelineSQLBuilder.buildInsertSQL(null, mockDataRecord("t2"), shardingColumnsMap);
assertThat(actual, is("INSERT INTO `t2`(`id`,`sc`,`c1`,`c2`,`c3`) VALUES(?,?,?,?,?)"));
}
@Test
public void assertBuildUpdateSQLWithPrimaryKey() {
- String actual = pipelineSQLBuilder.buildUpdateSQL(mockDataRecord("t2"), RecordUtil.extractPrimaryColumns(mockDataRecord("t2")), shardingColumnsMap);
+ String actual = pipelineSQLBuilder.buildUpdateSQL(null, mockDataRecord("t2"), RecordUtil.extractPrimaryColumns(mockDataRecord("t2")), shardingColumnsMap);
assertThat(actual, is("UPDATE `t2` SET `c1` = ?,`c2` = ?,`c3` = ? WHERE `id` = ?"));
}
@Test
public void assertBuildUpdateSQLWithShardingColumns() {
DataRecord dataRecord = mockDataRecord("t2");
- String actual = pipelineSQLBuilder.buildUpdateSQL(dataRecord, mockConditionColumns(dataRecord), shardingColumnsMap);
+ String actual = pipelineSQLBuilder.buildUpdateSQL(null, dataRecord, mockConditionColumns(dataRecord), shardingColumnsMap);
assertThat(actual, is("UPDATE `t2` SET `c1` = ?,`c2` = ?,`c3` = ? WHERE `id` = ? and `sc` = ?"));
}
@Test
public void assertBuildDeleteSQLWithPrimaryKey() {
- String actual = pipelineSQLBuilder.buildDeleteSQL(mockDataRecord("t3"), RecordUtil.extractPrimaryColumns(mockDataRecord("t3")));
+ String actual = pipelineSQLBuilder.buildDeleteSQL(null, mockDataRecord("t3"), RecordUtil.extractPrimaryColumns(mockDataRecord("t3")));
assertThat(actual, is("DELETE FROM `t3` WHERE `id` = ?"));
}
@Test
public void assertBuildDeleteSQLWithConditionColumns() {
DataRecord dataRecord = mockDataRecord("t3");
- String actual = pipelineSQLBuilder.buildDeleteSQL(dataRecord, mockConditionColumns(dataRecord));
+ String actual = pipelineSQLBuilder.buildDeleteSQL(null, dataRecord, mockConditionColumns(dataRecord));
assertThat(actual, is("DELETE FROM `t3` WHERE `id` = ? and `sc` = ?"));
}