sandynz commented on code in PR #20407:
URL: https://github.com/apache/shardingsphere/pull/20407#discussion_r951284832
##########
shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/MigrationJobAPIImpl.java:
##########
@@ -171,7 +171,8 @@ private MigrationJobConfiguration getJobConfiguration(final
JobConfigurationPOJO
public TaskConfiguration buildTaskConfiguration(final
MigrationJobConfiguration jobConfig, final int jobShardingItem, final
PipelineProcessConfiguration pipelineProcessConfig) {
Map<ActualTableName, LogicTableName> tableNameMap = new
LinkedHashMap<>();
tableNameMap.put(new ActualTableName(jobConfig.getSourceTableName()),
new LogicTableName(jobConfig.getSourceTableName()));
- Map<LogicTableName, String> tableNameSchemaMap =
TableNameSchemaNameMapping.convert(SchemaTableUtil.getSchemaTablesMapFromActual(jobConfig.getSource(),
jobConfig.getSourceTableName()));
+ Map<LogicTableName, String> tableNameSchemaMap =
TableNameSchemaNameMapping.convert(PipelineSchemaTableUtil.getSchemaTablesMapFromActual(jobConfig.getSource(),
+ jobConfig.getSourceTableName()));
Review Comment:
Could we reuse `schemaTablesMap` in jobConfig?
##########
shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/check/consistency/algorithm/DataMatchDataConsistencyCalculateAlgorithm.java:
##########
@@ -120,7 +120,7 @@ private String getQuerySQL(final
DataConsistencyCalculateParameter parameter) {
String logicTableName = parameter.getLogicTableName();
String schemaName =
parameter.getTableNameSchemaNameMapping().getSchemaName(logicTableName);
String uniqueKey = parameter.getUniqueKey();
- String cacheKey = schemaName.toLowerCase() + "." +
logicTableName.toLowerCase();
+ String cacheKey =
DatabaseTypeFactory.getInstance(parameter.getDatabaseType()).isSchemaAvailable()
? schemaName.toLowerCase() + "." + logicTableName.toLowerCase() :
logicTableName;
Review Comment:
Looks `logicTableName` miss toLowerCase()
##########
shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/check/consistency/DataConsistencyChecker.java:
##########
@@ -82,8 +84,7 @@ public final class DataConsistencyChecker {
public DataConsistencyChecker(final MigrationJobConfiguration jobConfig,
final JobRateLimitAlgorithm readRateLimitAlgorithm) {
this.jobConfig = jobConfig;
logicTableNames =
Collections.singletonList(jobConfig.getTargetTableName());
- // TODO need get from actual data source.
- Map<String, List<String>> schemaTablesMap =
SchemaTableUtil.getSchemaTablesMap(jobConfig.getTargetDatabaseName(),
Collections.singleton(jobConfig.getTargetTableName()));
+ Map<String, List<String>> schemaTablesMap =
PipelineSchemaTableUtil.getSchemaTablesMapFromActual(jobConfig.getSource(),
jobConfig.getSourceTableName());
Review Comment:
Could we reuse `schemaTablesMap` in jobConfig?
##########
shardingsphere-test/shardingsphere-integration-test/shardingsphere-integration-test-scaling/src/test/java/org/apache/shardingsphere/integration/data/pipeline/cases/base/BaseITCase.java:
##########
@@ -358,6 +358,7 @@ protected void waitMigrationFinished(final String jobId) {
}
protected void assertGreaterThanInitTableInitRows(final int tableInitRows,
final String schema) {
+ proxyExecuteWithLog("REFRESH TABLE METADATA", 2);
String countSQL = StringUtils.isBlank(schema) ? "SELECT COUNT(*) as
count FROM t_order" : String.format("SELECT COUNT(*) as count FROM %s.t_order",
schema);
Review Comment:
Could we put `REFRESH TABLE METADATA` after `CHECK MIGRATION => STOP
MIGRATION => CLEAN MIGRATION`?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
