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/15 11:44:39 UTC
[shardingsphere] branch master updated: Refactor shardingSize as configurable (#17670)
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 fccbd23dbff Refactor shardingSize as configurable (#17670)
fccbd23dbff is described below
commit fccbd23dbff4ed89b695d6f8562e4b1e5abd2540
Author: Hongsheng Zhong <sa...@126.com>
AuthorDate: Sun May 15 19:44:33 2022 +0800
Refactor shardingSize as configurable (#17670)
---
.../src/test/resources/yaml/sharding-rule.yaml | 1 +
.../src/test/resources/yaml/sharding-scaling.yaml | 1 +
.../ShardingScalingRuleStatementConverter.java | 2 +-
.../ShardingScalingRulesQueryResultSetTest.java | 2 +-
...ateShardingScalingRuleStatementUpdaterTest.java | 2 +-
.../src/main/antlr4/imports/scaling/Keyword.g4 | 4 ++
.../main/antlr4/imports/scaling/RDLStatement.g4 | 6 ++-
.../parser/core/ScalingSQLStatementVisitor.java | 11 ++++-
.../statement/segment/InputOrOutputSegment.java | 9 ++++
.../OnRuleAlteredActionConfiguration.java | 2 +
.../YamlOnRuleAlteredActionConfiguration.java | 57 +++++++++++++++++-----
...nRuleAlteredActionConfigurationYamlSwapper.java | 4 +-
...eAlteredActionConfigurationYamlSwapperTest.java | 8 +--
.../rulealtered/RuleAlteredJobConfiguration.java | 5 +-
.../scenario/rulealtered/RuleAlteredContext.java | 4 ++
.../rulealtered/prepare/InventoryTaskSplitter.java | 8 +--
.../data/pipeline/cases/base/BaseITCase.java | 2 +-
.../prepare/InventoryTaskSplitterTest.java | 2 -
.../config_sharding_sphere_jdbc_source.yaml | 1 +
19 files changed, 94 insertions(+), 37 deletions(-)
diff --git a/shardingsphere-features/shardingsphere-sharding/shardingsphere-sharding-core/src/test/resources/yaml/sharding-rule.yaml b/shardingsphere-features/shardingsphere-sharding/shardingsphere-sharding-core/src/test/resources/yaml/sharding-rule.yaml
index 6ea46f2de27..c35a52214e9 100644
--- a/shardingsphere-features/shardingsphere-sharding/shardingsphere-sharding-core/src/test/resources/yaml/sharding-rule.yaml
+++ b/shardingsphere-features/shardingsphere-sharding/shardingsphere-sharding-core/src/test/resources/yaml/sharding-rule.yaml
@@ -97,6 +97,7 @@ rules:
input:
workerThread: 40
batchSize: 1000
+ shardingSize: 10000000
output:
workerThread: 40
batchSize: 1000
diff --git a/shardingsphere-features/shardingsphere-sharding/shardingsphere-sharding-core/src/test/resources/yaml/sharding-scaling.yaml b/shardingsphere-features/shardingsphere-sharding/shardingsphere-sharding-core/src/test/resources/yaml/sharding-scaling.yaml
index 8cb794a0eca..dd628c3a77a 100644
--- a/shardingsphere-features/shardingsphere-sharding/shardingsphere-sharding-core/src/test/resources/yaml/sharding-scaling.yaml
+++ b/shardingsphere-features/shardingsphere-sharding/shardingsphere-sharding-core/src/test/resources/yaml/sharding-scaling.yaml
@@ -21,6 +21,7 @@ scaling:
input:
workerThread: 40
batchSize: 1000
+ shardingSize: 10000000
output:
workerThread: 40
batchSize: 1000
diff --git a/shardingsphere-features/shardingsphere-sharding/shardingsphere-sharding-distsql/shardingsphere-sharding-distsql-handler/src/main/java/org/apache/shardingsphere/scaling/distsql/handler/converter/ShardingScalingRuleStatementConverter.java b/shardingsphere-features/shardingsphere-sharding/shardingsphere-sharding-distsql/shardingsphere-sharding-distsql-handler/src/main/java/org/apache/shardingsphere/scaling/distsql/handler/converter/ShardingScalingRuleStatementConverter.java
index 2294572c42e..4f420ea05b2 100644
--- a/shardingsphere-features/shardingsphere-sharding/shardingsphere-sharding-distsql/shardingsphere-sharding-distsql-handler/src/main/java/org/apache/shardingsphere/scaling/distsql/handler/converter/ShardingScalingRuleStatementConverter.java
+++ b/shardingsphere-features/shardingsphere-sharding/shardingsphere-sharding-distsql/shardingsphere-sharding-distsql-handler/src/main/java/org/apache/shardingsphere/scaling/distsql/handler/converter/ShardingScalingRuleStatementConverter.java
@@ -52,7 +52,7 @@ public final class ShardingScalingRuleStatementConverter {
if (null == inputSegment) {
return null;
}
- return new InputConfiguration(inputSegment.getWorkerThread(), inputSegment.getBatchSize(), convertToAlgorithm(inputSegment.getRateLimiter()));
+ return new InputConfiguration(inputSegment.getWorkerThread(), inputSegment.getBatchSize(), inputSegment.getShardingSize(), convertToAlgorithm(inputSegment.getRateLimiter()));
}
private static OutputConfiguration convertToOutputConfiguration(final InputOrOutputSegment outputSegment) {
diff --git a/shardingsphere-features/shardingsphere-sharding/shardingsphere-sharding-distsql/shardingsphere-sharding-distsql-handler/src/test/java/org/apache/shardingsphere/scaling/distsql/handler/query/ShardingScalingRulesQueryResultSetTest.java b/shardingsphere-features/shardingsphere-sharding/shardingsphere-sharding-distsql/shardingsphere-sharding-distsql-handler/src/test/java/org/apache/shardingsphere/scaling/distsql/handler/query/ShardingScalingRulesQueryResultSetTest.java
index b6dcfaa81e0..64e71059592 100644
--- a/shardingsphere-features/shardingsphere-sharding/shardingsphere-sharding-distsql/shardingsphere-sharding-distsql-handler/src/test/java/org/apache/shardingsphere/scaling/distsql/handler/query/ShardingScalingRulesQueryResultSetTest.java
+++ b/shardingsphere-features/shardingsphere-sharding/shardingsphere-sharding-distsql/shardingsphere-sharding-distsql-handler/src/test/java/org/apache/shardingsphere/scaling/distsql/handler/query/ShardingScalingRulesQueryResultSetTest.java
@@ -81,7 +81,7 @@ public final class ShardingScalingRulesQueryResultSetTest {
}
private InputConfiguration createInputConfiguration(final String type, final Properties props) {
- return new InputConfiguration(10, 100, createAlgorithm(type, props));
+ return new InputConfiguration(10, 100, 10, createAlgorithm(type, props));
}
private OutputConfiguration createOutputConfiguration(final String type, final Properties props) {
diff --git a/shardingsphere-features/shardingsphere-sharding/shardingsphere-sharding-distsql/shardingsphere-sharding-distsql-handler/src/test/java/org/apache/shardingsphere/scaling/distsql/handler/update/CreateShardingScalingRuleStatementUpdaterTest.java b/shardingsphere-features/shardingsphere-sharding/shardingsphere-sharding-distsql/shardingsphere-sharding-distsql-handler/src/test/java/org/apache/shardingsphere/scaling/distsql/handler/update/CreateShardingScalingRuleStatementUpdaterTest.java
index c5f18782c07..8ae028652bb 100644
--- a/shardingsphere-features/shardingsphere-sharding/shardingsphere-sharding-distsql/shardingsphere-sharding-distsql-handler/src/test/java/org/apache/shardingsphere/scaling/distsql/handler/update/CreateShardingScalingRuleStatementUpdaterTest.java
+++ b/shardingsphere-features/shardingsphere-sharding/shardingsphere-sharding-distsql/shardingsphere-sharding-distsql-handler/src/test/java/org/apache/shardingsphere/scaling/distsql/handler/update/CreateShardingScalingRuleStatementUpdaterTest.java
@@ -152,7 +152,7 @@ public final class CreateShardingScalingRuleStatementUpdaterTest {
}
private InputOrOutputSegment createInputOrOutputSegment(final String type) {
- return new InputOrOutputSegment(10, 1000, new AlgorithmSegment(type, new Properties()));
+ return new InputOrOutputSegment(10, 1000, 10, new AlgorithmSegment(type, new Properties()));
}
private ShardingScalingRuleConfigurationSegment createConfigurationWithInvalidStreamChannel() {
diff --git a/shardingsphere-features/shardingsphere-sharding/shardingsphere-sharding-distsql/shardingsphere-sharding-distsql-parser/src/main/antlr4/imports/scaling/Keyword.g4 b/shardingsphere-features/shardingsphere-sharding/shardingsphere-sharding-distsql/shardingsphere-sharding-distsql-parser/src/main/antlr4/imports/scaling/Keyword.g4
index d99916176e1..0d0580e236c 100644
--- a/shardingsphere-features/shardingsphere-sharding/shardingsphere-sharding-distsql/shardingsphere-sharding-distsql-parser/src/main/antlr4/imports/scaling/Keyword.g4
+++ b/shardingsphere-features/shardingsphere-sharding/shardingsphere-sharding-distsql/shardingsphere-sharding-distsql-parser/src/main/antlr4/imports/scaling/Keyword.g4
@@ -143,6 +143,10 @@ BATCH_SIZE
: B A T C H UL_ S I Z E
;
+SHARDING_SIZE
+ : S H A R D I N G UL_ S I Z E
+ ;
+
RATE_LIMITER
: R A T E UL_ L I M I T E R
;
diff --git a/shardingsphere-features/shardingsphere-sharding/shardingsphere-sharding-distsql/shardingsphere-sharding-distsql-parser/src/main/antlr4/imports/scaling/RDLStatement.g4 b/shardingsphere-features/shardingsphere-sharding/shardingsphere-sharding-distsql/shardingsphere-sharding-distsql-parser/src/main/antlr4/imports/scaling/RDLStatement.g4
index 50be7b488f7..a67d038dede 100644
--- a/shardingsphere-features/shardingsphere-sharding/shardingsphere-sharding-distsql/shardingsphere-sharding-distsql-parser/src/main/antlr4/imports/scaling/RDLStatement.g4
+++ b/shardingsphere-features/shardingsphere-sharding/shardingsphere-sharding-distsql/shardingsphere-sharding-distsql-parser/src/main/antlr4/imports/scaling/RDLStatement.g4
@@ -44,7 +44,7 @@ scalingRuleDefinition
;
inputDefinition
- : INPUT LP workerThread? (COMMA? batchSize)? (COMMA? rateLimiter)? RP
+ : INPUT LP workerThread? (COMMA? batchSize)? (COMMA? shardingSize)? (COMMA? rateLimiter)? RP
;
outputDefinition
@@ -67,6 +67,10 @@ batchSize
: BATCH_SIZE EQ intValue
;
+shardingSize
+ : SHARDING_SIZE EQ intValue
+ ;
+
rateLimiter
: RATE_LIMITER LP algorithmDefinition RP
;
diff --git a/shardingsphere-features/shardingsphere-sharding/shardingsphere-sharding-distsql/shardingsphere-sharding-distsql-parser/src/main/java/org/apache/shardingsphere/scaling/distsql/parser/core/ScalingSQLStatementVisitor.java b/shardingsphere-features/shardingsphere-sharding/shardingsphere-sharding-distsql/shardingsphere-sharding-distsql-parser/src/main/java/org/apache/shardingsphere/scaling/distsql/parser/core/ScalingSQLStatementVisitor.java
index 2158bf72748..8b8197dd73f 100644
--- a/shardingsphere-features/shardingsphere-sharding/shardingsphere-sharding-distsql/shardingsphere-sharding-distsql-parser/src/main/java/org/apache/shardingsphere/scaling/distsql/parser/core/ScalingSQLStatementVisitor.java
+++ b/shardingsphere-features/shardingsphere-sharding/shardingsphere-sharding-distsql/shardingsphere-sharding-distsql-parser/src/main/java/org/apache/shardingsphere/scaling/distsql/parser/core/ScalingSQLStatementVisitor.java
@@ -38,6 +38,7 @@ import org.apache.shardingsphere.distsql.parser.autogen.ScalingStatementParser.R
import org.apache.shardingsphere.distsql.parser.autogen.ScalingStatementParser.RestoreScalingSourceWritingContext;
import org.apache.shardingsphere.distsql.parser.autogen.ScalingStatementParser.ScalingRuleDefinitionContext;
import org.apache.shardingsphere.distsql.parser.autogen.ScalingStatementParser.SchemaNameContext;
+import org.apache.shardingsphere.distsql.parser.autogen.ScalingStatementParser.ShardingSizeContext;
import org.apache.shardingsphere.distsql.parser.autogen.ScalingStatementParser.ShowScalingCheckAlgorithmsContext;
import org.apache.shardingsphere.distsql.parser.autogen.ScalingStatementParser.ShowScalingListContext;
import org.apache.shardingsphere.distsql.parser.autogen.ScalingStatementParser.ShowScalingStatusContext;
@@ -168,11 +169,12 @@ public final class ScalingSQLStatementVisitor extends ScalingStatementBaseVisito
public ASTNode visitInputDefinition(final InputDefinitionContext ctx) {
Integer workerThread = getWorkerThread(ctx.workerThread());
Integer batchSize = getBatchSize(ctx.batchSize());
+ Integer shardingSize = getShardingSize(ctx.shardingSize());
AlgorithmSegment rateLimiter = null;
if (null != ctx.rateLimiter()) {
rateLimiter = (AlgorithmSegment) visit(ctx.rateLimiter());
}
- return new InputOrOutputSegment(workerThread, batchSize, rateLimiter);
+ return new InputOrOutputSegment(workerThread, batchSize, shardingSize, rateLimiter);
}
@Override
@@ -200,6 +202,13 @@ public final class ScalingSQLStatementVisitor extends ScalingStatementBaseVisito
return Integer.parseInt(ctx.intValue().getText());
}
+ private Integer getShardingSize(final ShardingSizeContext ctx) {
+ if (null == ctx) {
+ return null;
+ }
+ return Integer.parseInt(ctx.intValue().getText());
+ }
+
@Override
public ASTNode visitRateLimiter(final RateLimiterContext ctx) {
return visit(ctx.algorithmDefinition());
diff --git a/shardingsphere-features/shardingsphere-sharding/shardingsphere-sharding-distsql/shardingsphere-sharding-distsql-statement/src/main/java/org/apache/shardingsphere/scaling/distsql/statement/segment/InputOrOutputSegment.java b/shardingsphere-features/shardingsphere-sharding/shardingsphere-sharding-distsql/shardingsphere-sharding-distsql-statement/src/main/java/org/apache/shardingsphere/scaling/distsql/statement/segment/InputOrOutputSegment.java
index 931afa58516..9007e8030d1 100644
--- a/shardingsphere-features/shardingsphere-sharding/shardingsphere-sharding-distsql/shardingsphere-sharding-distsql-statement/src/main/java/org/apache/shardingsphere/scaling/distsql/statement/segment/InputOrOutputSegment.java
+++ b/shardingsphere-features/shardingsphere-sharding/shardingsphere-sharding-distsql/shardingsphere-sharding-distsql-statement/src/main/java/org/apache/shardingsphere/scaling/distsql/statement/segment/InputOrOutputSegment.java
@@ -33,5 +33,14 @@ public final class InputOrOutputSegment implements ASTNode {
private final Integer batchSize;
+ private final Integer shardingSize;
+
private final AlgorithmSegment rateLimiter;
+
+ public InputOrOutputSegment(final Integer workerThread, final Integer batchSize, final AlgorithmSegment rateLimiter) {
+ this.workerThread = workerThread;
+ this.batchSize = batchSize;
+ this.shardingSize = 1000_0000;
+ this.rateLimiter = rateLimiter;
+ }
}
diff --git a/shardingsphere-infra/shardingsphere-infra-common/src/main/java/org/apache/shardingsphere/infra/config/rulealtered/OnRuleAlteredActionConfiguration.java b/shardingsphere-infra/shardingsphere-infra-common/src/main/java/org/apache/shardingsphere/infra/config/rulealtered/OnRuleAlteredActionConfiguration.java
index 87b88d38d48..604045d3822 100644
--- a/shardingsphere-infra/shardingsphere-infra-common/src/main/java/org/apache/shardingsphere/infra/config/rulealtered/OnRuleAlteredActionConfiguration.java
+++ b/shardingsphere-infra/shardingsphere-infra-common/src/main/java/org/apache/shardingsphere/infra/config/rulealtered/OnRuleAlteredActionConfiguration.java
@@ -49,6 +49,8 @@ public final class OnRuleAlteredActionConfiguration {
private final Integer batchSize;
+ private final Integer shardingSize;
+
private final ShardingSphereAlgorithmConfiguration rateLimiter;
}
diff --git a/shardingsphere-infra/shardingsphere-infra-common/src/main/java/org/apache/shardingsphere/infra/yaml/config/pojo/rulealtered/YamlOnRuleAlteredActionConfiguration.java b/shardingsphere-infra/shardingsphere-infra-common/src/main/java/org/apache/shardingsphere/infra/yaml/config/pojo/rulealtered/YamlOnRuleAlteredActionConfiguration.java
index 73a191b2e7a..5a1caa7dbef 100644
--- a/shardingsphere-infra/shardingsphere-infra-common/src/main/java/org/apache/shardingsphere/infra/yaml/config/pojo/rulealtered/YamlOnRuleAlteredActionConfiguration.java
+++ b/shardingsphere-infra/shardingsphere-infra-common/src/main/java/org/apache/shardingsphere/infra/yaml/config/pojo/rulealtered/YamlOnRuleAlteredActionConfiguration.java
@@ -45,9 +45,17 @@ public final class YamlOnRuleAlteredActionConfiguration implements YamlConfigura
@Data
public static final class YamlInputConfiguration implements YamlConfiguration {
- private Integer workerThread;
+ private static final Integer DEFAULT_WORKER_THREAD = 40;
- private Integer batchSize;
+ private static final Integer DEFAULT_BATCH_SIZE = 1000;
+
+ private static final Integer DEFAULT_SHARDING_SIZE = 1000_0000;
+
+ private Integer workerThread = DEFAULT_WORKER_THREAD;
+
+ private Integer batchSize = DEFAULT_BATCH_SIZE;
+
+ private Integer shardingSize = DEFAULT_SHARDING_SIZE;
private YamlShardingSphereAlgorithmConfiguration rateLimiter;
@@ -57,19 +65,35 @@ public final class YamlOnRuleAlteredActionConfiguration implements YamlConfigura
* @return input configuration
*/
public static YamlInputConfiguration buildWithDefaultValue() {
- YamlInputConfiguration result = new YamlInputConfiguration();
- result.setWorkerThread(40);
- result.setBatchSize(1000);
- return result;
+ return new YamlInputConfiguration();
+ }
+
+ /**
+ * Fill in null fields with default value.
+ */
+ public void fillInNullFieldsWithDefaultValue() {
+ if (null == workerThread) {
+ workerThread = DEFAULT_WORKER_THREAD;
+ }
+ if (null == batchSize) {
+ batchSize = DEFAULT_BATCH_SIZE;
+ }
+ if (null == shardingSize) {
+ shardingSize = DEFAULT_SHARDING_SIZE;
+ }
}
}
@Data
public static final class YamlOutputConfiguration implements YamlConfiguration {
- private Integer workerThread;
+ private static final Integer DEFAULT_WORKER_THREAD = 40;
+
+ private static final Integer DEFAULT_BATCH_SIZE = 1000;
- private Integer batchSize;
+ private Integer workerThread = DEFAULT_WORKER_THREAD;
+
+ private Integer batchSize = DEFAULT_BATCH_SIZE;
private YamlShardingSphereAlgorithmConfiguration rateLimiter;
@@ -79,10 +103,19 @@ public final class YamlOnRuleAlteredActionConfiguration implements YamlConfigura
* @return output configuration
*/
public static YamlOutputConfiguration buildWithDefaultValue() {
- YamlOutputConfiguration result = new YamlOutputConfiguration();
- result.setWorkerThread(40);
- result.setBatchSize(1000);
- return result;
+ return new YamlOutputConfiguration();
+ }
+
+ /**
+ * Fill in null fields with default value.
+ */
+ public void fillInNullFieldsWithDefaultValue() {
+ if (null == workerThread) {
+ workerThread = DEFAULT_WORKER_THREAD;
+ }
+ if (null == batchSize) {
+ batchSize = DEFAULT_BATCH_SIZE;
+ }
}
}
}
diff --git a/shardingsphere-infra/shardingsphere-infra-common/src/main/java/org/apache/shardingsphere/infra/yaml/config/swapper/rulealtered/OnRuleAlteredActionConfigurationYamlSwapper.java b/shardingsphere-infra/shardingsphere-infra-common/src/main/java/org/apache/shardingsphere/infra/yaml/config/swapper/rulealtered/OnRuleAlteredActionConfigurationYamlSwapper.java
index 665e48df801..5774cebdac3 100644
--- a/shardingsphere-infra/shardingsphere-infra-common/src/main/java/org/apache/shardingsphere/infra/yaml/config/swapper/rulealtered/OnRuleAlteredActionConfigurationYamlSwapper.java
+++ b/shardingsphere-infra/shardingsphere-infra-common/src/main/java/org/apache/shardingsphere/infra/yaml/config/swapper/rulealtered/OnRuleAlteredActionConfigurationYamlSwapper.java
@@ -74,6 +74,7 @@ public final class OnRuleAlteredActionConfigurationYamlSwapper implements YamlCo
YamlInputConfiguration result = new YamlInputConfiguration();
result.setWorkerThread(data.getWorkerThread());
result.setBatchSize(data.getBatchSize());
+ result.setShardingSize(data.getShardingSize());
result.setRateLimiter(ALGORITHM_CONFIG_YAML_SWAPPER.swapToYamlConfiguration(data.getRateLimiter()));
return result;
}
@@ -83,7 +84,8 @@ public final class OnRuleAlteredActionConfigurationYamlSwapper implements YamlCo
if (null == yamlConfig) {
return null;
}
- return new InputConfiguration(yamlConfig.getWorkerThread(), yamlConfig.getBatchSize(), ALGORITHM_CONFIG_YAML_SWAPPER.swapToObject(yamlConfig.getRateLimiter()));
+ return new InputConfiguration(yamlConfig.getWorkerThread(), yamlConfig.getBatchSize(), yamlConfig.getShardingSize(),
+ ALGORITHM_CONFIG_YAML_SWAPPER.swapToObject(yamlConfig.getRateLimiter()));
}
}
diff --git a/shardingsphere-infra/shardingsphere-infra-common/src/test/java/org/apache/shardingsphere/infra/yaml/config/swapper/rulealtered/OnRuleAlteredActionConfigurationYamlSwapperTest.java b/shardingsphere-infra/shardingsphere-infra-common/src/test/java/org/apache/shardingsphere/infra/yaml/config/swapper/rulealtered/OnRuleAlteredActionConfigurationYamlSwapperTest.java
index e5f359d1c65..10af993a1f0 100644
--- a/shardingsphere-infra/shardingsphere-infra-common/src/test/java/org/apache/shardingsphere/infra/yaml/config/swapper/rulealtered/OnRuleAlteredActionConfigurationYamlSwapperTest.java
+++ b/shardingsphere-infra/shardingsphere-infra-common/src/test/java/org/apache/shardingsphere/infra/yaml/config/swapper/rulealtered/OnRuleAlteredActionConfigurationYamlSwapperTest.java
@@ -41,14 +41,10 @@ public final class OnRuleAlteredActionConfigurationYamlSwapperTest {
Properties rateLimiterProps = new Properties();
rateLimiterProps.setProperty("batch-size", "1000");
rateLimiterProps.setProperty("qps", "50");
- YamlInputConfiguration yamlInputConfig = new YamlInputConfiguration();
+ YamlInputConfiguration yamlInputConfig = YamlInputConfiguration.buildWithDefaultValue();
yamlConfig.setInput(yamlInputConfig);
- yamlInputConfig.setWorkerThread(40);
- yamlInputConfig.setBatchSize(1000);
yamlInputConfig.setRateLimiter(new YamlShardingSphereAlgorithmConfiguration("INPUT", rateLimiterProps));
- YamlOutputConfiguration yamlOutputConfig = new YamlOutputConfiguration();
- yamlOutputConfig.setWorkerThread(40);
- yamlOutputConfig.setBatchSize(1000);
+ YamlOutputConfiguration yamlOutputConfig = YamlOutputConfiguration.buildWithDefaultValue();
yamlOutputConfig.setRateLimiter(new YamlShardingSphereAlgorithmConfiguration("OUTPUT", rateLimiterProps));
yamlConfig.setOutput(yamlOutputConfig);
Properties streamChannelProps = new Properties();
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/config/rulealtered/RuleAlteredJobConfiguration.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/config/rulealtered/RuleAlteredJobConfiguration.java
index 242dcc22782..9565b9539c8 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/config/rulealtered/RuleAlteredJobConfiguration.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/config/rulealtered/RuleAlteredJobConfiguration.java
@@ -84,9 +84,6 @@ public final class RuleAlteredJobConfiguration implements PipelineJobConfigurati
private String logicTables;
- // TODO shardingSize should be configurable
- private int shardingSize = 1000 * 10000;
-
private String sourceDatabaseType;
private String targetDatabaseType;
@@ -171,7 +168,7 @@ public final class RuleAlteredJobConfiguration implements PipelineJobConfigurati
public String toString() {
return "RuleAlteredJobConfiguration{"
+ "jobId='" + jobId + '\'' + ", databaseName='" + databaseName + '\''
- + ", activeVersion=" + activeVersion + ", newVersion=" + newVersion + ", shardingSize=" + shardingSize
+ + ", activeVersion=" + activeVersion + ", newVersion=" + newVersion
+ ", sourceDatabaseType='" + sourceDatabaseType + '\'' + ", targetDatabaseType='" + targetDatabaseType + '\''
+ '}';
}
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/RuleAlteredContext.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/RuleAlteredContext.java
index 3d0761a9365..0888dd9cd7e 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/RuleAlteredContext.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/RuleAlteredContext.java
@@ -113,9 +113,13 @@ public final class RuleAlteredContext {
YamlOnRuleAlteredActionConfiguration yamlActionConfig = SWAPPER.swapToYamlConfiguration(actionConfig);
if (null == yamlActionConfig.getInput()) {
yamlActionConfig.setInput(YamlInputConfiguration.buildWithDefaultValue());
+ } else {
+ yamlActionConfig.getInput().fillInNullFieldsWithDefaultValue();
}
if (null == yamlActionConfig.getOutput()) {
yamlActionConfig.setOutput(YamlOutputConfiguration.buildWithDefaultValue());
+ } else {
+ yamlActionConfig.getOutput().fillInNullFieldsWithDefaultValue();
}
if (null == yamlActionConfig.getStreamChannel()) {
yamlActionConfig.setStreamChannel(new YamlShardingSphereAlgorithmConfiguration(MemoryPipelineChannelCreator.TYPE, new Properties()));
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 4b2972c4fce..593e3b47979 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
@@ -44,8 +44,6 @@ import org.apache.shardingsphere.data.pipeline.scenario.rulealtered.RuleAlteredJ
import org.apache.shardingsphere.data.pipeline.spi.ingest.channel.PipelineChannelCreator;
import org.apache.shardingsphere.data.pipeline.spi.ratelimit.JobRateLimitAlgorithm;
import org.apache.shardingsphere.infra.config.rulealtered.OnRuleAlteredActionConfiguration.InputConfiguration;
-import org.apache.shardingsphere.infra.yaml.config.pojo.rulealtered.YamlOnRuleAlteredActionConfiguration.YamlInputConfiguration;
-import org.apache.shardingsphere.infra.yaml.config.swapper.rulealtered.OnRuleAlteredActionConfigurationYamlSwapper.InputConfigurationSwapper;
import javax.sql.DataSource;
import java.sql.Connection;
@@ -110,9 +108,6 @@ public final class InventoryTaskSplitter {
Collection<InventoryDumperConfiguration> result = new LinkedList<>();
RuleAlteredContext ruleAlteredContext = jobContext.getRuleAlteredContext();
InputConfiguration inputConfig = ruleAlteredContext.getOnRuleAlteredActionConfig().getInput();
- if (null == inputConfig) {
- inputConfig = new InputConfigurationSwapper().swapToObject(YamlInputConfiguration.buildWithDefaultValue());
- }
int batchSize = inputConfig.getBatchSize();
JobRateLimitAlgorithm rateLimitAlgorithm = ruleAlteredContext.getInputRateLimitAlgorithm();
Collection<IngestPosition<?>> inventoryPositions = getInventoryPositions(jobContext, dumperConfig, dataSource, metaDataLoader);
@@ -183,6 +178,7 @@ public final class InventoryTaskSplitter {
RuleAlteredJobConfiguration jobConfig = jobContext.getJobConfig();
String sql = PipelineSQLBuilderFactory.getInstance(jobConfig.getSourceDatabaseType())
.buildSplitByPrimaryKeyRangeSQL(dumperConfig.getSchemaName(new LogicTableName(dumperConfig.getLogicTableName())), dumperConfig.getActualTableName(), dumperConfig.getPrimaryKey());
+ int shardingSize = jobContext.getRuleAlteredContext().getOnRuleAlteredActionConfig().getInput().getShardingSize();
try (
Connection connection = dataSource.getConnection();
PreparedStatement ps = connection.prepareStatement(sql)) {
@@ -190,7 +186,7 @@ public final class InventoryTaskSplitter {
long beginId = 0;
for (int i = 0; i < Integer.MAX_VALUE; i++) {
ps.setLong(1, beginId);
- ps.setLong(2, jobConfig.getShardingSize());
+ ps.setLong(2, shardingSize);
try (ResultSet rs = ps.executeQuery()) {
if (!rs.next()) {
log.info("getPositionByPrimaryKeyRange, rs.next false, break");
diff --git a/shardingsphere-test/shardingsphere-integration-test/shardingsphere-integration-test-scaling/src/test/java/org/apache/shardingsphere/integration/data/pipeline/cases/base/BaseITCase.java b/shardingsphere-test/shardingsphere-integration-test/shardingsphere-integration-test-scaling/src/test/java/org/apache/shardingsphere/integration/data/pipeline/cases/base/BaseITCase.java
index 2a12c7e1da3..4d5b3995788 100644
--- a/shardingsphere-test/shardingsphere-integration-test/shardingsphere-integration-test-scaling/src/test/java/org/apache/shardingsphere/integration/data/pipeline/cases/base/BaseITCase.java
+++ b/shardingsphere-test/shardingsphere-integration-test/shardingsphere-integration-test-scaling/src/test/java/org/apache/shardingsphere/integration/data/pipeline/cases/base/BaseITCase.java
@@ -162,7 +162,7 @@ public abstract class BaseITCase {
}
protected void createScalingRule() {
- jdbcTemplate.execute("CREATE SHARDING SCALING RULE scaling_manual (DATA_CONSISTENCY_CHECKER(TYPE(NAME=DATA_MATCH)))");
+ jdbcTemplate.execute("CREATE SHARDING SCALING RULE scaling_manual (INPUT(SHARDING_SIZE=1000), DATA_CONSISTENCY_CHECKER(TYPE(NAME=DATA_MATCH)))");
}
protected void createSchema(final String schemaName) {
diff --git a/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/prepare/InventoryTaskSplitterTest.java b/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/prepare/InventoryTaskSplitterTest.java
index 213402e8531..bce1657359f 100644
--- a/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/prepare/InventoryTaskSplitterTest.java
+++ b/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/prepare/InventoryTaskSplitterTest.java
@@ -74,7 +74,6 @@ public final class InventoryTaskSplitterTest {
@Test
public void assertSplitInventoryDataWithEmptyTable() throws SQLException {
- taskConfig.getJobConfig().setShardingSize(10);
initEmptyTablePrimaryEnvironment(taskConfig.getDumperConfig());
List<InventoryTask> actual = inventoryTaskSplitter.splitInventoryData(jobContext);
assertThat(actual.size(), is(1));
@@ -84,7 +83,6 @@ public final class InventoryTaskSplitterTest {
@Test
public void assertSplitInventoryDataWithIntPrimary() throws SQLException {
- taskConfig.getJobConfig().setShardingSize(10);
initIntPrimaryEnvironment(taskConfig.getDumperConfig());
List<InventoryTask> actual = inventoryTaskSplitter.splitInventoryData(jobContext);
assertThat(actual.size(), is(10));
diff --git a/shardingsphere-test/shardingsphere-pipeline-test/src/test/resources/config_sharding_sphere_jdbc_source.yaml b/shardingsphere-test/shardingsphere-pipeline-test/src/test/resources/config_sharding_sphere_jdbc_source.yaml
index 4fd9e105258..116ab7bcbd9 100644
--- a/shardingsphere-test/shardingsphere-pipeline-test/src/test/resources/config_sharding_sphere_jdbc_source.yaml
+++ b/shardingsphere-test/shardingsphere-pipeline-test/src/test/resources/config_sharding_sphere_jdbc_source.yaml
@@ -55,6 +55,7 @@ rules:
input:
workerThread: 40
batchSize: 1000
+ shardingSize: 10
output:
workerThread: 40
batchSize: 1000