You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@shardingsphere.apache.org by yx...@apache.org on 2022/11/03 08:47:20 UTC
[shardingsphere] branch master updated: Add sharding columns in SQL condition for migration incremental task (#21928)
This is an automated email from the ASF dual-hosted git repository.
yx9o 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 23775c88db6 Add sharding columns in SQL condition for migration incremental task (#21928)
23775c88db6 is described below
commit 23775c88db62d9cb00aec3f81b04fa04c6029372
Author: Hongsheng Zhong <zh...@apache.org>
AuthorDate: Thu Nov 3 16:47:05 2022 +0800
Add sharding columns in SQL condition for migration incremental task (#21928)
* Use tables rule for t_order in IT
* Fix ShardingSphereSQLException missed exception message
* Update table rule DistSQL
* Fix sharding algorithm "inline" case-sensitive
* Print exception
* Change sharding column; Change user_id generator
* Add ShardingColumnsExtractor and impl
* Use shardingColumnsMap for migration job
---
.../pipeline/ShardingColumnsExtractorImpl.java | 81 ++++++++++++++++++++++
....pipeline.spi.sharding.ShardingColumnsExtractor | 18 +++++
.../external/sql/ShardingSphereSQLException.java | 7 +-
.../ShardingSpherePipelineDataSourceCreator.java | 2 +-
.../spi/sharding/ShardingColumnsExtractor.java | 41 +++++++++++
.../sharding/ShardingColumnsExtractorFactory.java | 37 +++++-----
...RC32MatchDataConsistencyCalculateAlgorithm.java | 1 +
...DataMatchDataConsistencyCalculateAlgorithm.java | 1 +
.../scenario/migration/MigrationJobAPIImpl.java | 6 +-
.../migration/general/MySQLMigrationGeneralIT.java | 14 ++--
.../general/PostgreSQLMigrationGeneralIT.java | 13 ++--
.../framework/helper/ScalingCaseHelper.java | 6 +-
.../util/AutoIncrementKeyGenerateAlgorithm.java | 6 +-
.../resources/env/common/migration-command.xml | 4 +-
.../ShardingColumnsExtractorFactoryTest.java} | 28 +++-----
15 files changed, 200 insertions(+), 65 deletions(-)
diff --git a/features/sharding/core/src/main/java/org/apache/shardingsphere/sharding/data/pipeline/ShardingColumnsExtractorImpl.java b/features/sharding/core/src/main/java/org/apache/shardingsphere/sharding/data/pipeline/ShardingColumnsExtractorImpl.java
new file mode 100644
index 00000000000..0711d86e71f
--- /dev/null
+++ b/features/sharding/core/src/main/java/org/apache/shardingsphere/sharding/data/pipeline/ShardingColumnsExtractorImpl.java
@@ -0,0 +1,81 @@
+/*
+ * 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.
+ */
+
+package org.apache.shardingsphere.sharding.data.pipeline;
+
+import org.apache.shardingsphere.data.pipeline.api.metadata.LogicTableName;
+import org.apache.shardingsphere.data.pipeline.spi.sharding.ShardingColumnsExtractor;
+import org.apache.shardingsphere.infra.yaml.config.pojo.rule.YamlRuleConfiguration;
+import org.apache.shardingsphere.sharding.api.config.ShardingRuleConfiguration;
+import org.apache.shardingsphere.sharding.api.config.rule.ShardingAutoTableRuleConfiguration;
+import org.apache.shardingsphere.sharding.api.config.rule.ShardingTableRuleConfiguration;
+import org.apache.shardingsphere.sharding.api.config.strategy.sharding.ComplexShardingStrategyConfiguration;
+import org.apache.shardingsphere.sharding.api.config.strategy.sharding.ShardingStrategyConfiguration;
+import org.apache.shardingsphere.sharding.api.config.strategy.sharding.StandardShardingStrategyConfiguration;
+import org.apache.shardingsphere.sharding.yaml.swapper.ShardingRuleConfigurationConverter;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Sharding columns extractor implementation.
+ */
+public final class ShardingColumnsExtractorImpl implements ShardingColumnsExtractor {
+
+ @Override
+ public Map<LogicTableName, Set<String>> getShardingColumnsMap(final Collection<YamlRuleConfiguration> yamlRuleConfigs, final Set<LogicTableName> logicTableNames) {
+ ShardingRuleConfiguration shardingRuleConfig = ShardingRuleConfigurationConverter.findAndConvertShardingRuleConfiguration(yamlRuleConfigs);
+ Set<String> defaultDatabaseShardingColumns = extractShardingColumns(shardingRuleConfig.getDefaultDatabaseShardingStrategy());
+ Set<String> defaultTableShardingColumns = extractShardingColumns(shardingRuleConfig.getDefaultTableShardingStrategy());
+ Map<LogicTableName, Set<String>> result = new ConcurrentHashMap<>();
+ for (ShardingTableRuleConfiguration each : shardingRuleConfig.getTables()) {
+ LogicTableName logicTableName = new LogicTableName(each.getLogicTable());
+ if (!logicTableNames.contains(logicTableName)) {
+ continue;
+ }
+ Set<String> shardingColumns = new HashSet<>();
+ shardingColumns.addAll(null == each.getDatabaseShardingStrategy() ? defaultDatabaseShardingColumns : extractShardingColumns(each.getDatabaseShardingStrategy()));
+ shardingColumns.addAll(null == each.getTableShardingStrategy() ? defaultTableShardingColumns : extractShardingColumns(each.getTableShardingStrategy()));
+ result.put(logicTableName, shardingColumns);
+ }
+ for (ShardingAutoTableRuleConfiguration each : shardingRuleConfig.getAutoTables()) {
+ LogicTableName logicTableName = new LogicTableName(each.getLogicTable());
+ if (!logicTableNames.contains(logicTableName)) {
+ continue;
+ }
+ ShardingStrategyConfiguration shardingStrategy = each.getShardingStrategy();
+ Set<String> shardingColumns = new HashSet<>(extractShardingColumns(shardingStrategy));
+ result.put(logicTableName, shardingColumns);
+ }
+ return result;
+ }
+
+ private Set<String> extractShardingColumns(final ShardingStrategyConfiguration shardingStrategy) {
+ if (shardingStrategy instanceof StandardShardingStrategyConfiguration) {
+ return new HashSet<>(Collections.singleton(((StandardShardingStrategyConfiguration) shardingStrategy).getShardingColumn()));
+ }
+ if (shardingStrategy instanceof ComplexShardingStrategyConfiguration) {
+ return new HashSet<>(Arrays.asList(((ComplexShardingStrategyConfiguration) shardingStrategy).getShardingColumns().split(",")));
+ }
+ return Collections.emptySet();
+ }
+}
diff --git a/features/sharding/core/src/main/resources/META-INF/services/org.apache.shardingsphere.data.pipeline.spi.sharding.ShardingColumnsExtractor b/features/sharding/core/src/main/resources/META-INF/services/org.apache.shardingsphere.data.pipeline.spi.sharding.ShardingColumnsExtractor
new file mode 100644
index 00000000000..c52c2307e57
--- /dev/null
+++ b/features/sharding/core/src/main/resources/META-INF/services/org.apache.shardingsphere.data.pipeline.spi.sharding.ShardingColumnsExtractor
@@ -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.sharding.data.pipeline.ShardingColumnsExtractorImpl
diff --git a/infra/util/src/main/java/org/apache/shardingsphere/infra/util/exception/external/sql/ShardingSphereSQLException.java b/infra/util/src/main/java/org/apache/shardingsphere/infra/util/exception/external/sql/ShardingSphereSQLException.java
index 76c412c2224..80ae0ee691c 100644
--- a/infra/util/src/main/java/org/apache/shardingsphere/infra/util/exception/external/sql/ShardingSphereSQLException.java
+++ b/infra/util/src/main/java/org/apache/shardingsphere/infra/util/exception/external/sql/ShardingSphereSQLException.java
@@ -40,9 +40,14 @@ public abstract class ShardingSphereSQLException extends ShardingSphereExternalE
}
public ShardingSphereSQLException(final String sqlState, final int typeOffset, final int errorCode, final String reason, final Object... messageArguments) {
+ this(sqlState, typeOffset, errorCode, null == reason ? null : String.format(reason, messageArguments));
+ }
+
+ private ShardingSphereSQLException(final String sqlState, final int typeOffset, final int errorCode, final String reason) {
+ super(reason);
this.sqlState = sqlState;
vendorCode = typeOffset * 10000 + errorCode;
- this.reason = null == reason ? null : String.format(reason, messageArguments);
+ this.reason = reason;
}
/**
diff --git a/jdbc/core/src/main/java/org/apache/shardingsphere/driver/data/pipeline/datasource/creator/ShardingSpherePipelineDataSourceCreator.java b/jdbc/core/src/main/java/org/apache/shardingsphere/driver/data/pipeline/datasource/creator/ShardingSpherePipelineDataSourceCreator.java
index 5ae9d20b1b0..908b5460d2f 100644
--- a/jdbc/core/src/main/java/org/apache/shardingsphere/driver/data/pipeline/datasource/creator/ShardingSpherePipelineDataSourceCreator.java
+++ b/jdbc/core/src/main/java/org/apache/shardingsphere/driver/data/pipeline/datasource/creator/ShardingSpherePipelineDataSourceCreator.java
@@ -59,7 +59,7 @@ public final class ShardingSpherePipelineDataSourceCreator implements PipelineDa
private void enableRangeQueryForInline(final YamlShardingRuleConfiguration shardingRuleConfig) {
for (YamlAlgorithmConfiguration each : shardingRuleConfig.getShardingAlgorithms().values()) {
- if ("INLINE".equals(each.getType())) {
+ if ("INLINE".equalsIgnoreCase(each.getType())) {
each.getProps().put("allow-range-query-with-inline-sharding", Boolean.TRUE.toString());
}
}
diff --git a/kernel/data-pipeline/api/src/main/java/org/apache/shardingsphere/data/pipeline/spi/sharding/ShardingColumnsExtractor.java b/kernel/data-pipeline/api/src/main/java/org/apache/shardingsphere/data/pipeline/spi/sharding/ShardingColumnsExtractor.java
new file mode 100644
index 00000000000..353bd5e1b9f
--- /dev/null
+++ b/kernel/data-pipeline/api/src/main/java/org/apache/shardingsphere/data/pipeline/spi/sharding/ShardingColumnsExtractor.java
@@ -0,0 +1,41 @@
+/*
+ * 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.
+ */
+
+package org.apache.shardingsphere.data.pipeline.spi.sharding;
+
+import org.apache.shardingsphere.data.pipeline.api.metadata.LogicTableName;
+import org.apache.shardingsphere.infra.util.spi.type.required.RequiredSPI;
+import org.apache.shardingsphere.infra.yaml.config.pojo.rule.YamlRuleConfiguration;
+
+import java.util.Collection;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * Sharding columns extractor.
+ */
+public interface ShardingColumnsExtractor extends RequiredSPI {
+
+ /**
+ * Get sharding columns map.
+ *
+ * @param yamlRuleConfigs YAML rule configurations
+ * @param logicTableNames logic table names
+ * @return sharding columns map
+ */
+ Map<LogicTableName, Set<String>> getShardingColumnsMap(Collection<YamlRuleConfiguration> yamlRuleConfigs, Set<LogicTableName> logicTableNames);
+}
diff --git a/test/integration-test/scaling/src/test/java/org/apache/shardingsphere/integration/data/pipeline/util/AutoIncrementKeyGenerateAlgorithm.java b/kernel/data-pipeline/api/src/main/java/org/apache/shardingsphere/data/pipeline/spi/sharding/ShardingColumnsExtractorFactory.java
similarity index 51%
copy from test/integration-test/scaling/src/test/java/org/apache/shardingsphere/integration/data/pipeline/util/AutoIncrementKeyGenerateAlgorithm.java
copy to kernel/data-pipeline/api/src/main/java/org/apache/shardingsphere/data/pipeline/spi/sharding/ShardingColumnsExtractorFactory.java
index 9fba5b1d037..882bffd8595 100644
--- a/test/integration-test/scaling/src/test/java/org/apache/shardingsphere/integration/data/pipeline/util/AutoIncrementKeyGenerateAlgorithm.java
+++ b/kernel/data-pipeline/api/src/main/java/org/apache/shardingsphere/data/pipeline/spi/sharding/ShardingColumnsExtractorFactory.java
@@ -15,28 +15,29 @@
* limitations under the License.
*/
-package org.apache.shardingsphere.integration.data.pipeline.util;
+package org.apache.shardingsphere.data.pipeline.spi.sharding;
-import org.apache.shardingsphere.sharding.spi.KeyGenerateAlgorithm;
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+import org.apache.shardingsphere.infra.util.spi.ShardingSphereServiceLoader;
+import org.apache.shardingsphere.infra.util.spi.type.required.RequiredSPIRegistry;
-import java.util.Properties;
-import java.util.concurrent.atomic.AtomicLong;
-
-public final class AutoIncrementKeyGenerateAlgorithm implements KeyGenerateAlgorithm {
-
- private final AtomicLong idGen = new AtomicLong(1);
-
- @Override
- public Comparable<?> generateKey() {
- return idGen.getAndIncrement();
- }
+/**
+ * Sharding columns extractor factory.
+ */
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public final class ShardingColumnsExtractorFactory {
- @Override
- public Properties getProps() {
- return null;
+ static {
+ ShardingSphereServiceLoader.register(ShardingColumnsExtractor.class);
}
- @Override
- public void init(final Properties props) {
+ /**
+ * Get sharding columns extractor instance.
+ *
+ * @return sharding columns extractor
+ */
+ public static ShardingColumnsExtractor getInstance() {
+ return RequiredSPIRegistry.getRegisteredService(ShardingColumnsExtractor.class);
}
}
diff --git a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/check/consistency/algorithm/CRC32MatchDataConsistencyCalculateAlgorithm.java b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/check/consistency/algorithm/CRC32MatchDataConsistencyCalculateAlgorithm.java
index 493e006f77a..1026be07b83 100644
--- a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/check/consistency/algorithm/CRC32MatchDataConsistencyCalculateAlgorithm.java
+++ b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/check/consistency/algorithm/CRC32MatchDataConsistencyCalculateAlgorithm.java
@@ -82,6 +82,7 @@ public final class CRC32MatchDataConsistencyCalculateAlgorithm extends AbstractD
int recordsCount = resultSet.getInt(2);
return new CalculatedItem(crc32, recordsCount);
} catch (final SQLException ex) {
+ log.error("calculateCRC32 failed, tableName={}", logicTableName, ex);
throw new PipelineTableDataConsistencyCheckLoadingFailedException(logicTableName);
}
}
diff --git a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/check/consistency/algorithm/DataMatchDataConsistencyCalculateAlgorithm.java b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/check/consistency/algorithm/DataMatchDataConsistencyCalculateAlgorithm.java
index 1415cee9d0c..1875a55f58a 100644
--- a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/check/consistency/algorithm/DataMatchDataConsistencyCalculateAlgorithm.java
+++ b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/check/consistency/algorithm/DataMatchDataConsistencyCalculateAlgorithm.java
@@ -122,6 +122,7 @@ public final class DataMatchDataConsistencyCalculateAlgorithm extends AbstractSt
}
return records.isEmpty() ? Optional.empty() : Optional.of(new CalculatedResult(maxUniqueKeyValue, records.size(), records));
} catch (final SQLException ex) {
+ log.error("calculateChunk failed, schemaName={}, tableName={}", parameter.getSchemaName(), parameter.getLogicTableName(), ex);
throw new PipelineTableDataConsistencyCheckLoadingFailedException(parameter.getLogicTableName());
}
}
diff --git a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/MigrationJobAPIImpl.java b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/MigrationJobAPIImpl.java
index 0e5b65d7b84..d20f547f4c1 100644
--- a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/MigrationJobAPIImpl.java
+++ b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/MigrationJobAPIImpl.java
@@ -64,6 +64,7 @@ import org.apache.shardingsphere.data.pipeline.core.metadata.loader.StandardPipe
import org.apache.shardingsphere.data.pipeline.core.sqlbuilder.PipelineSQLBuilderFactory;
import org.apache.shardingsphere.data.pipeline.scenario.consistencycheck.ConsistencyCheckJobAPIFactory;
import org.apache.shardingsphere.data.pipeline.scenario.consistencycheck.ConsistencyCheckJobItemContext;
+import org.apache.shardingsphere.data.pipeline.spi.sharding.ShardingColumnsExtractorFactory;
import org.apache.shardingsphere.data.pipeline.spi.sqlbuilder.PipelineSQLBuilder;
import org.apache.shardingsphere.data.pipeline.yaml.job.YamlMigrationJobConfiguration;
import org.apache.shardingsphere.data.pipeline.yaml.job.YamlMigrationJobConfigurationSwapper;
@@ -188,8 +189,9 @@ public final class MigrationJobAPIImpl extends AbstractInventoryIncrementalJobAP
TableNameSchemaNameMapping tableNameSchemaNameMapping = new TableNameSchemaNameMapping(tableNameSchemaMap);
CreateTableConfiguration createTableConfig = buildCreateTableConfiguration(jobConfig);
DumperConfiguration dumperConfig = buildDumperConfiguration(jobConfig.getJobId(), jobConfig.getSourceResourceName(), jobConfig.getSource(), tableNameMap, tableNameSchemaNameMapping);
- // TODO now shardingColumnsMap always empty,
- ImporterConfiguration importerConfig = buildImporterConfiguration(jobConfig, pipelineProcessConfig, Collections.emptyMap(), tableNameSchemaNameMapping);
+ Map<LogicTableName, Set<String>> shardingColumnsMap = ShardingColumnsExtractorFactory.getInstance().getShardingColumnsMap(
+ ((ShardingSpherePipelineDataSourceConfiguration) jobConfig.getTarget()).getRootConfig().getRules(), Collections.singleton(new LogicTableName(jobConfig.getTargetTableName())));
+ ImporterConfiguration importerConfig = buildImporterConfiguration(jobConfig, pipelineProcessConfig, shardingColumnsMap, tableNameSchemaNameMapping);
MigrationTaskConfiguration result = new MigrationTaskConfiguration(jobConfig.getSourceResourceName(), createTableConfig, dumperConfig, importerConfig);
log.info("buildTaskConfiguration, sourceResourceName={}, result={}", jobConfig.getSourceResourceName(), result);
return result;
diff --git a/test/integration-test/scaling/src/test/java/org/apache/shardingsphere/integration/data/pipeline/cases/migration/general/MySQLMigrationGeneralIT.java b/test/integration-test/scaling/src/test/java/org/apache/shardingsphere/integration/data/pipeline/cases/migration/general/MySQLMigrationGeneralIT.java
index 6cb495c3062..401fae91a1f 100644
--- a/test/integration-test/scaling/src/test/java/org/apache/shardingsphere/integration/data/pipeline/cases/migration/general/MySQLMigrationGeneralIT.java
+++ b/test/integration-test/scaling/src/test/java/org/apache/shardingsphere/integration/data/pipeline/cases/migration/general/MySQLMigrationGeneralIT.java
@@ -26,7 +26,6 @@ import org.apache.shardingsphere.integration.data.pipeline.env.enums.ITEnvTypeEn
import org.apache.shardingsphere.integration.data.pipeline.framework.helper.ScalingCaseHelper;
import org.apache.shardingsphere.integration.data.pipeline.framework.param.ScalingParameterized;
import org.apache.shardingsphere.integration.data.pipeline.util.AutoIncrementKeyGenerateAlgorithm;
-import org.apache.shardingsphere.sharding.spi.KeyGenerateAlgorithm;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
@@ -87,7 +86,7 @@ public final class MySQLMigrationGeneralIT extends AbstractMigrationITCase {
createTargetOrderTableRule();
createTargetOrderTableEncryptRule();
createTargetOrderItemTableRule();
- KeyGenerateAlgorithm keyGenerateAlgorithm = new AutoIncrementKeyGenerateAlgorithm();
+ AutoIncrementKeyGenerateAlgorithm keyGenerateAlgorithm = new AutoIncrementKeyGenerateAlgorithm();
JdbcTemplate jdbcTemplate = new JdbcTemplate(getSourceDataSource());
Pair<List<Object[]>, List<Object[]>> dataPair = ScalingCaseHelper.generateFullInsertData(keyGenerateAlgorithm, parameterized.getDatabaseType(), 3000);
log.info("init data begin: {}", LocalDateTime.now());
@@ -96,19 +95,18 @@ public final class MySQLMigrationGeneralIT extends AbstractMigrationITCase {
log.info("init data end: {}", LocalDateTime.now());
startMigration(getSourceTableOrderName(), getTargetTableOrderName());
startMigration("t_order_item", "t_order_item");
+ // TODO wait preparation done (binlog position got), else insert/update/delete might be consumed in inventory task
startIncrementTask(new MySQLIncrementTask(jdbcTemplate, getSourceTableOrderName(), keyGenerateAlgorithm, 30));
String orderJobId = getJobIdByTableName(getSourceTableOrderName());
String orderItemJobId = getJobIdByTableName("t_order_item");
assertMigrationSuccessById(orderJobId, "DATA_MATCH");
assertMigrationSuccessById(orderItemJobId, "DATA_MATCH");
assertMigrationSuccessById(orderItemJobId, "CRC32_MATCH");
- if (ENV.getItEnvType() == ITEnvTypeEnum.DOCKER) {
- for (String each : listJobId()) {
- commitMigrationByJobId(each);
- }
- List<String> lastJobIds = listJobId();
- assertThat(lastJobIds.size(), is(0));
+ for (String each : listJobId()) {
+ commitMigrationByJobId(each);
}
+ List<String> lastJobIds = listJobId();
+ assertThat(lastJobIds.size(), is(0));
assertGreaterThanOrderTableInitRows(TABLE_INIT_ROW_COUNT, "");
}
diff --git a/test/integration-test/scaling/src/test/java/org/apache/shardingsphere/integration/data/pipeline/cases/migration/general/PostgreSQLMigrationGeneralIT.java b/test/integration-test/scaling/src/test/java/org/apache/shardingsphere/integration/data/pipeline/cases/migration/general/PostgreSQLMigrationGeneralIT.java
index 34fe2e7816e..f4ab9cada7d 100644
--- a/test/integration-test/scaling/src/test/java/org/apache/shardingsphere/integration/data/pipeline/cases/migration/general/PostgreSQLMigrationGeneralIT.java
+++ b/test/integration-test/scaling/src/test/java/org/apache/shardingsphere/integration/data/pipeline/cases/migration/general/PostgreSQLMigrationGeneralIT.java
@@ -27,7 +27,6 @@ import org.apache.shardingsphere.integration.data.pipeline.env.enums.ITEnvTypeEn
import org.apache.shardingsphere.integration.data.pipeline.framework.helper.ScalingCaseHelper;
import org.apache.shardingsphere.integration.data.pipeline.framework.param.ScalingParameterized;
import org.apache.shardingsphere.integration.data.pipeline.util.AutoIncrementKeyGenerateAlgorithm;
-import org.apache.shardingsphere.sharding.spi.KeyGenerateAlgorithm;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
@@ -50,7 +49,7 @@ import static org.hamcrest.MatcherAssert.assertThat;
@Slf4j
public final class PostgreSQLMigrationGeneralIT extends AbstractMigrationITCase {
- private static final KeyGenerateAlgorithm KEY_GENERATE_ALGORITHM = new AutoIncrementKeyGenerateAlgorithm();
+ private static final AutoIncrementKeyGenerateAlgorithm KEY_GENERATE_ALGORITHM = new AutoIncrementKeyGenerateAlgorithm();
private final ScalingParameterized parameterized;
@@ -100,13 +99,11 @@ public final class PostgreSQLMigrationGeneralIT extends AbstractMigrationITCase
log.info("init data end: {}", LocalDateTime.now());
checkOrderMigration(jdbcTemplate);
checkOrderItemMigration();
- if (ENV.getItEnvType() == ITEnvTypeEnum.DOCKER) {
- for (String each : listJobId()) {
- commitMigrationByJobId(each);
- }
- List<String> lastJobIds = listJobId();
- assertThat(lastJobIds.size(), is(0));
+ for (String each : listJobId()) {
+ commitMigrationByJobId(each);
}
+ List<String> lastJobIds = listJobId();
+ assertThat(lastJobIds.size(), is(0));
assertGreaterThanOrderTableInitRows(TABLE_INIT_ROW_COUNT, SCHEMA_NAME);
}
diff --git a/test/integration-test/scaling/src/test/java/org/apache/shardingsphere/integration/data/pipeline/framework/helper/ScalingCaseHelper.java b/test/integration-test/scaling/src/test/java/org/apache/shardingsphere/integration/data/pipeline/framework/helper/ScalingCaseHelper.java
index fd8ed623eb7..8742fc131a7 100644
--- a/test/integration-test/scaling/src/test/java/org/apache/shardingsphere/integration/data/pipeline/framework/helper/ScalingCaseHelper.java
+++ b/test/integration-test/scaling/src/test/java/org/apache/shardingsphere/integration/data/pipeline/framework/helper/ScalingCaseHelper.java
@@ -21,8 +21,8 @@ import org.apache.commons.lang3.RandomStringUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.shardingsphere.infra.database.type.DatabaseType;
import org.apache.shardingsphere.infra.database.type.dialect.MySQLDatabaseType;
+import org.apache.shardingsphere.integration.data.pipeline.util.AutoIncrementKeyGenerateAlgorithm;
import org.apache.shardingsphere.sharding.algorithm.keygen.SnowflakeKeyGenerateAlgorithm;
-import org.apache.shardingsphere.sharding.spi.KeyGenerateAlgorithm;
import java.math.BigDecimal;
import java.sql.Timestamp;
@@ -57,7 +57,7 @@ public final class ScalingCaseHelper {
* @param insertRows insert rows
* @return insert data list
*/
- public static Pair<List<Object[]>, List<Object[]>> generateFullInsertData(final KeyGenerateAlgorithm orderIdGenerate, final DatabaseType databaseType, final int insertRows) {
+ public static Pair<List<Object[]>, List<Object[]>> generateFullInsertData(final AutoIncrementKeyGenerateAlgorithm orderIdGenerate, final DatabaseType databaseType, final int insertRows) {
if (insertRows < 0) {
return Pair.of(null, null);
}
@@ -65,7 +65,7 @@ public final class ScalingCaseHelper {
List<Object[]> orderItemData = new ArrayList<>(insertRows);
for (int i = 0; i < insertRows; i++) {
Comparable<?> orderId = orderIdGenerate.generateKey();
- int userId = generateInt(0, 6);
+ int userId = orderIdGenerate.generateKey();
LocalDateTime now = LocalDateTime.now();
int randomInt = generateInt(-100, 100);
int randomUnsignedInt = generateInt(0, 100);
diff --git a/test/integration-test/scaling/src/test/java/org/apache/shardingsphere/integration/data/pipeline/util/AutoIncrementKeyGenerateAlgorithm.java b/test/integration-test/scaling/src/test/java/org/apache/shardingsphere/integration/data/pipeline/util/AutoIncrementKeyGenerateAlgorithm.java
index 9fba5b1d037..00011b10ff4 100644
--- a/test/integration-test/scaling/src/test/java/org/apache/shardingsphere/integration/data/pipeline/util/AutoIncrementKeyGenerateAlgorithm.java
+++ b/test/integration-test/scaling/src/test/java/org/apache/shardingsphere/integration/data/pipeline/util/AutoIncrementKeyGenerateAlgorithm.java
@@ -20,14 +20,14 @@ package org.apache.shardingsphere.integration.data.pipeline.util;
import org.apache.shardingsphere.sharding.spi.KeyGenerateAlgorithm;
import java.util.Properties;
-import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicInteger;
public final class AutoIncrementKeyGenerateAlgorithm implements KeyGenerateAlgorithm {
- private final AtomicLong idGen = new AtomicLong(1);
+ private final AtomicInteger idGen = new AtomicInteger(1);
@Override
- public Comparable<?> generateKey() {
+ public Integer generateKey() {
return idGen.getAndIncrement();
}
diff --git a/test/integration-test/scaling/src/test/resources/env/common/migration-command.xml b/test/integration-test/scaling/src/test/resources/env/common/migration-command.xml
index 11a0fa76b34..17757c96b60 100644
--- a/test/integration-test/scaling/src/test/resources/env/common/migration-command.xml
+++ b/test/integration-test/scaling/src/test/resources/env/common/migration-command.xml
@@ -66,10 +66,10 @@
<create-target-order-table-rule>
CREATE SHARDING TABLE RULE t_order(
STORAGE_UNITS(ds_2,ds_3,ds_4),
- SHARDING_COLUMN=order_id,
+ SHARDING_COLUMN=user_id,
TYPE(NAME="hash_mod",PROPERTIES("sharding-count"="6")),
KEY_GENERATE_STRATEGY(COLUMN=order_id,TYPE(NAME="snowflake"))
- )
+ );
</create-target-order-table-rule>
<create-target-order-item-table-rule>
diff --git a/test/integration-test/scaling/src/test/java/org/apache/shardingsphere/integration/data/pipeline/util/AutoIncrementKeyGenerateAlgorithm.java b/test/pipeline/src/test/java/org/apache/shardingsphere/data/pipeline/spi/sharding/ShardingColumnsExtractorFactoryTest.java
similarity index 57%
copy from test/integration-test/scaling/src/test/java/org/apache/shardingsphere/integration/data/pipeline/util/AutoIncrementKeyGenerateAlgorithm.java
copy to test/pipeline/src/test/java/org/apache/shardingsphere/data/pipeline/spi/sharding/ShardingColumnsExtractorFactoryTest.java
index 9fba5b1d037..4532c7a84c7 100644
--- a/test/integration-test/scaling/src/test/java/org/apache/shardingsphere/integration/data/pipeline/util/AutoIncrementKeyGenerateAlgorithm.java
+++ b/test/pipeline/src/test/java/org/apache/shardingsphere/data/pipeline/spi/sharding/ShardingColumnsExtractorFactoryTest.java
@@ -15,28 +15,18 @@
* limitations under the License.
*/
-package org.apache.shardingsphere.integration.data.pipeline.util;
+package org.apache.shardingsphere.data.pipeline.spi.sharding;
-import org.apache.shardingsphere.sharding.spi.KeyGenerateAlgorithm;
+import org.apache.shardingsphere.sharding.data.pipeline.ShardingColumnsExtractorImpl;
+import org.junit.Test;
-import java.util.Properties;
-import java.util.concurrent.atomic.AtomicLong;
+import static org.hamcrest.CoreMatchers.instanceOf;
+import static org.hamcrest.MatcherAssert.assertThat;
-public final class AutoIncrementKeyGenerateAlgorithm implements KeyGenerateAlgorithm {
+public final class ShardingColumnsExtractorFactoryTest {
- private final AtomicLong idGen = new AtomicLong(1);
-
- @Override
- public Comparable<?> generateKey() {
- return idGen.getAndIncrement();
- }
-
- @Override
- public Properties getProps() {
- return null;
- }
-
- @Override
- public void init(final Properties props) {
+ @Test
+ public void assertGetInstance() {
+ assertThat(ShardingColumnsExtractorFactory.getInstance(), instanceOf(ShardingColumnsExtractorImpl.class));
}
}