This is an automated email from the ASF dual-hosted git repository.
zhangliang 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 7fc36aa4c09 Refactor QualifiedTable (#34679)
7fc36aa4c09 is described below
commit 7fc36aa4c09d58b2f174908d3b293e904c846c37
Author: Liang Zhang <[email protected]>
AuthorDate: Sat Feb 15 17:59:58 2025 +0800
Refactor QualifiedTable (#34679)
---
.../infra/metadata/database/schema/QualifiedSchema.java | 6 +++++-
.../infra/metadata/database/schema/QualifiedTable.java | 10 +++++-----
.../infra/metadata/database/schema/QualifiedSchemaTest.java | 5 +++++
.../{QualifiedSchemaTest.java => QualifiedTableTest.java} | 9 +++++++--
.../metadata/database/schema/util/IndexMetaDataUtilsTest.java | 2 +-
.../pipeline/core/checker/PipelineDataSourceCheckEngine.java | 2 +-
.../preparer/datasource/PipelineJobDataSourcePreparer.java | 6 +++---
.../core/sqlbuilder/segment/PipelineSQLSegmentBuilder.java | 2 +-
.../check/consistency/MigrationDataConsistencyChecker.java | 2 +-
.../single/decider/SingleSQLFederationDecider.java | 4 ++--
.../shardingsphere/single/route/engine/SingleRouteEngine.java | 4 ++--
.../java/org/apache/shardingsphere/single/rule/SingleRule.java | 6 +++---
.../org/apache/shardingsphere/single/rule/SingleRuleTest.java | 4 ++--
.../test/e2e/data/pipeline/cases/cdc/CDCE2EIT.java | 2 +-
14 files changed, 39 insertions(+), 25 deletions(-)
diff --git
a/infra/common/src/main/java/org/apache/shardingsphere/infra/metadata/database/schema/QualifiedSchema.java
b/infra/common/src/main/java/org/apache/shardingsphere/infra/metadata/database/schema/QualifiedSchema.java
index c4bec5ff504..2f4c8893e97 100644
---
a/infra/common/src/main/java/org/apache/shardingsphere/infra/metadata/database/schema/QualifiedSchema.java
+++
b/infra/common/src/main/java/org/apache/shardingsphere/infra/metadata/database/schema/QualifiedSchema.java
@@ -30,6 +30,10 @@ public final class QualifiedSchema {
private final ShardingSphereIdentifier schemaName;
+ public QualifiedSchema(final String schemaName) {
+ this(null, schemaName);
+ }
+
public QualifiedSchema(final String databaseName, final String schemaName)
{
this.databaseName = new ShardingSphereIdentifier(databaseName);
this.schemaName = new ShardingSphereIdentifier(schemaName);
@@ -55,6 +59,6 @@ public final class QualifiedSchema {
@Override
public String toString() {
- return String.join(".", getDatabaseName(), getSchemaName());
+ return null == getDatabaseName() ? getSchemaName() : String.join(".",
getDatabaseName(), getSchemaName());
}
}
diff --git
a/infra/common/src/main/java/org/apache/shardingsphere/infra/metadata/database/schema/QualifiedTable.java
b/infra/common/src/main/java/org/apache/shardingsphere/infra/metadata/database/schema/QualifiedTable.java
index d8a33d3e6d3..266bef509b4 100644
---
a/infra/common/src/main/java/org/apache/shardingsphere/infra/metadata/database/schema/QualifiedTable.java
+++
b/infra/common/src/main/java/org/apache/shardingsphere/infra/metadata/database/schema/QualifiedTable.java
@@ -26,12 +26,12 @@ import
org.apache.shardingsphere.infra.metadata.identifier.ShardingSphereIdentif
@EqualsAndHashCode
public final class QualifiedTable {
- private final ShardingSphereIdentifier schemaName;
+ private final QualifiedSchema schema;
private final ShardingSphereIdentifier tableName;
public QualifiedTable(final String schemaName, final String tableName) {
- this.schemaName = new ShardingSphereIdentifier(schemaName);
+ this.schema = new QualifiedSchema(schemaName);
this.tableName = new ShardingSphereIdentifier(tableName);
}
@@ -40,8 +40,8 @@ public final class QualifiedTable {
*
* @return schema name
*/
- public String getSchemaName() {
- return schemaName.getValue();
+ public String getSchema() {
+ return schema.getSchemaName();
}
/**
@@ -55,6 +55,6 @@ public final class QualifiedTable {
@Override
public String toString() {
- return null == schemaName.getValue() ? tableName.getValue() :
String.join(".", schemaName.getValue(), tableName.getValue());
+ return null == getSchema() ? tableName.getValue() : String.join(".",
getSchema(), tableName.getValue());
}
}
diff --git
a/infra/common/src/test/java/org/apache/shardingsphere/infra/metadata/database/schema/QualifiedSchemaTest.java
b/infra/common/src/test/java/org/apache/shardingsphere/infra/metadata/database/schema/QualifiedSchemaTest.java
index 6ff633852e7..02e63665587 100644
---
a/infra/common/src/test/java/org/apache/shardingsphere/infra/metadata/database/schema/QualifiedSchemaTest.java
+++
b/infra/common/src/test/java/org/apache/shardingsphere/infra/metadata/database/schema/QualifiedSchemaTest.java
@@ -24,6 +24,11 @@ import static org.hamcrest.MatcherAssert.assertThat;
class QualifiedSchemaTest {
+ @Test
+ void assertToStringWithoutDatabase() {
+ assertThat(new QualifiedSchema("foo_schema").toString(),
is("foo_schema"));
+ }
+
@Test
void assertToString() {
assertThat(new QualifiedSchema("foo_db", "foo_schema").toString(),
is("foo_db.foo_schema"));
diff --git
a/infra/common/src/test/java/org/apache/shardingsphere/infra/metadata/database/schema/QualifiedSchemaTest.java
b/infra/common/src/test/java/org/apache/shardingsphere/infra/metadata/database/schema/QualifiedTableTest.java
similarity index 79%
copy from
infra/common/src/test/java/org/apache/shardingsphere/infra/metadata/database/schema/QualifiedSchemaTest.java
copy to
infra/common/src/test/java/org/apache/shardingsphere/infra/metadata/database/schema/QualifiedTableTest.java
index 6ff633852e7..242498a03bc 100644
---
a/infra/common/src/test/java/org/apache/shardingsphere/infra/metadata/database/schema/QualifiedSchemaTest.java
+++
b/infra/common/src/test/java/org/apache/shardingsphere/infra/metadata/database/schema/QualifiedTableTest.java
@@ -22,10 +22,15 @@ import org.junit.jupiter.api.Test;
import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.MatcherAssert.assertThat;
-class QualifiedSchemaTest {
+class QualifiedTableTest {
+
+ @Test
+ void assertToStringWithoutSchema() {
+ assertThat(new QualifiedTable(null, "foo_tbl").toString(),
is("foo_tbl"));
+ }
@Test
void assertToString() {
- assertThat(new QualifiedSchema("foo_db", "foo_schema").toString(),
is("foo_db.foo_schema"));
+ assertThat(new QualifiedTable("foo_schema", "foo_tbl").toString(),
is("foo_schema.foo_tbl"));
}
}
diff --git
a/infra/common/src/test/java/org/apache/shardingsphere/infra/metadata/database/schema/util/IndexMetaDataUtilsTest.java
b/infra/common/src/test/java/org/apache/shardingsphere/infra/metadata/database/schema/util/IndexMetaDataUtilsTest.java
index 71d5a2c6cdb..116ff865a93 100644
---
a/infra/common/src/test/java/org/apache/shardingsphere/infra/metadata/database/schema/util/IndexMetaDataUtilsTest.java
+++
b/infra/common/src/test/java/org/apache/shardingsphere/infra/metadata/database/schema/util/IndexMetaDataUtilsTest.java
@@ -79,7 +79,7 @@ class IndexMetaDataUtilsTest {
IndexSegment indexSegment = new IndexSegment(0, 0, new
IndexNameSegment(0, 0, new IdentifierValue("foo_idx")));
Collection<QualifiedTable> actual =
IndexMetaDataUtils.getTableNames(buildDatabase(),
TypedSPILoader.getService(DatabaseType.class, "FIXTURE"),
Collections.singleton(indexSegment));
assertThat(actual.size(), is(1));
- assertThat(actual.iterator().next().getSchemaName(), is("foo_db"));
+ assertThat(actual.iterator().next().getSchema(), is("foo_db"));
assertThat(actual.iterator().next().getTableName(), is("foo_tbl"));
}
diff --git
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/checker/PipelineDataSourceCheckEngine.java
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/checker/PipelineDataSourceCheckEngine.java
index 7440198ea08..2506c934ad8 100644
---
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/checker/PipelineDataSourceCheckEngine.java
+++
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/checker/PipelineDataSourceCheckEngine.java
@@ -108,7 +108,7 @@ public final class PipelineDataSourceCheckEngine {
* @throws SQLException if there's database operation failure
*/
public boolean checkEmptyTable(final DataSource dataSource, final
QualifiedTable qualifiedTable) throws SQLException {
- String sql =
sqlBuilder.buildCheckEmptyTableSQL(qualifiedTable.getSchemaName(),
qualifiedTable.getTableName());
+ String sql =
sqlBuilder.buildCheckEmptyTableSQL(qualifiedTable.getSchema(),
qualifiedTable.getTableName());
try (
Connection connection = dataSource.getConnection();
PreparedStatement preparedStatement =
connection.prepareStatement(sql);
diff --git
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/preparer/datasource/PipelineJobDataSourcePreparer.java
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/preparer/datasource/PipelineJobDataSourcePreparer.java
index 9d127680312..27bfb8b14a7 100644
---
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/preparer/datasource/PipelineJobDataSourcePreparer.java
+++
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/preparer/datasource/PipelineJobDataSourcePreparer.java
@@ -77,7 +77,7 @@ public final class PipelineJobDataSourcePreparer {
PipelinePrepareSQLBuilder pipelineSQLBuilder = new
PipelinePrepareSQLBuilder(targetDatabaseType);
Collection<String> createdSchemaNames = new
HashSet<>(param.getCreateTableConfigurations().size(), 1F);
for (CreateTableConfiguration each :
param.getCreateTableConfigurations()) {
- String targetSchemaName = each.getTargetName().getSchemaName();
+ String targetSchemaName = each.getTargetName().getSchema();
if (null == targetSchemaName ||
targetSchemaName.equalsIgnoreCase(defaultSchema) ||
createdSchemaNames.contains(targetSchemaName)) {
continue;
}
@@ -132,7 +132,7 @@ public final class PipelineJobDataSourcePreparer {
private List<String> getCreateTargetTableSQL(final
CreateTableConfiguration createTableConfig, final PipelineDataSourceManager
dataSourceManager) throws SQLException {
DatabaseType databaseType =
createTableConfig.getSourceDataSourceConfig().getDatabaseType();
DataSource sourceDataSource =
dataSourceManager.getDataSource(createTableConfig.getSourceDataSourceConfig());
- String schemaName = createTableConfig.getSourceName().getSchemaName();
+ String schemaName = createTableConfig.getSourceName().getSchema();
String sourceTableName =
createTableConfig.getSourceName().getTableName();
String targetTableName =
createTableConfig.getTargetName().getTableName();
return PipelineDDLGenerator.generateLogicDDL(databaseType,
sourceDataSource, schemaName, sourceTableName, targetTableName);
@@ -140,7 +140,7 @@ public final class PipelineJobDataSourcePreparer {
private Optional<String> decorateTargetTableSQL(final
CreateTableConfiguration createTableConfig, final SQLParserEngine
sqlParserEngine,
final
ShardingSphereMetaData metaData, final String targetDatabaseName, final String
sql) {
- String schemaName = createTableConfig.getSourceName().getSchemaName();
+ String schemaName = createTableConfig.getSourceName().getSchema();
String targetTableName =
createTableConfig.getTargetName().getTableName();
Optional<String> decoratedSQL = new
PipelineDDLDecorator(metaData).decorate(databaseType, targetDatabaseName,
schemaName, targetTableName, sqlParserEngine, sql);
return decoratedSQL.map(String::trim).filter(trimmedSql ->
!Strings.isNullOrEmpty(trimmedSql));
diff --git
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/segment/PipelineSQLSegmentBuilder.java
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/segment/PipelineSQLSegmentBuilder.java
index 2327ffb7a34..095bff3e23e 100644
---
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/segment/PipelineSQLSegmentBuilder.java
+++
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/segment/PipelineSQLSegmentBuilder.java
@@ -67,6 +67,6 @@ public final class PipelineSQLSegmentBuilder {
* @return qualified table name
*/
public String getQualifiedTableName(final QualifiedTable qualifiedTable) {
- return getQualifiedTableName(qualifiedTable.getSchemaName(),
qualifiedTable.getTableName());
+ return getQualifiedTableName(qualifiedTable.getSchema(),
qualifiedTable.getTableName());
}
}
diff --git
a/kernel/data-pipeline/scenario/migration/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/check/consistency/MigrationDataConsistencyChecker.java
b/kernel/data-pipeline/scenario/migration/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/check/consistency/MigrationDataConsistencyChecker.java
index 06c9a484ef1..4f424b06bdd 100644
---
a/kernel/data-pipeline/scenario/migration/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/check/consistency/MigrationDataConsistencyChecker.java
+++
b/kernel/data-pipeline/scenario/migration/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/check/consistency/MigrationDataConsistencyChecker.java
@@ -133,7 +133,7 @@ public final class MigrationDataConsistencyChecker
implements PipelineDataConsis
ShardingSpherePreconditions.checkNotNull(tableMetaData,
() -> new
PipelineTableDataConsistencyCheckLoadingFailedException(new
QualifiedTable(dataNode.getSchemaName(), dataNode.getTableName())));
List<String> columnNames = tableMetaData.getColumnNames();
- List<PipelineColumnMetaData> uniqueKeys =
PipelineTableMetaDataUtils.getUniqueKeyColumns(sourceTable.getSchemaName(),
sourceTable.getTableName(), metaDataLoader);
+ List<PipelineColumnMetaData> uniqueKeys =
PipelineTableMetaDataUtils.getUniqueKeyColumns(sourceTable.getSchema(),
sourceTable.getTableName(), metaDataLoader);
TableInventoryCheckParameter param = new TableInventoryCheckParameter(
jobConfig.getJobId(), sourceDataSource, targetDataSource,
sourceTable, targetTable, columnNames, uniqueKeys, readRateLimitAlgorithm,
progressContext);
TableInventoryChecker tableInventoryChecker =
tableChecker.buildTableInventoryChecker(param);
diff --git
a/kernel/single/core/src/main/java/org/apache/shardingsphere/single/decider/SingleSQLFederationDecider.java
b/kernel/single/core/src/main/java/org/apache/shardingsphere/single/decider/SingleSQLFederationDecider.java
index e5e180db6e0..a8c2610cb1d 100644
---
a/kernel/single/core/src/main/java/org/apache/shardingsphere/single/decider/SingleSQLFederationDecider.java
+++
b/kernel/single/core/src/main/java/org/apache/shardingsphere/single/decider/SingleSQLFederationDecider.java
@@ -73,7 +73,7 @@ public final class SingleSQLFederationDecider implements
SQLFederationDecider<Si
private boolean containsView(final ShardingSphereDatabase database, final
Collection<QualifiedTable> singleTables) {
for (QualifiedTable each : singleTables) {
- if
(database.getSchema(each.getSchemaName()).containsView(each.getTableName())) {
+ if
(database.getSchema(each.getSchema()).containsView(each.getTableName())) {
return true;
}
}
@@ -83,7 +83,7 @@ public final class SingleSQLFederationDecider implements
SQLFederationDecider<Si
private Collection<DataNode> getTableDataNodes(final SingleRule rule,
final Collection<QualifiedTable> singleTables) {
Collection<DataNode> result = new HashSet<>(singleTables.size(), 1F);
for (QualifiedTable each : singleTables) {
-
rule.getAttributes().getAttribute(MutableDataNodeRuleAttribute.class).findTableDataNode(each.getSchemaName(),
each.getTableName()).ifPresent(result::add);
+
rule.getAttributes().getAttribute(MutableDataNodeRuleAttribute.class).findTableDataNode(each.getSchema(),
each.getTableName()).ifPresent(result::add);
}
return result;
}
diff --git
a/kernel/single/core/src/main/java/org/apache/shardingsphere/single/route/engine/SingleRouteEngine.java
b/kernel/single/core/src/main/java/org/apache/shardingsphere/single/route/engine/SingleRouteEngine.java
index 4223fbc364a..a201ea2efc2 100644
---
a/kernel/single/core/src/main/java/org/apache/shardingsphere/single/route/engine/SingleRouteEngine.java
+++
b/kernel/single/core/src/main/java/org/apache/shardingsphere/single/route/engine/SingleRouteEngine.java
@@ -95,7 +95,7 @@ public final class SingleRouteEngine {
private void routeDDLStatement(final RouteContext routeContext, final
SingleRule rule) {
if (sqlStatement instanceof CreateTableStatement) {
QualifiedTable table = singleTables.iterator().next();
- Optional<DataNode> dataNode =
rule.getAttributes().getAttribute(MutableDataNodeRuleAttribute.class).findTableDataNode(table.getSchemaName(),
table.getTableName());
+ Optional<DataNode> dataNode =
rule.getAttributes().getAttribute(MutableDataNodeRuleAttribute.class).findTableDataNode(table.getSchema(),
table.getTableName());
boolean containsIfNotExists = ((CreateTableStatement)
sqlStatement).isIfNotExists();
if (dataNode.isPresent()) {
routeDDLStatementWithExistTable(routeContext,
containsIfNotExists, dataNode.get(), table);
@@ -121,7 +121,7 @@ public final class SingleRouteEngine {
private void fillRouteContext(final SingleRule singleRule, final
RouteContext routeContext, final Collection<QualifiedTable> logicTables) {
for (QualifiedTable each : logicTables) {
String tableName = each.getTableName();
- DataNode dataNode =
singleRule.getAttributes().getAttribute(MutableDataNodeRuleAttribute.class).findTableDataNode(each.getSchemaName(),
tableName)
+ DataNode dataNode =
singleRule.getAttributes().getAttribute(MutableDataNodeRuleAttribute.class).findTableDataNode(each.getSchema(),
tableName)
.orElseThrow(() -> new
SingleTableNotFoundException(tableName));
String dataSource = dataNode.getDataSourceName();
routeContext.putRouteUnit(new RouteMapper(dataSource, dataSource),
Collections.singletonList(new RouteMapper(tableName, tableName)));
diff --git
a/kernel/single/core/src/main/java/org/apache/shardingsphere/single/rule/SingleRule.java
b/kernel/single/core/src/main/java/org/apache/shardingsphere/single/rule/SingleRule.java
index 50211a03eb3..4ae95eb3bf7 100644
---
a/kernel/single/core/src/main/java/org/apache/shardingsphere/single/rule/SingleRule.java
+++
b/kernel/single/core/src/main/java/org/apache/shardingsphere/single/rule/SingleRule.java
@@ -106,7 +106,7 @@ public final class SingleRule implements DatabaseRule {
return false;
}
QualifiedTable sampleTable = singleTables.iterator().next();
- Optional<DataNode> sampleDataNode =
mutableDataNodeRuleAttribute.findTableDataNode(sampleTable.getSchemaName(),
sampleTable.getTableName());
+ Optional<DataNode> sampleDataNode =
mutableDataNodeRuleAttribute.findTableDataNode(sampleTable.getSchema(),
sampleTable.getTableName());
if (sampleDataNode.isPresent()) {
for (DataNode each : dataNodes) {
if
(!isSameComputeNode(sampleDataNode.get().getDataSourceName(),
each.getDataSourceName())) {
@@ -124,7 +124,7 @@ public final class SingleRule implements DatabaseRule {
private boolean isSingleTablesInSameComputeNode(final
Collection<QualifiedTable> singleTables) {
String sampleDataSourceName = null;
for (QualifiedTable each : singleTables) {
- Optional<DataNode> dataNode =
mutableDataNodeRuleAttribute.findTableDataNode(each.getSchemaName(),
each.getTableName());
+ Optional<DataNode> dataNode =
mutableDataNodeRuleAttribute.findTableDataNode(each.getSchema(),
each.getTableName());
if (!dataNode.isPresent()) {
continue;
}
@@ -158,7 +158,7 @@ public final class SingleRule implements DatabaseRule {
private boolean containsDataNode(final QualifiedTable qualifiedTable,
final Collection<DataNode> dataNodes) {
for (DataNode each : dataNodes) {
- if
(qualifiedTable.getSchemaName().equalsIgnoreCase(each.getSchemaName())) {
+ if
(qualifiedTable.getSchema().equalsIgnoreCase(each.getSchemaName())) {
return true;
}
}
diff --git
a/kernel/single/core/src/test/java/org/apache/shardingsphere/single/rule/SingleRuleTest.java
b/kernel/single/core/src/test/java/org/apache/shardingsphere/single/rule/SingleRuleTest.java
index e27025edeb8..2323eb5f87d 100644
---
a/kernel/single/core/src/test/java/org/apache/shardingsphere/single/rule/SingleRuleTest.java
+++
b/kernel/single/core/src/test/java/org/apache/shardingsphere/single/rule/SingleRuleTest.java
@@ -176,7 +176,7 @@ class SingleRuleTest {
SingleRule singleRule = new SingleRule(ruleConfig, "foo_db", new
H2DatabaseType(), dataSourceMap,
Collections.singleton(mock(ShardingSphereRule.class, RETURNS_DEEP_STUBS)));
Collection<QualifiedTable> tableNames = new LinkedList<>();
tableNames.add(new QualifiedTable("foo_db", "employee"));
-
assertThat(singleRule.getSingleTables(tableNames).iterator().next().getSchemaName(),
is("foo_db"));
+
assertThat(singleRule.getSingleTables(tableNames).iterator().next().getSchema(),
is("foo_db"));
assertThat(singleRule.getSingleTables(tableNames).iterator().next().getTableName(),
is("employee"));
}
@@ -188,7 +188,7 @@ class SingleRuleTest {
singleRule.getAttributes().getAttribute(MutableDataNodeRuleAttribute.class).put(dataSourceName,
"foo_db", tableName);
Collection<QualifiedTable> tableNames = new LinkedList<>();
tableNames.add(new QualifiedTable("foo_db", "teacher"));
-
assertThat(singleRule.getSingleTables(tableNames).iterator().next().getSchemaName(),
is("foo_db"));
+
assertThat(singleRule.getSingleTables(tableNames).iterator().next().getSchema(),
is("foo_db"));
assertThat(singleRule.getSingleTables(tableNames).iterator().next().getTableName(),
is("teacher"));
assertTrue(singleRule.getAttributes().getAttribute(TableMapperRuleAttribute.class).getLogicTableNames().contains("employee"));
assertTrue(singleRule.getAttributes().getAttribute(TableMapperRuleAttribute.class).getLogicTableNames().contains("student"));
diff --git
a/test/e2e/operation/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/data/pipeline/cases/cdc/CDCE2EIT.java
b/test/e2e/operation/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/data/pipeline/cases/cdc/CDCE2EIT.java
index 8297becec5e..9fff931f7f8 100644
---
a/test/e2e/operation/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/data/pipeline/cases/cdc/CDCE2EIT.java
+++
b/test/e2e/operation/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/data/pipeline/cases/cdc/CDCE2EIT.java
@@ -188,7 +188,7 @@ class CDCE2EIT {
private void assertDataMatched(final PipelineDataSource sourceDataSource,
final PipelineDataSource targetDataSource, final QualifiedTable qualifiedTable)
{
StandardPipelineTableMetaDataLoader metaDataLoader = new
StandardPipelineTableMetaDataLoader(targetDataSource);
- PipelineTableMetaData tableMetaData =
metaDataLoader.getTableMetaData(qualifiedTable.getSchemaName(),
qualifiedTable.getTableName());
+ PipelineTableMetaData tableMetaData =
metaDataLoader.getTableMetaData(qualifiedTable.getSchema(),
qualifiedTable.getTableName());
List<PipelineColumnMetaData> uniqueKeys =
Collections.singletonList(tableMetaData.getColumnMetaData(tableMetaData.getPrimaryKeyColumns().get(0)));
ConsistencyCheckJobItemProgressContext progressContext = new
ConsistencyCheckJobItemProgressContext("", 0,
sourceDataSource.getDatabaseType().getType());
TableInventoryCheckParameter param = new
TableInventoryCheckParameter("", sourceDataSource, targetDataSource,
qualifiedTable, qualifiedTable,
