This is an automated email from the ASF dual-hosted git repository.
terrymanu 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 223b6ceaca4 Refactor PipelineRequiredColumnsExtractor to use rule
configurations (#39843)
223b6ceaca4 is described below
commit 223b6ceaca428b4b473171b5329f0540b7bc1d7d
Author: Liang Zhang <[email protected]>
AuthorDate: Mon Sep 14 16:40:30 2026 +0800
Refactor PipelineRequiredColumnsExtractor to use rule configurations
(#39843)
---
.../importer/PipelineRequiredColumnsExtractor.java | 10 +--
.../sharding/PipelineShardingColumnsExtractor.java | 11 ++-
.../PipelineShardingColumnsExtractorTest.java | 78 ++++++++++------------
.../shardingsphere/data/pipeline/cdc/CDCJob.java | 7 +-
.../data/pipeline/cdc/CDCJobTest.java | 24 ++++++-
.../migration/MigrationJobExecutorCallback.java | 8 ++-
.../MigrationJobExecutorCallbackTest.java | 48 +++++++++++++
.../pipeline/core/util/PipelineContextUtils.java | 6 +-
8 files changed, 126 insertions(+), 66 deletions(-)
diff --git
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/importer/PipelineRequiredColumnsExtractor.java
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/importer/PipelineRequiredColumnsExtractor.java
index 3a36a04aa35..d04a3866d5f 100644
---
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/importer/PipelineRequiredColumnsExtractor.java
+++
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/importer/PipelineRequiredColumnsExtractor.java
@@ -17,10 +17,10 @@
package org.apache.shardingsphere.data.pipeline.core.importer;
+import org.apache.shardingsphere.infra.config.rule.RuleConfiguration;
import
org.apache.shardingsphere.infra.metadata.identifier.ShardingSphereIdentifier;
import org.apache.shardingsphere.infra.spi.annotation.SingletonSPI;
import org.apache.shardingsphere.infra.spi.type.ordered.OrderedSPI;
-import
org.apache.shardingsphere.infra.yaml.config.pojo.rule.YamlRuleConfiguration;
import java.util.Collection;
import java.util.Map;
@@ -28,17 +28,17 @@ import java.util.Map;
/**
* Pipeline required columns extractor.
*
- * @param <T> type of YAML rule configuration
+ * @param <T> type of rule configuration
*/
@SingletonSPI
-public interface PipelineRequiredColumnsExtractor<T extends
YamlRuleConfiguration> extends OrderedSPI<T> {
+public interface PipelineRequiredColumnsExtractor<T extends RuleConfiguration>
extends OrderedSPI<T> {
/**
* Get table and required columns map.
*
- * @param yamlRuleConfig YAML rule configuration
+ * @param ruleConfig rule configuration
* @param logicTableNames logic table names
* @return table and required columns map
*/
- Map<ShardingSphereIdentifier, Collection<String>>
getTableAndRequiredColumnsMap(T yamlRuleConfig,
Collection<ShardingSphereIdentifier> logicTableNames);
+ Map<ShardingSphereIdentifier, Collection<String>>
getTableAndRequiredColumnsMap(T ruleConfig,
Collection<ShardingSphereIdentifier> logicTableNames);
}
diff --git
a/kernel/data-pipeline/feature/sharding/src/main/java/org/apache/shardingsphere/data/pipeline/sharding/PipelineShardingColumnsExtractor.java
b/kernel/data-pipeline/feature/sharding/src/main/java/org/apache/shardingsphere/data/pipeline/sharding/PipelineShardingColumnsExtractor.java
index f647a08207c..42e847a2774 100644
---
a/kernel/data-pipeline/feature/sharding/src/main/java/org/apache/shardingsphere/data/pipeline/sharding/PipelineShardingColumnsExtractor.java
+++
b/kernel/data-pipeline/feature/sharding/src/main/java/org/apache/shardingsphere/data/pipeline/sharding/PipelineShardingColumnsExtractor.java
@@ -26,8 +26,6 @@ import
org.apache.shardingsphere.sharding.api.config.strategy.sharding.ComplexSh
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.constant.ShardingOrder;
-import
org.apache.shardingsphere.sharding.yaml.config.YamlShardingRuleConfiguration;
-import
org.apache.shardingsphere.sharding.yaml.swapper.YamlShardingRuleConfigurationSwapper;
import java.util.Arrays;
import java.util.Collection;
@@ -40,12 +38,11 @@ import java.util.concurrent.ConcurrentHashMap;
/**
* Pipeline sharding columns extractor.
*/
-public final class PipelineShardingColumnsExtractor implements
PipelineRequiredColumnsExtractor<YamlShardingRuleConfiguration> {
+public final class PipelineShardingColumnsExtractor implements
PipelineRequiredColumnsExtractor<ShardingRuleConfiguration> {
@Override
- public Map<ShardingSphereIdentifier, Collection<String>>
getTableAndRequiredColumnsMap(final YamlShardingRuleConfiguration
yamlRuleConfig,
+ public Map<ShardingSphereIdentifier, Collection<String>>
getTableAndRequiredColumnsMap(final ShardingRuleConfiguration ruleConfig,
final Collection<ShardingSphereIdentifier> logicTableNames) {
- ShardingRuleConfiguration ruleConfig = new
YamlShardingRuleConfigurationSwapper().swapToObject(yamlRuleConfig);
Set<String> defaultDatabaseShardingColumns =
extractShardingColumns(ruleConfig.getDefaultDatabaseShardingStrategy());
Set<String> defaultTableShardingColumns =
extractShardingColumns(ruleConfig.getDefaultTableShardingStrategy());
// TODO check is it need to be ConcurrentHashMap?
@@ -85,7 +82,7 @@ public final class PipelineShardingColumnsExtractor
implements PipelineRequiredC
}
@Override
- public Class<YamlShardingRuleConfiguration> getTypeClass() {
- return YamlShardingRuleConfiguration.class;
+ public Class<ShardingRuleConfiguration> getTypeClass() {
+ return ShardingRuleConfiguration.class;
}
}
diff --git
a/kernel/data-pipeline/feature/sharding/src/test/java/org/apache/shardingsphere/data/pipeline/sharding/PipelineShardingColumnsExtractorTest.java
b/kernel/data-pipeline/feature/sharding/src/test/java/org/apache/shardingsphere/data/pipeline/sharding/PipelineShardingColumnsExtractorTest.java
index d54f95e150e..79a7ad64665 100644
---
a/kernel/data-pipeline/feature/sharding/src/test/java/org/apache/shardingsphere/data/pipeline/sharding/PipelineShardingColumnsExtractorTest.java
+++
b/kernel/data-pipeline/feature/sharding/src/test/java/org/apache/shardingsphere/data/pipeline/sharding/PipelineShardingColumnsExtractorTest.java
@@ -20,12 +20,11 @@ package org.apache.shardingsphere.data.pipeline.sharding;
import
org.apache.shardingsphere.data.pipeline.core.importer.PipelineRequiredColumnsExtractor;
import
org.apache.shardingsphere.infra.metadata.identifier.ShardingSphereIdentifier;
import org.apache.shardingsphere.infra.spi.type.ordered.OrderedSPILoader;
-import
org.apache.shardingsphere.sharding.yaml.config.YamlShardingRuleConfiguration;
-import
org.apache.shardingsphere.sharding.yaml.config.rule.YamlShardingAutoTableRuleConfiguration;
-import
org.apache.shardingsphere.sharding.yaml.config.rule.YamlTableRuleConfiguration;
-import
org.apache.shardingsphere.sharding.yaml.config.strategy.sharding.YamlComplexShardingStrategyConfiguration;
-import
org.apache.shardingsphere.sharding.yaml.config.strategy.sharding.YamlShardingStrategyConfiguration;
-import
org.apache.shardingsphere.sharding.yaml.config.strategy.sharding.YamlStandardShardingStrategyConfiguration;
+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.StandardShardingStrategyConfiguration;
import org.junit.jupiter.api.Test;
import java.util.Arrays;
@@ -36,66 +35,59 @@ import java.util.Map;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.containsInAnyOrder;
import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.isA;
import static org.junit.jupiter.api.Assertions.assertTrue;
class PipelineShardingColumnsExtractorTest {
@SuppressWarnings("unchecked")
- private final
PipelineRequiredColumnsExtractor<YamlShardingRuleConfiguration> extractor =
OrderedSPILoader.getServicesByClass(
- PipelineRequiredColumnsExtractor.class,
Collections.singleton(YamlShardingRuleConfiguration.class)).get(YamlShardingRuleConfiguration.class);
+ private final PipelineRequiredColumnsExtractor<ShardingRuleConfiguration>
extractor = OrderedSPILoader.getServicesByClass(
+ PipelineRequiredColumnsExtractor.class,
Collections.singleton(ShardingRuleConfiguration.class)).get(ShardingRuleConfiguration.class);
+
+ @Test
+ void assertOrderedSPILoader() {
+ assertThat(extractor, isA(PipelineShardingColumnsExtractor.class));
+ }
@Test
void assertGetTableAndRequiredColumnsMap() {
- YamlShardingRuleConfiguration yamlConfig = new
YamlShardingRuleConfiguration();
-
yamlConfig.setDefaultDatabaseStrategy(createYAMLStandardStrategyConfiguration("default_db_col"));
-
yamlConfig.setDefaultTableStrategy(createYAMLStandardStrategyConfiguration("default_tbl_col"));
- yamlConfig.getTables().put("t_explicit",
getYamlExplicitTableRuleConfiguration());
- yamlConfig.getTables().put("t_default",
getYamlTableRuleConfiguration("t_default"));
- yamlConfig.getTables().put("t_ignored",
getYamlTableRuleConfiguration("t_ignored"));
- yamlConfig.getAutoTables().put("t_auto",
getYamlShardingAutoTableRuleConfiguration("t_auto"));
- yamlConfig.getAutoTables().put("t_ignored_auto",
getYamlShardingAutoTableRuleConfiguration("t_ignored_auto"));
+ ShardingRuleConfiguration ruleConfig = new ShardingRuleConfiguration();
+
ruleConfig.setDefaultDatabaseShardingStrategy(createStandardShardingStrategyConfiguration("default_db_col"));
+
ruleConfig.setDefaultTableShardingStrategy(createStandardShardingStrategyConfiguration("default_tbl_col"));
+ ruleConfig.getTables().add(getExplicitTableRuleConfiguration());
+ ruleConfig.getTables().add(getTableRuleConfiguration("t_default"));
+ ruleConfig.getTables().add(getTableRuleConfiguration("t_ignored"));
+
ruleConfig.getAutoTables().add(getShardingAutoTableRuleConfiguration("t_auto"));
+
ruleConfig.getAutoTables().add(getShardingAutoTableRuleConfiguration("t_ignored_auto"));
Collection<ShardingSphereIdentifier> logicTables = Arrays.asList(
- new ShardingSphereIdentifier("t_explicit"), new
ShardingSphereIdentifier("t_default"), new ShardingSphereIdentifier("t_auto"));
- Map<ShardingSphereIdentifier, Collection<String>> actual =
extractor.getTableAndRequiredColumnsMap(yamlConfig, logicTables);
+ new ShardingSphereIdentifier("T_EXPLICIT"), new
ShardingSphereIdentifier("t_default"), new ShardingSphereIdentifier("t_auto"));
+ Map<ShardingSphereIdentifier, Collection<String>> actual =
extractor.getTableAndRequiredColumnsMap(ruleConfig, logicTables);
assertThat(actual.size(), is(3));
assertThat(actual.get(new ShardingSphereIdentifier("t_explicit")),
containsInAnyOrder("user_id", "order_id", "item_id"));
assertThat(actual.get(new ShardingSphereIdentifier("t_default")),
containsInAnyOrder("default_db_col", "default_tbl_col"));
assertTrue(actual.get(new
ShardingSphereIdentifier("t_auto")).isEmpty());
}
- private YamlTableRuleConfiguration getYamlExplicitTableRuleConfiguration()
{
- YamlTableRuleConfiguration result = new YamlTableRuleConfiguration();
- result.setLogicTable("t_explicit");
-
result.setDatabaseStrategy(createYAMLComplexStrategyConfiguration("user_id,order_id"));
-
result.setTableStrategy(createYAMLComplexStrategyConfiguration("item_id"));
+ private ShardingTableRuleConfiguration getExplicitTableRuleConfiguration()
{
+ ShardingTableRuleConfiguration result = new
ShardingTableRuleConfiguration("t_explicit", "ds_0.t_explicit");
+
result.setDatabaseShardingStrategy(createComplexShardingStrategyConfiguration("user_id,order_id"));
+
result.setTableShardingStrategy(createComplexShardingStrategyConfiguration("item_id"));
return result;
}
- private YamlTableRuleConfiguration getYamlTableRuleConfiguration(final
String tableName) {
- YamlTableRuleConfiguration result = new YamlTableRuleConfiguration();
- result.setLogicTable(tableName);
- return result;
+ private ShardingTableRuleConfiguration getTableRuleConfiguration(final
String tableName) {
+ return new ShardingTableRuleConfiguration(tableName, "ds_0." +
tableName);
}
- private YamlShardingAutoTableRuleConfiguration
getYamlShardingAutoTableRuleConfiguration(final String tableName) {
- YamlShardingAutoTableRuleConfiguration result = new
YamlShardingAutoTableRuleConfiguration();
- result.setLogicTable(tableName);
- return result;
+ private ShardingAutoTableRuleConfiguration
getShardingAutoTableRuleConfiguration(final String tableName) {
+ return new ShardingAutoTableRuleConfiguration(tableName, "ds_0");
}
- private YamlShardingStrategyConfiguration
createYAMLStandardStrategyConfiguration(final String column) {
- YamlShardingStrategyConfiguration result = new
YamlShardingStrategyConfiguration();
- YamlStandardShardingStrategyConfiguration standard = new
YamlStandardShardingStrategyConfiguration();
- standard.setShardingColumn(column);
- result.setStandard(standard);
- return result;
+ private StandardShardingStrategyConfiguration
createStandardShardingStrategyConfiguration(final String column) {
+ return new StandardShardingStrategyConfiguration(column,
"foo_algorithm");
}
- private YamlShardingStrategyConfiguration
createYAMLComplexStrategyConfiguration(final String columns) {
- YamlShardingStrategyConfiguration result = new
YamlShardingStrategyConfiguration();
- YamlComplexShardingStrategyConfiguration complex = new
YamlComplexShardingStrategyConfiguration();
- complex.setShardingColumns(columns);
- result.setComplex(complex);
- return result;
+ private ComplexShardingStrategyConfiguration
createComplexShardingStrategyConfiguration(final String columns) {
+ return new ComplexShardingStrategyConfiguration(columns,
"foo_algorithm");
}
}
diff --git
a/kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/CDCJob.java
b/kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/CDCJob.java
index d5624f749b8..d7a54de4d30 100644
---
a/kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/CDCJob.java
+++
b/kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/CDCJob.java
@@ -64,10 +64,11 @@ import
org.apache.shardingsphere.data.pipeline.core.ratelimit.JobRateLimitAlgori
import
org.apache.shardingsphere.data.pipeline.core.registrycenter.repository.PipelineGovernanceFacade;
import org.apache.shardingsphere.data.pipeline.core.task.PipelineTask;
import org.apache.shardingsphere.elasticjob.api.ShardingContext;
+import org.apache.shardingsphere.infra.config.rule.RuleConfiguration;
import
org.apache.shardingsphere.infra.metadata.identifier.ShardingSphereIdentifier;
import org.apache.shardingsphere.infra.spi.type.ordered.OrderedSPILoader;
import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
-import
org.apache.shardingsphere.infra.yaml.config.pojo.rule.YamlRuleConfiguration;
+import
org.apache.shardingsphere.infra.yaml.config.swapper.rule.YamlRuleConfigurationSwapperEngine;
import java.util.Collection;
import java.util.HashMap;
@@ -163,9 +164,9 @@ public final class CDCJob implements PipelineJob {
@SuppressWarnings({"rawtypes", "unchecked"})
private Map<ShardingSphereIdentifier, Collection<String>>
getTableAndRequiredColumnsMap(final CDCJobConfiguration jobConfig) {
Map<ShardingSphereIdentifier, Collection<String>> result = new
HashMap<>();
- Collection<YamlRuleConfiguration> yamlRuleConfigs =
jobConfig.getDataSourceConfig().getRootConfig().getRules();
+ Collection<RuleConfiguration> ruleConfigs = new
YamlRuleConfigurationSwapperEngine().swapToRuleConfigurations(jobConfig.getDataSourceConfig().getRootConfig().getRules());
Collection<ShardingSphereIdentifier> targetTableNames =
jobConfig.getSchemaTableNames().stream().map(ShardingSphereIdentifier::new).collect(Collectors.toSet());
- for (Entry<YamlRuleConfiguration, PipelineRequiredColumnsExtractor>
entry : OrderedSPILoader.getServices(PipelineRequiredColumnsExtractor.class,
yamlRuleConfigs).entrySet()) {
+ for (Entry<RuleConfiguration, PipelineRequiredColumnsExtractor> entry
: OrderedSPILoader.getServices(PipelineRequiredColumnsExtractor.class,
ruleConfigs).entrySet()) {
result.putAll(entry.getValue().getTableAndRequiredColumnsMap(entry.getKey(),
targetTableNames));
}
return result;
diff --git
a/kernel/data-pipeline/scenario/cdc/core/src/test/java/org/apache/shardingsphere/data/pipeline/cdc/CDCJobTest.java
b/kernel/data-pipeline/scenario/cdc/core/src/test/java/org/apache/shardingsphere/data/pipeline/cdc/CDCJobTest.java
index eac44d4a422..a2f161eae5b 100644
---
a/kernel/data-pipeline/scenario/cdc/core/src/test/java/org/apache/shardingsphere/data/pipeline/cdc/CDCJobTest.java
+++
b/kernel/data-pipeline/scenario/cdc/core/src/test/java/org/apache/shardingsphere/data/pipeline/cdc/CDCJobTest.java
@@ -60,6 +60,7 @@ import
org.apache.shardingsphere.data.pipeline.core.util.PipelineDistributedBarr
import org.apache.shardingsphere.elasticjob.api.ShardingContext;
import org.apache.shardingsphere.elasticjob.infra.spi.ElasticJobServiceLoader;
import
org.apache.shardingsphere.infra.algorithm.core.config.AlgorithmConfiguration;
+import org.apache.shardingsphere.infra.config.rule.RuleConfiguration;
import org.apache.shardingsphere.infra.datanode.DataNode;
import org.apache.shardingsphere.infra.instance.metadata.InstanceType;
import
org.apache.shardingsphere.infra.metadata.identifier.ShardingSphereIdentifier;
@@ -67,6 +68,7 @@ import
org.apache.shardingsphere.infra.spi.type.ordered.OrderedSPILoader;
import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
import org.apache.shardingsphere.infra.yaml.config.pojo.YamlRootConfiguration;
import
org.apache.shardingsphere.infra.yaml.config.pojo.rule.YamlRuleConfiguration;
+import
org.apache.shardingsphere.infra.yaml.config.swapper.rule.YamlRuleConfigurationSwapperEngine;
import
org.apache.shardingsphere.test.infra.framework.extension.mock.AutoMockExtension;
import
org.apache.shardingsphere.test.infra.framework.extension.mock.StaticMockSettings;
import org.junit.jupiter.api.Test;
@@ -87,6 +89,7 @@ import java.util.concurrent.atomic.AtomicReference;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.sameInstance;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.ArgumentMatchers.any;
@@ -195,14 +198,28 @@ class CDCJobTest {
when(PipelineAPIFactory.getPipelineGovernanceFacade(CONTEXT_KEY)).thenReturn(mock(PipelineGovernanceFacade.class,
RETURNS_DEEP_STUBS));
when(PipelineDistributedBarrier.getInstance(CONTEXT_KEY)).thenReturn(mock(PipelineDistributedBarrier.class));
when(PipelineDataSourceConfigurationFactory.newInstance(anyString(),
anyString())).thenReturn(mock(PipelineDataSourceConfiguration.class));
- YamlRuleConfiguration ruleConfig = mock(YamlRuleConfiguration.class);
+ YamlRuleConfiguration yamlRuleConfig =
mock(YamlRuleConfiguration.class);
+ Collection<YamlRuleConfiguration> yamlRuleConfigs =
Collections.singleton(yamlRuleConfig);
+
jobConfig.getDataSourceConfig().getRootConfig().setRules(yamlRuleConfigs);
+ RuleConfiguration ruleConfig = mock(RuleConfiguration.class);
+ Collection<RuleConfiguration> ruleConfigs =
Collections.singleton(ruleConfig);
PipelineRequiredColumnsExtractor extractor =
mock(PipelineRequiredColumnsExtractor.class);
Map<ShardingSphereIdentifier, Collection<String>> requiredColumns =
Collections.singletonMap(new ShardingSphereIdentifier("logic_tbl"),
Collections.singleton("id"));
when(extractor.getTableAndRequiredColumnsMap(eq(ruleConfig),
anyCollection())).thenReturn(requiredColumns);
-
when(OrderedSPILoader.getServices(eq(PipelineRequiredColumnsExtractor.class),
anyCollection())).thenReturn(Collections.singletonMap(ruleConfig, extractor));
+ AtomicReference<Collection<RuleConfiguration>> capturedRuleConfigs =
new AtomicReference<>();
+
when(OrderedSPILoader.getServices(eq(PipelineRequiredColumnsExtractor.class),
anyCollection())).thenAnswer(invocation -> {
+ capturedRuleConfigs.set(invocation.getArgument(1));
+ return Collections.singletonMap(ruleConfig, extractor);
+ });
+ AtomicReference<Collection<YamlRuleConfiguration>>
capturedYamlRuleConfigs = new AtomicReference<>();
AtomicReference<CDCJobItemContext> capturedContext = new
AtomicReference<>();
try (
MockedStatic<TypedSPILoader> typedSPILoader =
mockStatic(TypedSPILoader.class);
+ MockedConstruction<YamlRuleConfigurationSwapperEngine>
ignoredRuleConfigSwapper = mockConstruction(
+ YamlRuleConfigurationSwapperEngine.class, (mock,
context) ->
when(mock.swapToRuleConfigurations(anyCollection())).thenAnswer(invocation -> {
+
capturedYamlRuleConfigs.set(invocation.getArgument(0));
+ return ruleConfigs;
+ }));
MockedConstruction<PipelineProcessConfigurationPersistService>
ignoredProcess = mockPersistService(processConfig);
MockedConstruction<CDCJobPreparer> ignoredPreparer =
mockConstruction(CDCJobPreparer.class, (mock, context) -> doAnswer(invocation
-> {
CDCJobItemContext jobItemContext =
((Collection<CDCJobItemContext>) invocation.getArgument(0)).iterator().next();
@@ -222,6 +239,9 @@ class CDCJobTest {
new CDCJob(mock(PipelineSink.class)).execute(shardingContext);
}
assertThat(capturedContext.get().getStatus(),
is(JobStatus.EXECUTE_INCREMENTAL_TASK));
+ assertThat(capturedYamlRuleConfigs.get(),
sameInstance(yamlRuleConfigs));
+ assertThat(capturedRuleConfigs.get(), sameInstance(ruleConfigs));
+
assertThat(capturedContext.get().getTaskConfig().getImporterConfig().getShardingColumns("logic_tbl"),
is(Collections.singleton("id")));
}
@SuppressWarnings("unchecked")
diff --git
a/kernel/data-pipeline/scenario/migration/core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/MigrationJobExecutorCallback.java
b/kernel/data-pipeline/scenario/migration/core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/MigrationJobExecutorCallback.java
index 35c4d125ec9..15723ed22ff 100644
---
a/kernel/data-pipeline/scenario/migration/core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/MigrationJobExecutorCallback.java
+++
b/kernel/data-pipeline/scenario/migration/core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/MigrationJobExecutorCallback.java
@@ -47,13 +47,14 @@ import
org.apache.shardingsphere.database.connector.core.metadata.identifier.Ide
import
org.apache.shardingsphere.database.connector.core.metadata.identifier.IdentifierNormalizeEngine;
import
org.apache.shardingsphere.database.connector.core.metadata.identifier.IdentifierScope;
import
org.apache.shardingsphere.database.connector.core.type.DatabaseTypeRegistry;
+import org.apache.shardingsphere.infra.config.rule.RuleConfiguration;
import org.apache.shardingsphere.infra.datanode.DataNode;
import
org.apache.shardingsphere.infra.datasource.pool.props.domain.DataSourcePoolProperties;
import
org.apache.shardingsphere.infra.metadata.database.resource.unit.StorageUnit;
import org.apache.shardingsphere.infra.metadata.database.schema.QualifiedTable;
import
org.apache.shardingsphere.infra.metadata.identifier.ShardingSphereIdentifier;
import org.apache.shardingsphere.infra.spi.type.ordered.OrderedSPILoader;
-import
org.apache.shardingsphere.infra.yaml.config.pojo.rule.YamlRuleConfiguration;
+import
org.apache.shardingsphere.infra.yaml.config.swapper.rule.YamlRuleConfigurationSwapperEngine;
import java.sql.SQLException;
import java.util.Collection;
@@ -127,9 +128,10 @@ public final class MigrationJobExecutorCallback implements
DistributedPipelineJo
@SuppressWarnings({"rawtypes", "unchecked"})
private Map<ShardingSphereIdentifier, Collection<String>>
getTableAndRequiredColumnsMap(final MigrationJobConfiguration jobConfig) {
Map<ShardingSphereIdentifier, Collection<String>> result = new
HashMap<>();
- Collection<YamlRuleConfiguration> yamlRuleConfigs =
((ShardingSpherePipelineDataSourceConfiguration)
jobConfig.getTarget()).getRootConfig().getRules();
+ Collection<RuleConfiguration> ruleConfigs = new
YamlRuleConfigurationSwapperEngine().swapToRuleConfigurations(
+ ((ShardingSpherePipelineDataSourceConfiguration)
jobConfig.getTarget()).getRootConfig().getRules());
Collection<ShardingSphereIdentifier> targetTableNames =
jobConfig.getTargetTableNames().stream().map(ShardingSphereIdentifier::new).collect(Collectors.toSet());
- for (Entry<YamlRuleConfiguration, PipelineRequiredColumnsExtractor>
entry : OrderedSPILoader.getServices(PipelineRequiredColumnsExtractor.class,
yamlRuleConfigs).entrySet()) {
+ for (Entry<RuleConfiguration, PipelineRequiredColumnsExtractor> entry
: OrderedSPILoader.getServices(PipelineRequiredColumnsExtractor.class,
ruleConfigs).entrySet()) {
result.putAll(entry.getValue().getTableAndRequiredColumnsMap(entry.getKey(),
targetTableNames));
}
return result;
diff --git
a/kernel/data-pipeline/scenario/migration/core/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/migration/MigrationJobExecutorCallbackTest.java
b/kernel/data-pipeline/scenario/migration/core/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/migration/MigrationJobExecutorCallbackTest.java
index beeadf15773..bd0ec4a0cf7 100644
---
a/kernel/data-pipeline/scenario/migration/core/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/migration/MigrationJobExecutorCallbackTest.java
+++
b/kernel/data-pipeline/scenario/migration/core/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/migration/MigrationJobExecutorCallbackTest.java
@@ -28,6 +28,7 @@ import
org.apache.shardingsphere.data.pipeline.core.datanode.JobDataNodeLine;
import
org.apache.shardingsphere.data.pipeline.core.datasource.PipelineDataSource;
import
org.apache.shardingsphere.data.pipeline.core.datasource.PipelineDataSourceManager;
import
org.apache.shardingsphere.data.pipeline.core.importer.ImporterConfiguration;
+import
org.apache.shardingsphere.data.pipeline.core.importer.PipelineRequiredColumnsExtractor;
import org.apache.shardingsphere.data.pipeline.core.job.api.PipelineAPIFactory;
import org.apache.shardingsphere.data.pipeline.core.job.id.PipelineJobIdUtils;
import
org.apache.shardingsphere.data.pipeline.core.job.progress.config.PipelineProcessConfiguration;
@@ -44,13 +45,18 @@ import
org.apache.shardingsphere.database.connector.core.metadata.identifier.Ide
import org.apache.shardingsphere.database.connector.core.type.DatabaseType;
import
org.apache.shardingsphere.database.connector.core.type.DatabaseTypeFactory;
import
org.apache.shardingsphere.database.connector.core.type.DatabaseTypeRegistry;
+import org.apache.shardingsphere.infra.config.rule.RuleConfiguration;
import org.apache.shardingsphere.infra.datanode.DataNode;
import
org.apache.shardingsphere.infra.datasource.pool.props.domain.DataSourcePoolProperties;
import org.apache.shardingsphere.infra.instance.metadata.InstanceType;
import
org.apache.shardingsphere.infra.metadata.database.resource.unit.StorageUnit;
import org.apache.shardingsphere.infra.metadata.database.schema.QualifiedTable;
+import
org.apache.shardingsphere.infra.metadata.identifier.ShardingSphereIdentifier;
+import org.apache.shardingsphere.infra.spi.type.ordered.OrderedSPILoader;
import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
import org.apache.shardingsphere.infra.yaml.config.pojo.YamlRootConfiguration;
+import
org.apache.shardingsphere.infra.yaml.config.pojo.rule.YamlRuleConfiguration;
+import
org.apache.shardingsphere.infra.yaml.config.swapper.rule.YamlRuleConfigurationSwapperEngine;
import org.apache.shardingsphere.mode.manager.ContextManager;
import org.junit.jupiter.api.Test;
import org.mockito.MockedConstruction;
@@ -60,10 +66,13 @@ import java.util.Collection;
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.Map;
+import java.util.concurrent.atomic.AtomicReference;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.sameInstance;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyCollection;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.ArgumentMatchers.isNull;
@@ -111,6 +120,45 @@ class MigrationJobExecutorCallbackTest {
}
}
+ @SuppressWarnings({"rawtypes", "unchecked"})
+ @Test
+ void assertBuildJobItemContextWithConvertedRuleConfigurations() {
+ YamlRuleConfiguration yamlRuleConfig =
mock(YamlRuleConfiguration.class);
+ Collection<YamlRuleConfiguration> yamlRuleConfigs =
Collections.singleton(yamlRuleConfig);
+ RuleConfiguration ruleConfig = mock(RuleConfiguration.class);
+ Collection<RuleConfiguration> ruleConfigs =
Collections.singleton(ruleConfig);
+ PipelineRequiredColumnsExtractor extractor =
mock(PipelineRequiredColumnsExtractor.class);
+ Map<ShardingSphereIdentifier, Collection<String>> requiredColumns =
Collections.singletonMap(new ShardingSphereIdentifier("t_order"),
Collections.singleton("id"));
+ when(extractor.getTableAndRequiredColumnsMap(eq(ruleConfig),
anyCollection())).thenReturn(requiredColumns);
+ AtomicReference<Collection<YamlRuleConfiguration>>
capturedYamlRuleConfigs = new AtomicReference<>();
+ AtomicReference<Collection<RuleConfiguration>> capturedRuleConfigs =
new AtomicReference<>();
+ try (
+ MockedConstruction<YamlRuleConfigurationSwapperEngine>
ignoredRuleConfigSwapper = mockConstruction(
+ YamlRuleConfigurationSwapperEngine.class, (mock,
context) ->
when(mock.swapToRuleConfigurations(anyCollection())).thenAnswer(invocation -> {
+
capturedYamlRuleConfigs.set(invocation.getArgument(0));
+ return ruleConfigs;
+ }));
+ MockedStatic<OrderedSPILoader> orderedSPILoader =
mockStatic(OrderedSPILoader.class);
+ MockedStatic<DatabaseTypeFactory> databaseTypeFactory =
mockStatic(DatabaseTypeFactory.class);
+ MockedStatic<PipelineAPIFactory> pipelineAPIFactory =
mockStatic(PipelineAPIFactory.class);
+ MockedStatic<PipelineContextManager> pipelineContextManager =
mockStatic(PipelineContextManager.class)) {
+ orderedSPILoader.when(() ->
OrderedSPILoader.getServices(eq(PipelineRequiredColumnsExtractor.class),
anyCollection())).thenAnswer(invocation -> {
+ capturedRuleConfigs.set(invocation.getArgument(1));
+ return Collections.singletonMap(ruleConfig, extractor);
+ });
+ mockDatabaseTypeFactory(databaseTypeFactory);
+ MigrationJobConfiguration jobConfig = createJobConfiguration();
+ ((ShardingSpherePipelineDataSourceConfiguration)
jobConfig.getTarget()).getRootConfig().setRules(yamlRuleConfigs);
+ mockGovernanceFacade(pipelineAPIFactory,
createSourceDataSourceYaml(10, 20));
+ mockProxyContext(pipelineContextManager);
+ MigrationJobItemContext actual = new
MigrationJobExecutorCallback().buildJobItemContext(
+ jobConfig, 0, null, mockProcessContext(),
mock(PipelineDataSourceManager.class));
+ assertThat(capturedYamlRuleConfigs.get(),
sameInstance(yamlRuleConfigs));
+ assertThat(capturedRuleConfigs.get(), sameInstance(ruleConfigs));
+
assertThat(actual.getTaskConfig().getImporterConfig().getShardingColumns("t_order"),
is(Collections.singleton("id")));
+ }
+ }
+
@Test
void assertBuildJobItemContextWithoutMigrationSourceStorageUnit() {
try (
diff --git
a/test/it/pipeline/src/test/java/org/apache/shardingsphere/test/it/data/pipeline/core/util/PipelineContextUtils.java
b/test/it/pipeline/src/test/java/org/apache/shardingsphere/test/it/data/pipeline/core/util/PipelineContextUtils.java
index d61c0d122aa..19eb0dfc313 100644
---
a/test/it/pipeline/src/test/java/org/apache/shardingsphere/test/it/data/pipeline/core/util/PipelineContextUtils.java
+++
b/test/it/pipeline/src/test/java/org/apache/shardingsphere/test/it/data/pipeline/core/util/PipelineContextUtils.java
@@ -60,7 +60,6 @@ import
org.apache.shardingsphere.infra.spi.type.ordered.OrderedSPILoader;
import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
import org.apache.shardingsphere.infra.util.file.SystemResourceFileUtils;
import org.apache.shardingsphere.infra.yaml.config.pojo.YamlRootConfiguration;
-import
org.apache.shardingsphere.infra.yaml.config.pojo.rule.YamlRuleConfiguration;
import
org.apache.shardingsphere.infra.yaml.config.swapper.mode.YamlModeConfigurationSwapper;
import
org.apache.shardingsphere.infra.yaml.config.swapper.resource.YamlDataSourceConfigurationSwapper;
import
org.apache.shardingsphere.infra.yaml.config.swapper.rule.YamlRuleConfigurationSwapperEngine;
@@ -221,9 +220,10 @@ public final class PipelineContextUtils {
@SuppressWarnings({"rawtypes", "unchecked"})
private static Map<ShardingSphereIdentifier, Collection<String>>
getTableAndRequiredColumnsMap(final MigrationJobConfiguration jobConfig) {
Map<ShardingSphereIdentifier, Collection<String>> result = new
HashMap<>();
- Collection<YamlRuleConfiguration> yamlRuleConfigs =
((ShardingSpherePipelineDataSourceConfiguration)
jobConfig.getTarget()).getRootConfig().getRules();
+ Collection<RuleConfiguration> ruleConfigs = new
YamlRuleConfigurationSwapperEngine().swapToRuleConfigurations(
+ ((ShardingSpherePipelineDataSourceConfiguration)
jobConfig.getTarget()).getRootConfig().getRules());
Set<ShardingSphereIdentifier> targetTableNames =
jobConfig.getTargetTableNames().stream().map(ShardingSphereIdentifier::new).collect(Collectors.toSet());
- for (Entry<YamlRuleConfiguration, PipelineRequiredColumnsExtractor>
entry : OrderedSPILoader.getServices(PipelineRequiredColumnsExtractor.class,
yamlRuleConfigs).entrySet()) {
+ for (Entry<RuleConfiguration, PipelineRequiredColumnsExtractor> entry
: OrderedSPILoader.getServices(PipelineRequiredColumnsExtractor.class,
ruleConfigs).entrySet()) {
result.putAll(entry.getValue().getTableAndRequiredColumnsMap(entry.getKey(),
targetTableNames));
}
return result;
