You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@shardingsphere.apache.org by GitBox <gi...@apache.org> on 2022/08/20 05:43:26 UTC
[GitHub] [shardingsphere] azexcy opened a new pull request, #20316: Refactor the migration job and relation implementation
azexcy opened a new pull request, #20316:
URL: https://github.com/apache/shardingsphere/pull/20316
Changes proposed in this pull request:
- Refactor the migration job
- Change job id generate
- Fix unit test
- Remove unused code
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@shardingsphere.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [shardingsphere] sandynz commented on a diff in pull request #20316: Refactor the migration job and relation implementation
Posted by GitBox <gi...@apache.org>.
sandynz commented on code in PR #20316:
URL: https://github.com/apache/shardingsphere/pull/20316#discussion_r950655347
##########
shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/MigrationJobId.java:
##########
@@ -31,11 +30,7 @@
@ToString(callSuper = true)
public final class MigrationJobId extends AbstractPipelineJobId {
- public static final String CURRENT_VERSION = "01";
Review Comment:
Could we keep CURRENT_VERSION for now, it might still be used in job id.
##########
shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/MigrationJobAPIImpl.java:
##########
@@ -432,6 +472,69 @@ public void dropMigrationSourceResources(final Collection<String> resourceNames)
pipelineResourceAPI.persistMetaDataDataSource(JobType.MIGRATION, metaDataDataSource);
}
+ @Override
+ public void createJobAndStart(final CreateMigrationJobParameter parameter) {
+ YamlMigrationJobConfiguration result = new YamlMigrationJobConfiguration();
+ Map<String, DataSourceProperties> metaDataDataSource = pipelineResourceAPI.getMetaDataDataSource(JobType.MIGRATION);
+ Map<String, Object> sourceDataSourceProps = DATA_SOURCE_CONFIG_SWAPPER.swapToMap(metaDataDataSource.get(parameter.getSourceDataSourceName()));
+ YamlPipelineDataSourceConfiguration sourcePipelineDataSourceConfiguration = createYamlPipelineDataSourceConfiguration(StandardPipelineDataSourceConfiguration.TYPE,
+ YamlEngine.marshal(sourceDataSourceProps));
+ result.setSource(sourcePipelineDataSourceConfiguration);
+ result.setSourceDatabaseType(new StandardPipelineDataSourceConfiguration(sourceDataSourceProps).getDatabaseType().getType());
+ result.setSourceDataSourceName(parameter.getSourceDataSourceName());
+ result.setSourceTableName(parameter.getSourceTableName());
+ Map<String, Map<String, Object>> targetDataSourceProperties = new HashMap<>();
+ for (Entry<String, DataSource> entry : parameter.getTargetDataSources().entrySet()) {
+ Map<String, Object> dataSourceProps = DATA_SOURCE_CONFIG_SWAPPER.swapToMap(DataSourcePropertiesCreator.create(entry.getValue()));
+ targetDataSourceProperties.put(entry.getKey(), dataSourceProps);
+ }
+ String targetDatabaseName = parameter.getTargetDatabaseName();
+ YamlRootConfiguration targetRootConfig = getYamlRootConfiguration(targetDatabaseName, targetDataSourceProperties, parameter.getTargetShardingRuleConfig());
+ PipelineDataSourceConfiguration targetPipelineDataSource = new ShardingSpherePipelineDataSourceConfiguration(targetRootConfig);
+ result.setTarget(createYamlPipelineDataSourceConfiguration(targetPipelineDataSource.getType(), YamlEngine.marshal(targetPipelineDataSource.getDataSourceConfiguration())));
+ result.setTargetDatabaseType(targetPipelineDataSource.getDatabaseType().getType());
+ result.setTargetDatabaseName(targetDatabaseName);
+ result.setTargetTableName(parameter.getTargetTableName());
+ result.setSchemaTablesMap(getSchemaNameMap(sourcePipelineDataSourceConfiguration, parameter.getSourceTableName()));
+ extendYamlJobConfiguration(result);
+ MigrationJobConfiguration jobConfiguration = new YamlMigrationJobConfigurationSwapper().swapToObject(result);
+ start(jobConfiguration);
+ }
+
+ private YamlRootConfiguration getYamlRootConfiguration(final String databaseName, final Map<String, Map<String, Object>> yamlDataSources, final YamlRuleConfiguration shardingRule) {
+ YamlRootConfiguration result = new YamlRootConfiguration();
+ result.setDatabaseName(databaseName);
+ result.setDataSources(yamlDataSources);
+ Collection<YamlRuleConfiguration> yamlRuleConfigs = Collections.singletonList(shardingRule);
+ result.setRules(yamlRuleConfigs);
+ return result;
+ }
+
+ private YamlPipelineDataSourceConfiguration createYamlPipelineDataSourceConfiguration(final String type, final String parameter) {
+ YamlPipelineDataSourceConfiguration result = new YamlPipelineDataSourceConfiguration();
+ result.setType(type);
+ result.setParameter(parameter);
+ return result;
+ }
+
+ private Map<String, List<String>> getSchemaNameMap(final YamlPipelineDataSourceConfiguration pipelineDataSourceConfig, final String tableName) {
+ Map<String, List<String>> result = new HashMap<>();
+ try (PipelineDataSourceWrapper dataSource = PipelineDataSourceFactory.newInstance(PIPELINE_DATA_SOURCE_CONFIG_SWAPPER.swapToObject(pipelineDataSourceConfig))) {
+ try (Connection connection = dataSource.getConnection()) {
+ DatabaseMetaData metaData = connection.getMetaData();
+ ResultSet resultSet = metaData.getTables(null, null, tableName, new String[]{"TABLE"});
+ while (resultSet.next()) {
+ String schemaName = resultSet.getString("TABLE_SCHEM");
+ result.computeIfAbsent(schemaName, k -> new ArrayList<>()).add(resultSet.getString("TABLE_NAME"));
+ }
+ }
+ } catch (final SQLException ex) {
+ log.error("Get schema name map error", ex);
+ throw new AddMigrationSourceResourceException(ex.getMessage());
+ }
+ return result;
+ }
Review Comment:
Could we put it in SchemaTableUtil?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@shardingsphere.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [shardingsphere] sandynz commented on a diff in pull request #20316: Refactor the migration job and relation implementation
Posted by GitBox <gi...@apache.org>.
sandynz commented on code in PR #20316:
URL: https://github.com/apache/shardingsphere/pull/20316#discussion_r950947797
##########
shardingsphere-proxy/shardingsphere-proxy-backend/src/main/java/org/apache/shardingsphere/proxy/backend/handler/distsql/rdl/rule/RuleDefinitionBackendHandler.java:
##########
@@ -60,9 +51,6 @@
@Slf4j
public final class RuleDefinitionBackendHandler<T extends RuleDefinitionStatement> extends DatabaseRequiredBackendHandler<T> {
- private static final Set<String> RULE_ALTERED_ACTIONS = new HashSet<>(Arrays.asList(AlterShardingTableRuleStatement.class.getName(), AlterShardingAlgorithmStatement.class.getName(),
- AlterDefaultShardingStrategyStatement.class.getName(), AlterShardingBindingTableRulesStatement.class.getName(), AlterShardingBroadcastTableRulesStatement.class.getName()));
Review Comment:
We could keep it, it'll be used later
##########
shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/check/consistency/DataConsistencyChecker.java:
##########
@@ -77,8 +81,9 @@ public final class DataConsistencyChecker {
public DataConsistencyChecker(final MigrationJobConfiguration jobConfig, final JobRateLimitAlgorithm readRateLimitAlgorithm) {
this.jobConfig = jobConfig;
- logicTableNames = jobConfig.splitLogicTableNames();
- tableNameSchemaNameMapping = new TableNameSchemaNameMapping(TableNameSchemaNameMapping.convert(jobConfig.getSchemaTablesMap()));
+ logicTableNames = Collections.singletonList(jobConfig.getTargetTableName());
+ Map<String, List<String>> schemaTablesMap = SchemaTableUtil.getSchemaTablesMap(jobConfig.getTargetDatabaseName(), Collections.singleton(jobConfig.getTargetTableName()));
+ tableNameSchemaNameMapping = new TableNameSchemaNameMapping(TableNameSchemaNameMapping.convert(schemaTablesMap));
Review Comment:
We could still use `jobConfig.getSchemaTablesMap()` to translate table name and schema name, and use `SchemaTableUtil.getSchemaTablesMap(jobConfig.getTargetDatabaseName(), Collections.singleton(jobConfig.getTargetTableName()))` to get database metadata.
Since database metadata schema will use logical database name as schema name in MySQL.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@shardingsphere.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [shardingsphere] codecov-commenter commented on pull request #20316: Refactor the migration job and relation implementation
Posted by GitBox <gi...@apache.org>.
codecov-commenter commented on PR #20316:
URL: https://github.com/apache/shardingsphere/pull/20316#issuecomment-1221474954
# [Codecov](https://codecov.io/gh/apache/shardingsphere/pull/20316?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#20316](https://codecov.io/gh/apache/shardingsphere/pull/20316?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (49b2b51) into [master](https://codecov.io/gh/apache/shardingsphere/commit/831883021d08fac066c8336a057bf5c3c85a3536?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (8318830) will **increase** coverage by `0.16%`.
> The diff coverage is `6.74%`.
```diff
@@ Coverage Diff @@
## master #20316 +/- ##
============================================
+ Coverage 60.38% 60.55% +0.16%
- Complexity 2438 2439 +1
============================================
Files 3917 3915 -2
Lines 55359 55198 -161
Branches 9411 9372 -39
============================================
- Hits 33428 33424 -4
+ Misses 19072 18916 -156
+ Partials 2859 2858 -1
```
| [Impacted Files](https://codecov.io/gh/apache/shardingsphere/pull/20316?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [...on/distsql/handler/update/MigrateTableUpdater.java](https://codecov.io/gh/apache/shardingsphere/pull/20316/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2hhcmRpbmdzcGhlcmUtZmVhdHVyZXMvc2hhcmRpbmdzcGhlcmUtc2hhcmRpbmcvc2hhcmRpbmdzcGhlcmUtc2hhcmRpbmctZGlzdHNxbC9zaGFyZGluZ3NwaGVyZS1zaGFyZGluZy1kaXN0c3FsLWhhbmRsZXIvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3NoYXJkaW5nc3BoZXJlL21pZ3JhdGlvbi9kaXN0c3FsL2hhbmRsZXIvdXBkYXRlL01pZ3JhdGVUYWJsZVVwZGF0ZXIuamF2YQ==) | `0.00% <0.00%> (ø)` | |
| [...eator/ShardingSpherePipelineDataSourceCreator.java](https://codecov.io/gh/apache/shardingsphere/pull/20316/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2hhcmRpbmdzcGhlcmUtamRiYy9zaGFyZGluZ3NwaGVyZS1qZGJjLWNvcmUvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3NoYXJkaW5nc3BoZXJlL2RyaXZlci9kYXRhL3BpcGVsaW5lL2RhdGFzb3VyY2UvY3JlYXRvci9TaGFyZGluZ1NwaGVyZVBpcGVsaW5lRGF0YVNvdXJjZUNyZWF0b3IuamF2YQ==) | `0.00% <0.00%> (ø)` | |
| [...ata/pipeline/api/config/ImporterConfiguration.java](https://codecov.io/gh/apache/shardingsphere/pull/20316/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2hhcmRpbmdzcGhlcmUta2VybmVsL3NoYXJkaW5nc3BoZXJlLWRhdGEtcGlwZWxpbmUvc2hhcmRpbmdzcGhlcmUtZGF0YS1waXBlbGluZS1hcGkvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3NoYXJkaW5nc3BoZXJlL2RhdGEvcGlwZWxpbmUvYXBpL2NvbmZpZy9JbXBvcnRlckNvbmZpZ3VyYXRpb24uamF2YQ==) | `0.00% <0.00%> (ø)` | |
| [...ipeline/api/config/ingest/DumperConfiguration.java](https://codecov.io/gh/apache/shardingsphere/pull/20316/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2hhcmRpbmdzcGhlcmUta2VybmVsL3NoYXJkaW5nc3BoZXJlLWRhdGEtcGlwZWxpbmUvc2hhcmRpbmdzcGhlcmUtZGF0YS1waXBlbGluZS1hcGkvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3NoYXJkaW5nc3BoZXJlL2RhdGEvcGlwZWxpbmUvYXBpL2NvbmZpZy9pbmdlc3QvRHVtcGVyQ29uZmlndXJhdGlvbi5qYXZh) | `0.00% <ø> (ø)` | |
| [...pi/config/ingest/InventoryDumperConfiguration.java](https://codecov.io/gh/apache/shardingsphere/pull/20316/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2hhcmRpbmdzcGhlcmUta2VybmVsL3NoYXJkaW5nc3BoZXJlLWRhdGEtcGlwZWxpbmUvc2hhcmRpbmdzcGhlcmUtZGF0YS1waXBlbGluZS1hcGkvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3NoYXJkaW5nc3BoZXJlL2RhdGEvcGlwZWxpbmUvYXBpL2NvbmZpZy9pbmdlc3QvSW52ZW50b3J5RHVtcGVyQ29uZmlndXJhdGlvbi5qYXZh) | `0.00% <ø> (ø)` | |
| [...config/job/yaml/YamlMigrationJobConfiguration.java](https://codecov.io/gh/apache/shardingsphere/pull/20316/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2hhcmRpbmdzcGhlcmUta2VybmVsL3NoYXJkaW5nc3BoZXJlLWRhdGEtcGlwZWxpbmUvc2hhcmRpbmdzcGhlcmUtZGF0YS1waXBlbGluZS1hcGkvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3NoYXJkaW5nc3BoZXJlL2RhdGEvcGlwZWxpbmUvYXBpL2NvbmZpZy9qb2IveWFtbC9ZYW1sTWlncmF0aW9uSm9iQ29uZmlndXJhdGlvbi5qYXZh) | `28.57% <ø> (+1.90%)` | :arrow_up: |
| [...eline/spi/ingest/position/PositionInitializer.java](https://codecov.io/gh/apache/shardingsphere/pull/20316/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2hhcmRpbmdzcGhlcmUta2VybmVsL3NoYXJkaW5nc3BoZXJlLWRhdGEtcGlwZWxpbmUvc2hhcmRpbmdzcGhlcmUtZGF0YS1waXBlbGluZS1hcGkvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3NoYXJkaW5nc3BoZXJlL2RhdGEvcGlwZWxpbmUvc3BpL2luZ2VzdC9wb3NpdGlvbi9Qb3NpdGlvbkluaXRpYWxpemVyLmphdmE=) | `0.00% <ø> (ø)` | |
| [...core/check/consistency/DataConsistencyChecker.java](https://codecov.io/gh/apache/shardingsphere/pull/20316/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2hhcmRpbmdzcGhlcmUta2VybmVsL3NoYXJkaW5nc3BoZXJlLWRhdGEtcGlwZWxpbmUvc2hhcmRpbmdzcGhlcmUtZGF0YS1waXBlbGluZS1jb3JlL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9zaGFyZGluZ3NwaGVyZS9kYXRhL3BpcGVsaW5lL2NvcmUvY2hlY2svY29uc2lzdGVuY3kvRGF0YUNvbnNpc3RlbmN5Q2hlY2tlci5qYXZh) | `0.00% <0.00%> (ø)` | |
| [...hm/DataMatchDataConsistencyCalculateAlgorithm.java](https://codecov.io/gh/apache/shardingsphere/pull/20316/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2hhcmRpbmdzcGhlcmUta2VybmVsL3NoYXJkaW5nc3BoZXJlLWRhdGEtcGlwZWxpbmUvc2hhcmRpbmdzcGhlcmUtZGF0YS1waXBlbGluZS1jb3JlL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9zaGFyZGluZ3NwaGVyZS9kYXRhL3BpcGVsaW5lL2NvcmUvY2hlY2svY29uc2lzdGVuY3kvYWxnb3JpdGhtL0RhdGFNYXRjaERhdGFDb25zaXN0ZW5jeUNhbGN1bGF0ZUFsZ29yaXRobS5qYXZh) | `0.00% <0.00%> (ø)` | |
| [.../data/pipeline/core/execute/PipelineJobWorker.java](https://codecov.io/gh/apache/shardingsphere/pull/20316/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2hhcmRpbmdzcGhlcmUta2VybmVsL3NoYXJkaW5nc3BoZXJlLWRhdGEtcGlwZWxpbmUvc2hhcmRpbmdzcGhlcmUtZGF0YS1waXBlbGluZS1jb3JlL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9zaGFyZGluZ3NwaGVyZS9kYXRhL3BpcGVsaW5lL2NvcmUvZXhlY3V0ZS9QaXBlbGluZUpvYldvcmtlci5qYXZh) | `0.00% <ø> (ø)` | |
| ... and [16 more](https://codecov.io/gh/apache/shardingsphere/pull/20316/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |
:mega: We’re building smart automated test selection to slash your CI/CD build times. [Learn more](https://about.codecov.io/iterative-testing/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@shardingsphere.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [shardingsphere] sandynz merged pull request #20316: Refactor the migration job and relation implementation
Posted by GitBox <gi...@apache.org>.
sandynz merged PR #20316:
URL: https://github.com/apache/shardingsphere/pull/20316
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@shardingsphere.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [shardingsphere] azexcy commented on a diff in pull request #20316: Refactor the migration job and relation implementation
Posted by GitBox <gi...@apache.org>.
azexcy commented on code in PR #20316:
URL: https://github.com/apache/shardingsphere/pull/20316#discussion_r951103375
##########
shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/check/consistency/DataConsistencyChecker.java:
##########
@@ -77,8 +81,9 @@ public final class DataConsistencyChecker {
public DataConsistencyChecker(final MigrationJobConfiguration jobConfig, final JobRateLimitAlgorithm readRateLimitAlgorithm) {
this.jobConfig = jobConfig;
- logicTableNames = jobConfig.splitLogicTableNames();
- tableNameSchemaNameMapping = new TableNameSchemaNameMapping(TableNameSchemaNameMapping.convert(jobConfig.getSchemaTablesMap()));
+ logicTableNames = Collections.singletonList(jobConfig.getTargetTableName());
+ Map<String, List<String>> schemaTablesMap = SchemaTableUtil.getSchemaTablesMap(jobConfig.getTargetDatabaseName(), Collections.singleton(jobConfig.getTargetTableName()));
+ tableNameSchemaNameMapping = new TableNameSchemaNameMapping(TableNameSchemaNameMapping.convert(schemaTablesMap));
Review Comment:
I will fix id at next pr.
##########
shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/check/consistency/DataConsistencyChecker.java:
##########
@@ -77,8 +81,9 @@ public final class DataConsistencyChecker {
public DataConsistencyChecker(final MigrationJobConfiguration jobConfig, final JobRateLimitAlgorithm readRateLimitAlgorithm) {
this.jobConfig = jobConfig;
- logicTableNames = jobConfig.splitLogicTableNames();
- tableNameSchemaNameMapping = new TableNameSchemaNameMapping(TableNameSchemaNameMapping.convert(jobConfig.getSchemaTablesMap()));
+ logicTableNames = Collections.singletonList(jobConfig.getTargetTableName());
+ Map<String, List<String>> schemaTablesMap = SchemaTableUtil.getSchemaTablesMap(jobConfig.getTargetDatabaseName(), Collections.singleton(jobConfig.getTargetTableName()));
+ tableNameSchemaNameMapping = new TableNameSchemaNameMapping(TableNameSchemaNameMapping.convert(schemaTablesMap));
Review Comment:
I will fix this at next pr.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@shardingsphere.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [shardingsphere] sandynz commented on a diff in pull request #20316: Refactor the migration job and relation implementation
Posted by GitBox <gi...@apache.org>.
sandynz commented on code in PR #20316:
URL: https://github.com/apache/shardingsphere/pull/20316#discussion_r950807785
##########
shardingsphere-features/shardingsphere-sharding/shardingsphere-sharding-distsql/shardingsphere-sharding-distsql-handler/src/main/java/org/apache/shardingsphere/migration/distsql/handler/update/MigrateTableUpdater.java:
##########
@@ -17,22 +17,43 @@
package org.apache.shardingsphere.migration.distsql.handler.update;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.ObjectUtils;
import org.apache.shardingsphere.data.pipeline.api.MigrationJobPublicAPI;
import org.apache.shardingsphere.data.pipeline.api.PipelineJobPublicAPIFactory;
+import org.apache.shardingsphere.data.pipeline.api.pojo.CreateMigrationJobParameter;
+import org.apache.shardingsphere.data.pipeline.core.context.PipelineContext;
+import org.apache.shardingsphere.data.pipeline.core.exception.PipelineJobCreationException;
import org.apache.shardingsphere.infra.distsql.update.RALUpdater;
+import org.apache.shardingsphere.infra.metadata.database.ShardingSphereDatabase;
import org.apache.shardingsphere.migration.distsql.statement.MigrateTableStatement;
+import org.apache.shardingsphere.sharding.api.config.ShardingRuleConfiguration;
+import org.apache.shardingsphere.sharding.yaml.swapper.YamlShardingRuleConfigurationSwapper;
+
+import java.util.Objects;
/**
* Migrate table updater.
*/
+@Slf4j
public final class MigrateTableUpdater implements RALUpdater<MigrateTableStatement> {
private static final MigrationJobPublicAPI JOB_API = PipelineJobPublicAPIFactory.getMigrationJobPublicAPI();
+ private static final YamlShardingRuleConfigurationSwapper SHARDING_RULE_CONFIG_SWAPPER = new YamlShardingRuleConfigurationSwapper();
+
@Override
public void executeUpdate(final String databaseName, final MigrateTableStatement sqlStatement) {
- // TODO implement migrate table
- JOB_API.getType();
+ log.info("start migrate job by {}", sqlStatement);
+ ShardingSphereDatabase targetDatabase = PipelineContext.getContextManager().getMetaDataContexts().getMetaData().getDatabase(sqlStatement.getTargetDatabaseName());
+ ShardingRuleConfiguration targetShardingRule = targetDatabase.getRuleMetaData().getConfigurations().stream().filter(each -> each instanceof ShardingRuleConfiguration)
+ .map(each -> (ShardingRuleConfiguration) each).findFirst().orElseThrow(() -> new PipelineJobCreationException("No sharding rule found."));
+ targetShardingRule.getTables().removeIf(each -> !Objects.equals(each.getLogicTable(), sqlStatement.getTargetTableName()));
+ targetShardingRule.getAutoTables().removeIf(each -> !Objects.equals(each.getLogicTable(), sqlStatement.getTargetTableName()));
Review Comment:
`targetShardingRule` is from current context metadata, some tables and autoTables will be removed.
It's better to clone another ShardingRuleConfiguration and change it.
##########
shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/MigrationJobPreparer.java:
##########
@@ -138,24 +135,25 @@ private void prepareAndCheckTarget(final MigrationJobItemContext jobItemContext)
private void prepareTarget(final MigrationJobItemContext jobItemContext) throws SQLException {
MigrationJobConfiguration jobConfig = jobItemContext.getJobConfig();
TableNameSchemaNameMapping tableNameSchemaNameMapping = jobItemContext.getTaskConfig().getDumperConfig().getTableNameSchemaNameMapping();
+ // TODO fill schema
String targetDatabaseType = jobConfig.getTargetDatabaseType();
Review Comment:
Is `TODO fill schema` done?
##########
shardingsphere-test/shardingsphere-integration-test/shardingsphere-integration-test-scaling/src/test/java/org/apache/shardingsphere/integration/data/pipeline/cases/general/MySQLMigrationGeneralIT.java:
##########
@@ -67,35 +66,39 @@ public static Collection<ScalingParameterized> getParameters() {
}
@Test
- public void assertManualScalingSuccess() throws InterruptedException {
- addSourceResource();
- initShardingAlgorithm();
- assertTrue(waitShardingAlgorithmEffect(15));
+ public void assertMigrationSuccess() {
createScalingRule();
- createOrderTableRule();
- createOrderItemTableRule();
- createNoUseTable();
- createOrderTable();
- createOrderItemTable();
+ createSourceOrderTable();
+ createSourceOrderItemTable();
+ addSourceResource();
+ addTargetResource();
+ createTargetOrderTableRule();
+ createTargetOrderItemTableRule();
SnowflakeKeyGenerateAlgorithm keyGenerateAlgorithm = new SnowflakeKeyGenerateAlgorithm();
- for (int i = 0; i < TABLE_INIT_ROW_COUNT / 1000; i++) {
- Pair<List<Object[]>, List<Object[]>> dataPair = ScalingCaseHelper.generateFullInsertData(keyGenerateAlgorithm, parameterized.getDatabaseType(), 1000);
- getJdbcTemplate().batchUpdate(getExtraSQLCommand().getFullInsertOrder(), dataPair.getLeft());
- getJdbcTemplate().batchUpdate(getExtraSQLCommand().getFullInsertOrderItem(), dataPair.getRight());
+ JdbcTemplate jdbcTemplate = new JdbcTemplate(getSourceDataSource());
+ for (int i = 0; i < TABLE_INIT_ROW_COUNT / 10; i++) {
Review Comment:
`TABLE_INIT_ROW_COUNT / 10`? Looks need to recover as `1000`
##########
shardingsphere-test/shardingsphere-integration-test/shardingsphere-integration-test-scaling/src/test/java/org/apache/shardingsphere/integration/data/pipeline/cases/base/BaseITCase.java:
##########
@@ -131,12 +133,12 @@ public BaseITCase(final ScalingParameterized parameterized) {
if (ENV.getItEnvType() == ScalingITEnvTypeEnum.NATIVE) {
cleanUpDataSource();
}
- commonSQLCommand = JAXB.unmarshal(Objects.requireNonNull(BaseITCase.class.getClassLoader().getResource("env/common/command.xml")), CommonSQLCommand.class);
- scalingWatcher = new ScalingWatcher(composedContainer, jdbcTemplate);
+ migrationDistSQLCommand = JAXB.unmarshal(Objects.requireNonNull(BaseITCase.class.getClassLoader().getResource("env/common/command.xml")), MigrationDistSQLCommand.class);
+ scalingWatcher = new ScalingWatcher(composedContainer);
}
private void cleanUpDataSource() {
- for (String each : Arrays.asList(DS_0, DS_1, DS_2, DS_3, DS_4)) {
+ for (String each : Arrays.asList(DS_0, DS_2, DS_3, DS_4)) {
Review Comment:
Looks `DS_1` is not needed any more, we could remove it
##########
shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/check/consistency/algorithm/DataMatchDataConsistencyCalculateAlgorithm.java:
##########
@@ -188,6 +189,9 @@ public boolean equals(final @NonNull Object o) {
if (thisResult instanceof SQLXML && thatResult instanceof SQLXML) {
return ((SQLXML) thisResult).getString().equals(((SQLXML) thatResult).getString());
}
+ if (thisResult instanceof Integer && thatResult instanceof Long) {
+ return Objects.equals(Integer.toString((Integer) thisResult), Long.toString((Long) thatResult));
+ }
Review Comment:
1, Add TODO here;
2, Could we compare them by number?
##########
shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/MigrationJobAPIImpl.java:
##########
@@ -428,6 +464,52 @@ public void dropMigrationSourceResources(final Collection<String> resourceNames)
dataSourcePersistService.persist(getJobType(), metaDataDataSource);
}
+ @Override
+ public void createJobAndStart(final CreateMigrationJobParameter parameter) {
+ YamlMigrationJobConfiguration result = new YamlMigrationJobConfiguration();
Review Comment:
Variable name `result` could be `yamlJobConfig`, or else extract another method to create and return YamlMigrationJobConfiguration.
##########
shardingsphere-features/shardingsphere-sharding/shardingsphere-sharding-distsql/shardingsphere-sharding-distsql-handler/src/main/java/org/apache/shardingsphere/migration/distsql/handler/update/MigrateTableUpdater.java:
##########
@@ -17,22 +17,43 @@
package org.apache.shardingsphere.migration.distsql.handler.update;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.ObjectUtils;
import org.apache.shardingsphere.data.pipeline.api.MigrationJobPublicAPI;
import org.apache.shardingsphere.data.pipeline.api.PipelineJobPublicAPIFactory;
+import org.apache.shardingsphere.data.pipeline.api.pojo.CreateMigrationJobParameter;
+import org.apache.shardingsphere.data.pipeline.core.context.PipelineContext;
+import org.apache.shardingsphere.data.pipeline.core.exception.PipelineJobCreationException;
import org.apache.shardingsphere.infra.distsql.update.RALUpdater;
+import org.apache.shardingsphere.infra.metadata.database.ShardingSphereDatabase;
import org.apache.shardingsphere.migration.distsql.statement.MigrateTableStatement;
+import org.apache.shardingsphere.sharding.api.config.ShardingRuleConfiguration;
+import org.apache.shardingsphere.sharding.yaml.swapper.YamlShardingRuleConfigurationSwapper;
+
+import java.util.Objects;
/**
* Migrate table updater.
*/
+@Slf4j
public final class MigrateTableUpdater implements RALUpdater<MigrateTableStatement> {
private static final MigrationJobPublicAPI JOB_API = PipelineJobPublicAPIFactory.getMigrationJobPublicAPI();
+ private static final YamlShardingRuleConfigurationSwapper SHARDING_RULE_CONFIG_SWAPPER = new YamlShardingRuleConfigurationSwapper();
+
@Override
public void executeUpdate(final String databaseName, final MigrateTableStatement sqlStatement) {
- // TODO implement migrate table
- JOB_API.getType();
+ log.info("start migrate job by {}", sqlStatement);
+ ShardingSphereDatabase targetDatabase = PipelineContext.getContextManager().getMetaDataContexts().getMetaData().getDatabase(sqlStatement.getTargetDatabaseName());
+ ShardingRuleConfiguration targetShardingRule = targetDatabase.getRuleMetaData().getConfigurations().stream().filter(each -> each instanceof ShardingRuleConfiguration)
+ .map(each -> (ShardingRuleConfiguration) each).findFirst().orElseThrow(() -> new PipelineJobCreationException("No sharding rule found."));
+ targetShardingRule.getTables().removeIf(each -> !Objects.equals(each.getLogicTable(), sqlStatement.getTargetTableName()));
+ targetShardingRule.getAutoTables().removeIf(each -> !Objects.equals(each.getLogicTable(), sqlStatement.getTargetTableName()));
+ CreateMigrationJobParameter createMigrationJobParameter = new CreateMigrationJobParameter(sqlStatement.getSourceDatabaseName(), sqlStatement.getSourceTableName(),
+ ObjectUtils.defaultIfNull(sqlStatement.getTargetDatabaseName(), sqlStatement.getSourceDatabaseName()), sqlStatement.getTargetTableName(),
+ SHARDING_RULE_CONFIG_SWAPPER.swapToYamlConfiguration(targetShardingRule), targetDatabase.getResource().getDataSources());
Review Comment:
Looks the second parameter of `ObjectUtils.defaultIfNull(sqlStatement.getTargetDatabaseName(), sqlStatement.getSourceDatabaseName())` should be `databaseName` from method parameter.
It means if use doesn't define target database, then use current selected one.
##########
shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/pojo/CreateMigrationJobParameter.java:
##########
@@ -15,18 +15,28 @@
* limitations under the License.
*/
-package org.apache.shardingsphere.data.pipeline.spi.rulealtered;
+package org.apache.shardingsphere.data.pipeline.api.pojo;
-import org.apache.shardingsphere.data.pipeline.spi.fixture.RuleAlteredJobConfigurationPreparerFixture;
-import org.junit.Test;
+import lombok.Data;
+import lombok.RequiredArgsConstructor;
+import org.apache.shardingsphere.infra.yaml.config.pojo.rule.YamlRuleConfiguration;
-import static org.hamcrest.CoreMatchers.instanceOf;
-import static org.junit.Assert.assertThat;
+import javax.sql.DataSource;
+import java.util.Map;
-public final class RuleAlteredJobConfigurationPreparerFactoryTest {
+@Data
+@RequiredArgsConstructor
+public final class CreateMigrationJobParameter {
- @Test
- public void assertGetInstance() {
- assertThat(RuleAlteredJobConfigurationPreparerFactory.getInstance(), instanceOf(RuleAlteredJobConfigurationPreparerFixture.class));
- }
+ private final String sourceDataSourceName;
Review Comment:
Could we rename `sourceDataSourceName` to `sourceResourceName`? Keep it consistent with `ADD RESOURCE`
##########
shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/MigrationJobAPIImpl.java:
##########
@@ -428,6 +464,52 @@ public void dropMigrationSourceResources(final Collection<String> resourceNames)
dataSourcePersistService.persist(getJobType(), metaDataDataSource);
}
+ @Override
+ public void createJobAndStart(final CreateMigrationJobParameter parameter) {
+ YamlMigrationJobConfiguration result = new YamlMigrationJobConfiguration();
+ Map<String, DataSourceProperties> metaDataDataSource = dataSourcePersistService.load(JobType.MIGRATION);
+ Map<String, Object> sourceDataSourceProps = DATA_SOURCE_CONFIG_SWAPPER.swapToMap(metaDataDataSource.get(parameter.getSourceDataSourceName()));
+ YamlPipelineDataSourceConfiguration sourcePipelineDataSourceConfiguration = createYamlPipelineDataSourceConfiguration(StandardPipelineDataSourceConfiguration.TYPE,
+ YamlEngine.marshal(sourceDataSourceProps));
+ result.setSource(sourcePipelineDataSourceConfiguration);
+ result.setSourceDatabaseType(new StandardPipelineDataSourceConfiguration(sourceDataSourceProps).getDatabaseType().getType());
+ result.setSourceDataSourceName(parameter.getSourceDataSourceName());
+ result.setSourceTableName(parameter.getSourceTableName());
+ Map<String, Map<String, Object>> targetDataSourceProperties = new HashMap<>();
+ for (Entry<String, DataSource> entry : parameter.getTargetDataSources().entrySet()) {
+ Map<String, Object> dataSourceProps = DATA_SOURCE_CONFIG_SWAPPER.swapToMap(DataSourcePropertiesCreator.create(entry.getValue()));
+ targetDataSourceProperties.put(entry.getKey(), dataSourceProps);
+ }
+ String targetDatabaseName = parameter.getTargetDatabaseName();
+ YamlRootConfiguration targetRootConfig = getYamlRootConfiguration(targetDatabaseName, targetDataSourceProperties, parameter.getTargetShardingRuleConfig());
+ PipelineDataSourceConfiguration targetPipelineDataSource = new ShardingSpherePipelineDataSourceConfiguration(targetRootConfig);
+ result.setTarget(createYamlPipelineDataSourceConfiguration(targetPipelineDataSource.getType(), YamlEngine.marshal(targetPipelineDataSource.getDataSourceConfiguration())));
+ result.setTargetDatabaseType(targetPipelineDataSource.getDatabaseType().getType());
+ result.setTargetDatabaseName(targetDatabaseName);
+ result.setTargetTableName(parameter.getTargetTableName());
+ result.setSchemaTablesMap(SchemaTableUtil.getSchemaTablesMapFromActual(PIPELINE_DATA_SOURCE_CONFIG_SWAPPER.swapToObject(sourcePipelineDataSourceConfiguration),
+ parameter.getSourceTableName()));
+ extendYamlJobConfiguration(result);
+ MigrationJobConfiguration jobConfiguration = new YamlMigrationJobConfigurationSwapper().swapToObject(result);
+ start(jobConfiguration);
+ }
+
+ private YamlRootConfiguration getYamlRootConfiguration(final String databaseName, final Map<String, Map<String, Object>> yamlDataSources, final YamlRuleConfiguration shardingRule) {
+ YamlRootConfiguration result = new YamlRootConfiguration();
+ result.setDatabaseName(databaseName);
+ result.setDataSources(yamlDataSources);
+ Collection<YamlRuleConfiguration> yamlRuleConfigs = Collections.singletonList(shardingRule);
+ result.setRules(yamlRuleConfigs);
Review Comment:
```
Collection<YamlRuleConfiguration> yamlRuleConfigs = Collections.singletonList(shardingRule);
result.setRules(yamlRuleConfigs);
```
It will lose other rules, e.g. encrypt rule, so users could not do encryption on migration at the same time.
##########
shardingsphere-test/shardingsphere-integration-test/shardingsphere-integration-test-scaling/src/test/java/org/apache/shardingsphere/integration/data/pipeline/cases/general/MySQLMigrationGeneralIT.java:
##########
@@ -67,35 +66,39 @@ public static Collection<ScalingParameterized> getParameters() {
}
@Test
- public void assertManualScalingSuccess() throws InterruptedException {
- addSourceResource();
- initShardingAlgorithm();
- assertTrue(waitShardingAlgorithmEffect(15));
+ public void assertMigrationSuccess() {
createScalingRule();
- createOrderTableRule();
- createOrderItemTableRule();
- createNoUseTable();
- createOrderTable();
- createOrderItemTable();
+ createSourceOrderTable();
+ createSourceOrderItemTable();
+ addSourceResource();
+ addTargetResource();
+ createTargetOrderTableRule();
+ createTargetOrderItemTableRule();
SnowflakeKeyGenerateAlgorithm keyGenerateAlgorithm = new SnowflakeKeyGenerateAlgorithm();
- for (int i = 0; i < TABLE_INIT_ROW_COUNT / 1000; i++) {
- Pair<List<Object[]>, List<Object[]>> dataPair = ScalingCaseHelper.generateFullInsertData(keyGenerateAlgorithm, parameterized.getDatabaseType(), 1000);
- getJdbcTemplate().batchUpdate(getExtraSQLCommand().getFullInsertOrder(), dataPair.getLeft());
- getJdbcTemplate().batchUpdate(getExtraSQLCommand().getFullInsertOrderItem(), dataPair.getRight());
+ JdbcTemplate jdbcTemplate = new JdbcTemplate(getSourceDataSource());
+ for (int i = 0; i < TABLE_INIT_ROW_COUNT / 10; i++) {
+ Pair<List<Object[]>, List<Object[]>> dataPair = ScalingCaseHelper.generateFullInsertData(keyGenerateAlgorithm, parameterized.getDatabaseType(), 10);
+ jdbcTemplate.batchUpdate(getExtraSQLCommand().getFullInsertOrder(), dataPair.getLeft());
+ jdbcTemplate.batchUpdate(getExtraSQLCommand().getFullInsertOrderItem(), dataPair.getRight());
}
- addTargetResource();
- startIncrementTask(new MySQLIncrementTask(getJdbcTemplate(), keyGenerateAlgorithm, true, 20));
- executeWithLog(getCommonSQLCommand().getAlterOrderWithItemAutoTableRule());
- String jobId = getScalingJobId();
- waitScalingFinished(jobId);
- stopScaling(jobId);
- getJdbcTemplate().update("INSERT INTO t_order (id,order_id,user_id,status,t_json) VALUES (?, ?, ?, ?, ?)", keyGenerateAlgorithm.generateKey(), keyGenerateAlgorithm.generateKey(),
- 1, "afterStopScaling", "{}");
- startScaling(jobId);
+ startMigrationOrderItem();
+ checkOrderMigration(keyGenerateAlgorithm, jdbcTemplate);
+ checkOrderItemMigration();
+ }
+
+ private void checkOrderMigration(final KeyGenerateAlgorithm keyGenerateAlgorithm, final JdbcTemplate jdbcTemplate) {
+ startMigrationOrder();
+ startIncrementTask(new MySQLIncrementTask(jdbcTemplate, keyGenerateAlgorithm, true, 20));
+ String jobId = getJobIdByTableName("t_order");
+ waitMigrationFinished(jobId);
assertCheckScalingSuccess(jobId);
assertGreaterThanInitTableInitRows(TABLE_INIT_ROW_COUNT, "");
- applyScaling(jobId);
- assertPreviewTableSuccess("t_order", Arrays.asList("ds_2", "ds_3", "ds_4"));
- assertPreviewTableSuccess("t_order_item", Arrays.asList("ds_2", "ds_3", "ds_4"));
+ }
+
+ private void checkOrderItemMigration() {
+ startMigrationOrderItem();
+ String jobId = getJobIdByTableName("t_order");
+ waitMigrationFinished(jobId);
Review Comment:
Looks `getJobIdByTableName("t_order")` should be `getJobIdByTableName("t_order_item")`
##########
shardingsphere-test/shardingsphere-integration-test/shardingsphere-integration-test-scaling/src/test/java/org/apache/shardingsphere/integration/data/pipeline/cases/general/PostgreSQLMigrationGeneralIT.java:
##########
@@ -70,36 +68,39 @@ public static Collection<ScalingParameterized> getParameters() {
}
@Test
- public void assertManualScalingSuccess() throws InterruptedException {
- addSourceResource();
- initShardingAlgorithm();
- assertTrue(waitShardingAlgorithmEffect(15));
+ public void assertMigrationSuccess() {
createScalingRule();
- createSchema("test");
- createOrderTableRule();
- createOrderItemTableRule();
- createOrderTable();
- createOrderItemTable();
- // TODO wait kernel support create index if not exists
- // createTableIndexList("test");
- executeWithLog("COMMENT ON COLUMN test.t_order.user_id IS 'user id';");
+ createSourceSchema("test");
+ createSourceOrderTable();
+ createSourceOrderItemTable();
+ createSourceTableIndexList("test");
+ createSourceCommentOnList("test");
+ addSourceResource();
+ addTargetResource();
+ createTargetOrderTableRule();
+ createTargetOrderItemTableRule();
SnowflakeKeyGenerateAlgorithm keyGenerateAlgorithm = new SnowflakeKeyGenerateAlgorithm();
Pair<List<Object[]>, List<Object[]>> dataPair = ScalingCaseHelper.generateFullInsertData(keyGenerateAlgorithm, parameterized.getDatabaseType(), TABLE_INIT_ROW_COUNT);
- getJdbcTemplate().batchUpdate(getExtraSQLCommand().getFullInsertOrder(), dataPair.getLeft());
- getJdbcTemplate().batchUpdate(getExtraSQLCommand().getFullInsertOrderItem(), dataPair.getRight());
- addTargetResource();
- startIncrementTask(new PostgreSQLIncrementTask(getJdbcTemplate(), new SnowflakeKeyGenerateAlgorithm(), "test", true, 20));
- executeWithLog(getCommonSQLCommand().getAlterOrderWithItemAutoTableRule());
- String jobId = getScalingJobId();
- waitScalingFinished(jobId);
- stopScaling(jobId);
- executeWithLog(String.format("INSERT INTO test.t_order (id,order_id,user_id,status) VALUES (%s, %s, %s, '%s')", keyGenerateAlgorithm.generateKey(), System.currentTimeMillis(),
- 1, "afterStopScaling"));
- startScaling(jobId);
+ JdbcTemplate jdbcTemplate = new JdbcTemplate(getSourceDataSource());
+ jdbcTemplate.batchUpdate(getExtraSQLCommand().getFullInsertOrder(), dataPair.getLeft());
+ jdbcTemplate.batchUpdate(getExtraSQLCommand().getFullInsertOrderItem(), dataPair.getRight());
+ checkOrderMigration(jdbcTemplate);
+ checkOrderItemMigration();
+ }
+
+ private void checkOrderMigration(final JdbcTemplate jdbcTemplate) {
+ startMigrationOrder();
+ startIncrementTask(new PostgreSQLIncrementTask(jdbcTemplate, "test", false, 20));
+ String jobId = getJobIdByTableName("t_order");
+ waitMigrationFinished(jobId);
assertCheckScalingSuccess(jobId);
assertGreaterThanInitTableInitRows(TABLE_INIT_ROW_COUNT, "test");
- applyScaling(jobId);
- assertPreviewTableSuccess("t_order", Arrays.asList("ds_2", "ds_3", "ds_4"));
- assertPreviewTableSuccess("t_order_item", Arrays.asList("ds_2", "ds_3", "ds_4"));
+ }
+
+ private void checkOrderItemMigration() {
+ startMigrationOrderItem();
+ String jobId = getJobIdByTableName("t_order");
+ waitMigrationFinished(jobId);
Review Comment:
Looks `getJobIdByTableName("t_order")` should be `getJobIdByTableName("t_order_item")`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@shardingsphere.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [shardingsphere] sandynz commented on pull request #20316: Refactor the migration job and relation implementation
Posted by GitBox <gi...@apache.org>.
sandynz commented on PR #20316:
URL: https://github.com/apache/shardingsphere/pull/20316#issuecomment-1221509402
From scaling-it-test PostgreSQL CI log:
```
Error: 2022-08-21 05:20:31.130 [main] o.a.s.i.d.p.cases.base.BaseITCase - data access error
org.postgresql.util.PSQLException: ERROR: Count check failed for table 't_order'
at org.postgresql.core.v3.QueryExecutorImpl.receiveErrorResponse(QueryExecutorImpl.java:2676)
at org.postgresql.core.v3.QueryExecutorImpl.processResults(QueryExecutorImpl.java:2366)
at org.postgresql.core.v3.QueryExecutorImpl.execute(QueryExecutorImpl.java:356)
at org.postgresql.jdbc.PgStatement.executeInternal(PgStatement.java:490)
at org.postgresql.jdbc.PgStatement.execute(PgStatement.java:408)
at org.postgresql.jdbc.PgStatement.executeWithFlags(PgStatement.java:329)
at org.postgresql.jdbc.PgStatement.executeCachedSql(PgStatement.java:315)
at org.postgresql.jdbc.PgStatement.executeWithFlags(PgStatement.java:291)
at org.postgresql.jdbc.PgStatement.executeQuery(PgStatement.java:243)
at com.zaxxer.hikari.pool.ProxyStatement.executeQuery(ProxyStatement.java:111)
at com.zaxxer.hikari.pool.HikariProxyStatement.executeQuery(HikariProxyStatement.java)
at org.apache.shardingsphere.integration.data.pipeline.cases.base.BaseITCase.queryForListWithLog(BaseITCase.java:291)
at org.apache.shardingsphere.integration.data.pipeline.cases.base.BaseITCase.assertCheckScalingSuccess(BaseITCase.java:380)
at org.apache.shardingsphere.integration.data.pipeline.cases.general.PostgreSQLMigrationGeneralIT.checkOrderItemMigration(PostgreSQLMigrationGeneralIT.java:104)
at org.apache.shardingsphere.integration.data.pipeline.cases.general.PostgreSQLMigrationGeneralIT.assertMigrationSuccess(PostgreSQLMigrationGeneralIT.java:88)
```
It failed, but CI show successful. It need to do some improvement.
The root cause if `queryForListWithLog` catch SQLException and ignore it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@shardingsphere.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org