This is an automated email from the ASF dual-hosted git repository.
sandynz 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 63ab282c627 Preserve Identifier Case Across Migration Pipelines
(#39623)
63ab282c627 is described below
commit 63ab282c627d6262059276aa03532b6f3abaca56
Author: Hongsheng Zhong <[email protected]>
AuthorDate: Fri Aug 28 15:07:33 2026 +0800
Preserve Identifier Case Across Migration Pipelines (#39623)
* Preserve identifier case in Migration jobs
* Normalize source table identifiers in Migration jobs
* Preserve actual identifiers in pipeline E2E cleanup
* Preserve target identifier case in Migration jobs
* Add case-sensitive MySQL Migration E2E
* Use endpoint-specific identifiers in Migration checks
* Simplify table inventory check parameters
* Enable case-sensitive table names for MySQL 5.7 E2E
---
.../table/MatchingTableInventoryChecker.java | 2 +-
.../table/TableInventoryCheckParameter.java | 14 +-
.../ingest/incremental/wal/WALEventConverter.java | 36 +++++-
.../component/table/default/get_table_id.ftl | 2 +-
.../incremental/wal/WALEventConverterTest.java | 27 +++-
.../PostgreSQLPipelineFreemarkerManagerTest.java | 10 ++
.../migration/MigrationJobExecutorCallback.java | 50 ++++++--
.../MigrationDataConsistencyChecker.java | 21 ++-
.../MigrationJobExecutorCallbackTest.java | 87 ++++++++++++-
.../handler/update/MigrateTableExecutor.java | 24 ++--
.../handler/update/MigrateTableExecutorTest.java | 46 +++++--
.../core/MigrationDistSQLStatementVisitor.java | 17 +--
.../core/MigrationDistSQLStatementVisitorTest.java | 48 +++++++
.../segment/MigrationSourceTargetSegment.java | 58 ++++++++-
.../general/MySQLMigrationIdentifierCaseE2EIT.java | 114 +++++++++++++++++
.../PostgreSQLMigrationIdentifierCaseE2EIT.java | 142 +++++++++++++++++++++
.../natived/PipelineNativeContainerComposer.java | 5 +-
...eSQLPipelineNativeContainerDropTableOption.java | 2 +-
.../resources/env/container/mysql/cnf/5/my.cnf | 2 +-
.../resources/env/container/mysql/cnf/8/my.cnf | 2 +-
20 files changed, 648 insertions(+), 61 deletions(-)
diff --git
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/MatchingTableInventoryChecker.java
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/MatchingTableInventoryChecker.java
index d92c397660b..a2ae2586699 100644
---
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/MatchingTableInventoryChecker.java
+++
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/MatchingTableInventoryChecker.java
@@ -131,7 +131,7 @@ public abstract class MatchingTableInventoryChecker
implements TableInventoryChe
private TableInventoryCalculateParameter
getTableInventoryCalculateParameter(final TableInventoryCheckParameter param,
final TableCheckRangePosition checkRangePosition) {
TableInventoryCalculateParameter result = new
TableInventoryCalculateParameter(param.getTargetDataSource(),
param.getTargetTable(),
- param.getColumnNames(), param.getUniqueKeys(),
QueryType.RANGE_QUERY, param.getQueryCondition());
+ param.getTargetColumnNames(), param.getTargetUniqueKeys(),
QueryType.RANGE_QUERY, param.getQueryCondition());
result.setRange(Range.closed(null !=
checkRangePosition.getTargetPosition() ? checkRangePosition.getTargetPosition()
: checkRangePosition.getTargetRange().getLowerBound(),
checkRangePosition.getTargetRange().getUpperBound()));
return result;
diff --git
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/TableInventoryCheckParameter.java
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/TableInventoryCheckParameter.java
index 2b45ad445c0..5cb2a6e79eb 100644
---
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/TableInventoryCheckParameter.java
+++
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/TableInventoryCheckParameter.java
@@ -30,8 +30,8 @@ import java.util.List;
/**
* Table inventory check parameter.
*/
-@RequiredArgsConstructor
@Getter
+@RequiredArgsConstructor
public final class TableInventoryCheckParameter {
private final String jobId;
@@ -50,12 +50,24 @@ public final class TableInventoryCheckParameter {
private final List<PipelineColumnMetaData> uniqueKeys;
+ private final List<String> targetColumnNames;
+
+ private final List<PipelineColumnMetaData> targetUniqueKeys;
+
private final JobRateLimitAlgorithm readRateLimitAlgorithm;
private final ConsistencyCheckJobItemProgressContext progressContext;
private final String queryCondition;
+ public TableInventoryCheckParameter(final String jobId, final int
splittingItem, final PipelineDataSource sourceDataSource, final
PipelineDataSource targetDataSource,
+ final QualifiedTable sourceTable,
final QualifiedTable targetTable,
+ final List<String> columnNames, final
List<PipelineColumnMetaData> uniqueKeys,
+ final JobRateLimitAlgorithm
readRateLimitAlgorithm, final ConsistencyCheckJobItemProgressContext
progressContext, final String queryCondition) {
+ this(jobId, splittingItem, sourceDataSource, targetDataSource,
sourceTable, targetTable, columnNames, uniqueKeys, columnNames, uniqueKeys,
+ readRateLimitAlgorithm, progressContext, queryCondition);
+ }
+
public TableInventoryCheckParameter(final String jobId, final
PipelineDataSource sourceDataSource, final PipelineDataSource targetDataSource,
final QualifiedTable sourceTable,
final QualifiedTable targetTable,
final List<String> columnNames, final
List<PipelineColumnMetaData> uniqueKeys,
diff --git
a/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/ingest/incremental/wal/WALEventConverter.java
b/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/ingest/incremental/wal/WALEventConverter.java
index 61153669589..4f7209fa27a 100644
---
a/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/ingest/incremental/wal/WALEventConverter.java
+++
b/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/ingest/incremental/wal/WALEventConverter.java
@@ -36,10 +36,14 @@ import
org.apache.shardingsphere.database.connector.core.metadata.database.enums
import
org.apache.shardingsphere.database.connector.core.metadata.identifier.IdentifierCasePolicy;
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.metadata.identifier.LookupMode;
import
org.apache.shardingsphere.infra.exception.generic.UnsupportedSQLOperationException;
import
org.apache.shardingsphere.infra.metadata.identifier.ShardingSphereIdentifier;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
/**
* WAL event converter.
@@ -52,6 +56,12 @@ public final class WALEventConverter {
private final IdentifierCasePolicy schemaIdentifierCasePolicy;
+ private final IdentifierCasePolicy tableIdentifierCasePolicy;
+
+ private final Map<String, ShardingSphereIdentifier> quotedTableNameMap;
+
+ private final Map<String, ShardingSphereIdentifier> unquotedTableNameMap;
+
/**
* Create a WAL event converter.
*
@@ -63,6 +73,18 @@ public final class WALEventConverter {
this.metaDataLoader = metaDataLoader;
schemaIdentifierCasePolicy = IdentifierNormalizeEngine.resolvePolicy(
dumperContext.getCommonContext().getDataSourceConfig().getDatabaseType(), null,
IdentifierScope.SCHEMA);
+ tableIdentifierCasePolicy = IdentifierNormalizeEngine.resolvePolicy(
+
dumperContext.getCommonContext().getDataSourceConfig().getDatabaseType(), null,
IdentifierScope.TABLE);
+ Map<ShardingSphereIdentifier, ShardingSphereIdentifier> tableNameMap =
dumperContext.getCommonContext().getTableNameMapper().getTableNameMap();
+ quotedTableNameMap = new HashMap<>(tableNameMap.size(), 1F);
+ unquotedTableNameMap = new HashMap<>(tableNameMap.size(), 1F);
+ for (Entry<ShardingSphereIdentifier, ShardingSphereIdentifier> entry :
tableNameMap.entrySet()) {
+ String actualTableName = entry.getKey().getValue();
+ quotedTableNameMap.put(getLookupKey(actualTableName,
QuoteCharacter.QUOTE), entry.getValue());
+ if (tableIdentifierCasePolicy.matches(actualTableName,
actualTableName, QuoteCharacter.NONE)) {
+ unquotedTableNameMap.put(getLookupKey(actualTableName,
QuoteCharacter.NONE), entry.getValue());
+ }
+ }
}
/**
@@ -77,8 +99,9 @@ public final class WALEventConverter {
return createPlaceholderRecord(event);
}
AbstractRowEvent rowEvent = (AbstractRowEvent) event;
- String actualTableName = rowEvent.getTableName();
- ShardingSphereIdentifier logicTableName =
dumperContext.getCommonContext().getTableNameMapper().getLogicTableName(actualTableName);
+ QuoteCharacter tableQuoteCharacter =
QuoteCharacter.getQuoteCharacter(rowEvent.getTableName());
+ String actualTableName =
tableQuoteCharacter.unwrap(rowEvent.getTableName());
+ ShardingSphereIdentifier logicTableName =
getLogicTableName(actualTableName, tableQuoteCharacter);
if (null == logicTableName) {
return createPlaceholderRecord(event);
}
@@ -99,6 +122,15 @@ public final class WALEventConverter {
throw new UnsupportedSQLOperationException("");
}
+ private ShardingSphereIdentifier getLogicTableName(final String
actualTableName, final QuoteCharacter quoteCharacter) {
+ Map<String, ShardingSphereIdentifier> tableNameMap =
QuoteCharacter.NONE == quoteCharacter ? unquotedTableNameMap :
quotedTableNameMap;
+ return tableNameMap.get(getLookupKey(actualTableName, quoteCharacter));
+ }
+
+ private String getLookupKey(final String identifier, final QuoteCharacter
quoteCharacter) {
+ return LookupMode.EXACT ==
tableIdentifierCasePolicy.getLookupMode(quoteCharacter) ? identifier :
tableIdentifierCasePolicy.normalizeForLookup(identifier);
+ }
+
private boolean isDifferentSchemaName(final AbstractRowEvent event, final
String expectedSchemaName) {
if (!isConcreteSchemaName(expectedSchemaName)) {
return false;
diff --git
a/kernel/data-pipeline/dialect/postgresql/src/main/resources/template/component/table/default/get_table_id.ftl
b/kernel/data-pipeline/dialect/postgresql/src/main/resources/template/component/table/default/get_table_id.ftl
index 44578d29dea..ccd20324e10 100644
---
a/kernel/data-pipeline/dialect/postgresql/src/main/resources/template/component/table/default/get_table_id.ftl
+++
b/kernel/data-pipeline/dialect/postgresql/src/main/resources/template/component/table/default/get_table_id.ftl
@@ -15,4 +15,4 @@
~ limitations under the License.
-->
-SELECT '${schemaName}.${tableName}'::REGCLASS::OID AS tid;
+SELECT (pg_catalog.quote_ident('${schemaName}') || '.' ||
pg_catalog.quote_ident('${tableName}'))::REGCLASS::OID AS tid;
diff --git
a/kernel/data-pipeline/dialect/postgresql/src/test/java/org/apache/shardingsphere/data/pipeline/postgresql/ingest/incremental/wal/WALEventConverterTest.java
b/kernel/data-pipeline/dialect/postgresql/src/test/java/org/apache/shardingsphere/data/pipeline/postgresql/ingest/incremental/wal/WALEventConverterTest.java
index 13e40c965c9..99bb889d291 100644
---
a/kernel/data-pipeline/dialect/postgresql/src/test/java/org/apache/shardingsphere/data/pipeline/postgresql/ingest/incremental/wal/WALEventConverterTest.java
+++
b/kernel/data-pipeline/dialect/postgresql/src/test/java/org/apache/shardingsphere/data/pipeline/postgresql/ingest/incremental/wal/WALEventConverterTest.java
@@ -203,6 +203,27 @@ class WALEventConverterTest {
assertThat(walEventConverter.convert(mockUnknownTableEvent()),
isA(PlaceholderRecord.class));
}
+ @Test
+ void assertConvertWithQuotedUpperCaseTable() {
+ PipelineTableMetaDataLoader metaDataLoader =
mock(PipelineTableMetaDataLoader.class);
+ when(metaDataLoader.getTableMetaData("UPPER_SCHEMA",
"UPPER_TABLE")).thenReturn(pipelineTableMetaData);
+ WALEventConverter converter =
createSchemaAwareWALEventConverter("UPPER_TABLE", "UPPER_SCHEMA",
metaDataLoader);
+ WriteRowEvent event = createWriteRowEvent("\"UPPER_SCHEMA\"");
+ event.setTableName("\"UPPER_TABLE\"");
+ assertThat(converter.convert(event), isA(DataRecord.class));
+ verify(metaDataLoader).getTableMetaData("UPPER_SCHEMA", "UPPER_TABLE");
+ }
+
+ @Test
+ void assertConvertWithQuotedTableCaseCollision() {
+ PipelineTableMetaDataLoader metaDataLoader =
mock(PipelineTableMetaDataLoader.class);
+ WALEventConverter converter =
createSchemaAwareWALEventConverter("UPPER_TABLE", "UPPER_SCHEMA",
metaDataLoader);
+ WriteRowEvent event = createWriteRowEvent("\"UPPER_SCHEMA\"");
+ event.setTableName("\"upper_table\"");
+ assertThat(converter.convert(event), isA(PlaceholderRecord.class));
+ verifyNoInteractions(metaDataLoader);
+ }
+
@ParameterizedTest(name = "{0}")
@MethodSource("provideDifferentConcreteSchemaNames")
void assertConvertWithDifferentConcreteSchema(final String name, final
String expectedSchemaName, final String eventSchemaName) {
@@ -263,10 +284,14 @@ class WALEventConverterTest {
}
private WALEventConverter createSchemaAwareWALEventConverter(final String
expectedSchemaName, final PipelineTableMetaDataLoader metaDataLoader) {
+ return createSchemaAwareWALEventConverter("t_order",
expectedSchemaName, metaDataLoader);
+ }
+
+ private WALEventConverter createSchemaAwareWALEventConverter(final String
actualTableName, final String expectedSchemaName, final
PipelineTableMetaDataLoader metaDataLoader) {
PipelineDataSourceConfiguration dataSourceConfig =
mock(PipelineDataSourceConfiguration.class);
when(dataSourceConfig.getDatabaseType()).thenReturn(databaseType);
DumperCommonContext commonContext = new DumperCommonContext(null,
dataSourceConfig,
- new ActualAndLogicTableNameMapper(Collections.singletonMap(new
ShardingSphereIdentifier("t_order"), new ShardingSphereIdentifier("t_order"))),
+ new ActualAndLogicTableNameMapper(Collections.singletonMap(new
ShardingSphereIdentifier(actualTableName), new
ShardingSphereIdentifier("t_order"))),
new
TableAndSchemaNameMapper(Collections.singletonMap("t_order",
expectedSchemaName)));
return new WALEventConverter(new
IncrementalDumperContext(commonContext, null, false), metaDataLoader);
}
diff --git
a/kernel/data-pipeline/dialect/postgresql/src/test/java/org/apache/shardingsphere/data/pipeline/postgresql/sqlbuilder/template/PostgreSQLPipelineFreemarkerManagerTest.java
b/kernel/data-pipeline/dialect/postgresql/src/test/java/org/apache/shardingsphere/data/pipeline/postgresql/sqlbuilder/template/PostgreSQLPipelineFreemarkerManagerTest.java
index a3606b2d552..7d8d53c34d8 100644
---
a/kernel/data-pipeline/dialect/postgresql/src/test/java/org/apache/shardingsphere/data/pipeline/postgresql/sqlbuilder/template/PostgreSQLPipelineFreemarkerManagerTest.java
+++
b/kernel/data-pipeline/dialect/postgresql/src/test/java/org/apache/shardingsphere/data/pipeline/postgresql/sqlbuilder/template/PostgreSQLPipelineFreemarkerManagerTest.java
@@ -48,6 +48,16 @@ class PostgreSQLPipelineFreemarkerManagerTest {
assertNotNull(PostgreSQLPipelineFreemarkerManager.getSQLByVersion(dataModel,
"component/table/%s/get_columns_for_table.ftl", 10, 0));
}
+ @Test
+ void assertGetTableIdSQLWithActualIdentifiers() {
+ Map<String, Object> dataModel = new HashMap<>(2, 1F);
+ dataModel.put("schemaName", "UPPER_SCHEMA");
+ dataModel.put("tableName", "UPPER_TABLE");
+ String actual =
PostgreSQLPipelineFreemarkerManager.getSQLByVersion(dataModel,
"component/table/%s/get_table_id.ftl", 12, 0);
+ String expected = "\nSELECT (pg_catalog.quote_ident('UPPER_SCHEMA') ||
'.' || pg_catalog.quote_ident('UPPER_TABLE'))::REGCLASS::OID AS tid;\n";
+ assertThat(actual, is(expected));
+ }
+
@Test
void assertCreateTableTemplateRendersNormalizedSequenceNumbers() {
Map<String, Object> dataModel = new LinkedHashMap<>(8, 1F);
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 578edf19537..846fe9ec3e3 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
@@ -22,6 +22,7 @@ import
org.apache.shardingsphere.data.pipeline.api.type.ShardingSpherePipelineDa
import
org.apache.shardingsphere.data.pipeline.core.context.PipelineContextManager;
import
org.apache.shardingsphere.data.pipeline.core.context.TransmissionProcessContext;
import org.apache.shardingsphere.data.pipeline.core.datanode.JobDataNodeEntry;
+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;
@@ -42,6 +43,9 @@ import
org.apache.shardingsphere.data.pipeline.scenario.migration.config.Migrati
import
org.apache.shardingsphere.data.pipeline.scenario.migration.context.MigrationJobItemContext;
import
org.apache.shardingsphere.data.pipeline.scenario.migration.ingest.dumper.MigrationIncrementalDumperContextCreator;
import
org.apache.shardingsphere.data.pipeline.scenario.migration.preparer.MigrationJobPreparer;
+import
org.apache.shardingsphere.database.connector.core.metadata.identifier.IdentifierCasePolicy;
+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.datanode.DataNode;
import
org.apache.shardingsphere.infra.datasource.pool.props.domain.DataSourcePoolProperties;
@@ -53,7 +57,9 @@ import
org.apache.shardingsphere.infra.yaml.config.pojo.rule.YamlRuleConfigurati
import java.sql.SQLException;
import java.util.Collection;
+import java.util.Collections;
import java.util.HashMap;
+import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Map.Entry;
import java.util.stream.Collectors;
@@ -70,7 +76,7 @@ public final class MigrationJobExecutorCallback implements
DistributedPipelineJo
final
TransmissionJobItemProgress jobItemProgress, final TransmissionProcessContext
jobProcessContext,
final
PipelineDataSourceManager dataSourceManager) {
transformDataSourceConfigurations(jobConfig);
- MigrationTaskConfiguration taskConfig =
buildTaskConfiguration(jobConfig, shardingItem,
jobProcessContext.getProcessConfiguration());
+ MigrationTaskConfiguration taskConfig =
buildTaskConfiguration(jobConfig, shardingItem,
jobProcessContext.getProcessConfiguration(), dataSourceManager);
return new MigrationJobItemContext(jobConfig, shardingItem,
jobItemProgress, jobProcessContext, taskConfig, dataSourceManager);
}
@@ -90,15 +96,33 @@ public final class MigrationJobExecutorCallback implements
DistributedPipelineJo
: entry.getValue();
}
- private MigrationTaskConfiguration buildTaskConfiguration(final
MigrationJobConfiguration jobConfig, final int jobShardingItem, final
PipelineProcessConfiguration processConfig) {
+ private MigrationTaskConfiguration buildTaskConfiguration(final
MigrationJobConfiguration jobConfig, final int jobShardingItem, final
PipelineProcessConfiguration processConfig,
+ final
PipelineDataSourceManager dataSourceManager) {
Map<ShardingSphereIdentifier, Collection<String>>
tableAndRequiredColumnsMap = getTableAndRequiredColumnsMap(jobConfig);
IncrementalDumperContext incrementalDumperContext = new
MigrationIncrementalDumperContextCreator(jobConfig).createDumperContext(jobConfig.getJobDataNodeLine(jobShardingItem));
- Collection<CreateTableConfiguration> createTableConfigs =
buildCreateTableConfigurations(jobConfig,
incrementalDumperContext.getCommonContext().getTableAndSchemaNameMapper());
+ PipelineDataSource targetDataSource =
dataSourceManager.getDataSource(jobConfig.getTarget());
+ IdentifierCasePolicy targetSchemaIdentifierPolicy =
IdentifierNormalizeEngine.resolvePolicy(jobConfig.getTargetDatabaseType(),
targetDataSource, IdentifierScope.SCHEMA);
+ IdentifierCasePolicy targetTableIdentifierPolicy =
IdentifierNormalizeEngine.resolvePolicy(jobConfig.getTargetDatabaseType(),
targetDataSource, IdentifierScope.TABLE);
+ Collection<CreateTableConfiguration> createTableConfigs =
buildCreateTableConfigurations(
+ jobConfig,
incrementalDumperContext.getCommonContext().getTableAndSchemaNameMapper(),
targetSchemaIdentifierPolicy, targetTableIdentifierPolicy);
ImporterConfiguration importerConfig = buildImporterConfiguration(
- jobConfig, processConfig, tableAndRequiredColumnsMap,
incrementalDumperContext.getCommonContext().getTableAndSchemaNameMapper());
+ jobConfig, processConfig, tableAndRequiredColumnsMap,
getTargetTableAndSchemaNameMapper(jobConfig, targetSchemaIdentifierPolicy,
targetTableIdentifierPolicy));
return new
MigrationTaskConfiguration(incrementalDumperContext.getCommonContext().getDataSourceName(),
createTableConfigs, incrementalDumperContext, importerConfig);
}
+ private TableAndSchemaNameMapper getTargetTableAndSchemaNameMapper(final
MigrationJobConfiguration jobConfig, final IdentifierCasePolicy
targetSchemaIdentifierPolicy,
+ final
IdentifierCasePolicy targetTableIdentifierPolicy) {
+ if (null == jobConfig.getTargetTableSchemaMap()) {
+ return new TableAndSchemaNameMapper(Collections.emptyMap());
+ }
+ Map<String, String> targetTableSchemaMap = new
LinkedHashMap<>(jobConfig.getTargetTableSchemaMap().size(), 1F);
+ for (Entry<String, String> entry :
jobConfig.getTargetTableSchemaMap().entrySet()) {
+ String schemaName = null == entry.getValue() ? null :
IdentifierNormalizeEngine.normalize(targetSchemaIdentifierPolicy,
entry.getValue());
+
targetTableSchemaMap.put(IdentifierNormalizeEngine.normalize(targetTableIdentifierPolicy,
entry.getKey()), schemaName);
+ }
+ return new TableAndSchemaNameMapper(targetTableSchemaMap);
+ }
+
@SuppressWarnings({"rawtypes", "unchecked"})
private Map<ShardingSphereIdentifier, Collection<String>>
getTableAndRequiredColumnsMap(final MigrationJobConfiguration jobConfig) {
Map<ShardingSphereIdentifier, Collection<String>> result = new
HashMap<>();
@@ -110,17 +134,25 @@ public final class MigrationJobExecutorCallback
implements DistributedPipelineJo
return result;
}
- private Collection<CreateTableConfiguration>
buildCreateTableConfigurations(final MigrationJobConfiguration jobConfig, final
TableAndSchemaNameMapper mapper) {
- return
jobConfig.getTablesFirstDataNodes().getEntries().stream().map(each ->
getCreateTableConfiguration(jobConfig, mapper,
each)).collect(Collectors.toList());
+ private Collection<CreateTableConfiguration>
buildCreateTableConfigurations(final MigrationJobConfiguration jobConfig, final
TableAndSchemaNameMapper mapper,
+
final IdentifierCasePolicy targetSchemaIdentifierPolicy,
+
final IdentifierCasePolicy targetTableIdentifierPolicy) {
+ return jobConfig.getTablesFirstDataNodes().getEntries().stream()
+ .map(each -> getCreateTableConfiguration(jobConfig, mapper,
each, targetSchemaIdentifierPolicy,
targetTableIdentifierPolicy)).collect(Collectors.toList());
}
- private CreateTableConfiguration getCreateTableConfiguration(final
MigrationJobConfiguration jobConfig, final TableAndSchemaNameMapper mapper,
final JobDataNodeEntry jobDataNodeEntry) {
+ private CreateTableConfiguration getCreateTableConfiguration(final
MigrationJobConfiguration jobConfig, final TableAndSchemaNameMapper mapper,
+ final
JobDataNodeEntry jobDataNodeEntry, final IdentifierCasePolicy
targetSchemaIdentifierPolicy,
+ final
IdentifierCasePolicy targetTableIdentifierPolicy) {
DataNode dataNode = jobDataNodeEntry.getDataNodes().get(0);
PipelineDataSourceConfiguration sourceDataSourceConfig =
jobConfig.getSources().get(dataNode.getDataSourceName());
String sourceSchemaName =
mapper.getSchemaName(jobDataNodeEntry.getLogicTableName());
- String targetSchemaName = new
DatabaseTypeRegistry(jobConfig.getTargetDatabaseType()).getDialectDatabaseMetaData().getSchemaOption().isSchemaAvailable()
? sourceSchemaName : null;
+ String targetSchemaName = null != sourceSchemaName && new
DatabaseTypeRegistry(jobConfig.getTargetDatabaseType()).getDialectDatabaseMetaData().getSchemaOption().isSchemaAvailable()
+ ?
IdentifierNormalizeEngine.normalize(targetSchemaIdentifierPolicy,
sourceSchemaName)
+ : null;
+ String targetTableName =
IdentifierNormalizeEngine.normalize(targetTableIdentifierPolicy,
jobDataNodeEntry.getLogicTableName());
return new CreateTableConfiguration(sourceDataSourceConfig, new
QualifiedTable(sourceSchemaName, dataNode.getTableName()),
- jobConfig.getTarget(), new QualifiedTable(targetSchemaName,
jobDataNodeEntry.getLogicTableName()));
+ jobConfig.getTarget(), new QualifiedTable(targetSchemaName,
targetTableName));
}
private ImporterConfiguration buildImporterConfiguration(final
MigrationJobConfiguration jobConfig, final PipelineProcessConfiguration
pipelineProcessConfig,
diff --git
a/kernel/data-pipeline/scenario/migration/core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/check/consistency/MigrationDataConsistencyChecker.java
b/kernel/data-pipeline/scenario/migration/core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/check/consistency/MigrationDataConsistencyChecker.java
index 27af07ec36e..28a4db3c937 100644
---
a/kernel/data-pipeline/scenario/migration/core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/check/consistency/MigrationDataConsistencyChecker.java
+++
b/kernel/data-pipeline/scenario/migration/core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/check/consistency/MigrationDataConsistencyChecker.java
@@ -50,6 +50,10 @@ import
org.apache.shardingsphere.data.pipeline.core.ratelimit.JobRateLimitAlgori
import
org.apache.shardingsphere.data.pipeline.core.util.PipelineDataSourceConfigurationUtils;
import
org.apache.shardingsphere.data.pipeline.scenario.migration.MigrationJobType;
import
org.apache.shardingsphere.data.pipeline.scenario.migration.config.MigrationJobConfiguration;
+import
org.apache.shardingsphere.database.connector.core.metadata.identifier.IdentifierCasePolicy;
+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.datanode.DataNode;
import
org.apache.shardingsphere.infra.datasource.pool.props.domain.DataSourcePoolProperties;
import org.apache.shardingsphere.infra.exception.ShardingSpherePreconditions;
@@ -174,11 +178,22 @@ public final class MigrationDataConsistencyChecker
implements PipelineDataConsis
String targetTableName = checkRangePosition.getLogicTableName();
List<String> columnNames = tableMetaData.getColumnNames();
List<PipelineColumnMetaData> uniqueKeys =
PipelineTableMetaDataUtils.getUniqueKeyColumns(sourceTable.getSchemaName(),
sourceTable.getTableName(), metaDataLoader);
- QualifiedTable targetTable = new
QualifiedTable(dataNode.getSchemaName(), targetTableName);
PipelineDataSource targetDataSource =
dataSourceManager.getDataSource(jobConfig.getTarget());
+ IdentifierCasePolicy targetSchemaIdentifierPolicy =
IdentifierNormalizeEngine.resolvePolicy(jobConfig.getTargetDatabaseType(),
targetDataSource, IdentifierScope.SCHEMA);
+ IdentifierCasePolicy targetTableIdentifierPolicy =
IdentifierNormalizeEngine.resolvePolicy(jobConfig.getTargetDatabaseType(),
targetDataSource, IdentifierScope.TABLE);
+ IdentifierCasePolicy targetColumnIdentifierPolicy =
IdentifierNormalizeEngine.resolvePolicy(jobConfig.getTargetDatabaseType(),
targetDataSource, IdentifierScope.COLUMN);
+ String targetSchemaName = null != dataNode.getSchemaName()
+ && new
DatabaseTypeRegistry(jobConfig.getTargetDatabaseType()).getDialectDatabaseMetaData().getSchemaOption().isSchemaAvailable()
+ ?
IdentifierNormalizeEngine.normalize(targetSchemaIdentifierPolicy,
dataNode.getSchemaName())
+ : null;
+ QualifiedTable targetTable = new QualifiedTable(targetSchemaName,
IdentifierNormalizeEngine.normalize(targetTableIdentifierPolicy,
targetTableName));
+ List<String> targetColumnNames = columnNames.stream().map(each ->
IdentifierNormalizeEngine.normalize(targetColumnIdentifierPolicy,
each)).collect(Collectors.toList());
+ List<PipelineColumnMetaData> targetUniqueKeys =
uniqueKeys.stream().map(each -> new
PipelineColumnMetaData(each.getOrdinalPosition(),
+
IdentifierNormalizeEngine.normalize(targetColumnIdentifierPolicy,
each.getName()), each.getDataType(), each.getDataTypeName(),
+ each.isNullable(), each.isPrimaryKey(),
each.isUniqueKey())).collect(Collectors.toList());
TableInventoryCheckParameter param = new TableInventoryCheckParameter(
- jobConfig.getJobId(), checkRangePosition.getSplittingItem(),
sourceDataSource, targetDataSource, sourceTable, targetTable, columnNames,
uniqueKeys,
- readRateLimitAlgorithm, progressContext,
checkRangePosition.getQueryCondition());
+ jobConfig.getJobId(), checkRangePosition.getSplittingItem(),
sourceDataSource, targetDataSource, sourceTable, targetTable,
+ columnNames, uniqueKeys, targetColumnNames, targetUniqueKeys,
readRateLimitAlgorithm, progressContext,
checkRangePosition.getQueryCondition());
TableInventoryChecker tableInventoryChecker =
tableChecker.buildTableInventoryChecker(param);
currentTableInventoryChecker.set(tableInventoryChecker);
Optional<TableDataConsistencyCheckResult> preCheckResult =
tableInventoryChecker.preCheck();
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 1d826b5cdd4..17fd896d23c 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
@@ -25,6 +25,7 @@ import
org.apache.shardingsphere.data.pipeline.core.context.PipelineContextManag
import
org.apache.shardingsphere.data.pipeline.core.context.TransmissionProcessContext;
import org.apache.shardingsphere.data.pipeline.core.datanode.JobDataNodeEntry;
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.job.api.PipelineAPIFactory;
@@ -37,16 +38,22 @@ import
org.apache.shardingsphere.data.pipeline.core.registrycenter.repository.Pi
import
org.apache.shardingsphere.data.pipeline.scenario.migration.config.MigrationJobConfiguration;
import
org.apache.shardingsphere.data.pipeline.scenario.migration.config.MigrationTaskConfiguration;
import
org.apache.shardingsphere.data.pipeline.scenario.migration.context.MigrationJobItemContext;
+import
org.apache.shardingsphere.database.connector.core.metadata.identifier.IdentifierCasePolicyFactory;
+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.DatabaseType;
import
org.apache.shardingsphere.database.connector.core.type.DatabaseTypeFactory;
+import
org.apache.shardingsphere.database.connector.core.type.DatabaseTypeRegistry;
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.spi.type.typed.TypedSPILoader;
import org.apache.shardingsphere.infra.yaml.config.pojo.YamlRootConfiguration;
import org.apache.shardingsphere.mode.manager.ContextManager;
import org.junit.jupiter.api.Test;
+import org.mockito.MockedConstruction;
import org.mockito.MockedStatic;
import java.util.Collection;
@@ -58,9 +65,15 @@ import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.is;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.isNull;
+import static org.mockito.ArgumentMatchers.same;
+import static org.mockito.Mockito.CALLS_REAL_METHODS;
import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockConstruction;
import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.withSettings;
import static org.mockito.Mockito.when;
class MigrationJobExecutorCallbackTest {
@@ -122,6 +135,52 @@ class MigrationJobExecutorCallbackTest {
}
}
+ @Test
+ void assertBuildJobItemContextWithPostgreSQLTargetIdentifiers() {
+ try (
+ MockedConstruction<DatabaseTypeRegistry> ignored =
mockConstruction(DatabaseTypeRegistry.class,
withSettings().defaultAnswer(RETURNS_DEEP_STUBS),
+ (mock, context) ->
when(mock.getDialectDatabaseMetaData().getSchemaOption().isSchemaAvailable()).thenReturn(true));
+ MockedStatic<DatabaseTypeFactory> databaseTypeFactory =
mockStatic(DatabaseTypeFactory.class);
+ MockedStatic<PipelineAPIFactory> pipelineAPIFactory =
mockStatic(PipelineAPIFactory.class);
+ MockedStatic<PipelineContextManager> pipelineContextManager =
mockStatic(PipelineContextManager.class)) {
+ mockDatabaseTypeFactory(databaseTypeFactory);
+ mockGovernanceFacade(pipelineAPIFactory,
createSourceDataSourceYaml(10, 20));
+ mockProxyContext(pipelineContextManager);
+ MigrationJobItemContext actual = new
MigrationJobExecutorCallback().buildJobItemContext(
+ createPostgreSQLJobConfiguration(), 0, null,
mockProcessContext(), mock(PipelineDataSourceManager.class));
+ CreateTableConfiguration actualCreateTableConfig =
actual.getTaskConfig().getCreateTableConfigurations().iterator().next();
+ assertThat(actualCreateTableConfig.getSourceName(), is(new
QualifiedTable("UPPER_SCHEMA", "UPPER_TABLE")));
+ assertThat(actualCreateTableConfig.getTargetName(), is(new
QualifiedTable("upper_schema", "t_order")));
+
assertThat(actual.getTaskConfig().getImporterConfig().getTableAndSchemaNameMapper().getQualifiedTables(),
+ is(Collections.singletonList(new
QualifiedTable("upper_schema", "t_order"))));
+ }
+ }
+
+ @Test
+ void assertBuildJobItemContextWithCaseSensitiveMySQLTargetIdentifier() {
+ PipelineDataSource targetDataSource = mock(PipelineDataSource.class);
+ PipelineDataSourceManager dataSourceManager =
mock(PipelineDataSourceManager.class);
+
when(dataSourceManager.getDataSource(any())).thenReturn(targetDataSource);
+ try (
+ MockedStatic<DatabaseTypeFactory> databaseTypeFactory =
mockStatic(DatabaseTypeFactory.class);
+ MockedStatic<PipelineAPIFactory> pipelineAPIFactory =
mockStatic(PipelineAPIFactory.class);
+ MockedStatic<PipelineContextManager> pipelineContextManager =
mockStatic(PipelineContextManager.class);
+ MockedStatic<IdentifierNormalizeEngine> normalizeEngine =
mockStatic(IdentifierNormalizeEngine.class,
withSettings().defaultAnswer(CALLS_REAL_METHODS))) {
+ mockDatabaseTypeFactory(databaseTypeFactory);
+ mockGovernanceFacade(pipelineAPIFactory,
createSourceDataSourceYaml(10, 20));
+ mockProxyContext(pipelineContextManager);
+ normalizeEngine.when(() ->
IdentifierNormalizeEngine.resolvePolicy(any(DatabaseType.class), isNull(),
eq(IdentifierScope.TABLE)))
+
.thenReturn(IdentifierCasePolicyFactory.newLowerCaseInsensitivePolicySet().getPolicy(IdentifierScope.TABLE));
+ normalizeEngine.when(() ->
IdentifierNormalizeEngine.resolvePolicy(any(DatabaseType.class),
same(targetDataSource), eq(IdentifierScope.TABLE)))
+
.thenReturn(IdentifierCasePolicyFactory.newSensitivePolicySet().getPolicy(IdentifierScope.TABLE));
+ MigrationJobItemContext actual = new
MigrationJobExecutorCallback().buildJobItemContext(
+ createMySQLJobConfiguration(), 0, null,
mockProcessContext(), dataSourceManager);
+ CreateTableConfiguration actualCreateTableConfig =
actual.getTaskConfig().getCreateTableConfigurations().iterator().next();
+ assertThat(actualCreateTableConfig.getTargetName(), is(new
QualifiedTable(null, "T_ORDER")));
+ normalizeEngine.verify(() ->
IdentifierNormalizeEngine.resolvePolicy(any(DatabaseType.class),
same(targetDataSource), eq(IdentifierScope.TABLE)));
+ }
+ }
+
private void mockDatabaseTypeFactory(final
MockedStatic<DatabaseTypeFactory> databaseTypeFactory) {
databaseTypeFactory.when(() ->
DatabaseTypeFactory.get(anyString())).thenReturn(databaseType);
}
@@ -164,11 +223,35 @@ class MigrationJobExecutorCallbackTest {
sources.put(SOURCE_DATA_SOURCE_NAME, new
StandardPipelineDataSourceConfiguration(createDataSourceProperties(SOURCE_JDBC_URL,
2, 3)));
return new MigrationJobConfiguration(createJobId(), DATABASE_NAME,
databaseType, databaseType, sources,
new
ShardingSpherePipelineDataSourceConfiguration(createRootConfiguration(4, 5)),
Collections.singletonList("t_order"),
- Collections.singletonMap("t_order", "foo_schema"),
createJobDataNodeLine(), Collections.singletonList(createJobDataNodeLine()), 1,
3);
+ null, createJobDataNodeLine(),
Collections.singletonList(createJobDataNodeLine()), 1, 3);
+ }
+
+ private MigrationJobConfiguration createPostgreSQLJobConfiguration() {
+ JobDataNodeLine jobDataNodeLine = new
JobDataNodeLine(Collections.singletonList(
+ new JobDataNodeEntry("T_ORDER", Collections.singletonList(new
DataNode(SOURCE_DATA_SOURCE_NAME, "UPPER_SCHEMA", "UPPER_TABLE")))));
+ Map<String, PipelineDataSourceConfiguration> sources = new
LinkedHashMap<>(1, 1F);
+ sources.put(SOURCE_DATA_SOURCE_NAME, new
StandardPipelineDataSourceConfiguration(createDataSourceProperties(SOURCE_JDBC_URL,
2, 3)));
+ return new MigrationJobConfiguration(createJobId(jobDataNodeLine),
DATABASE_NAME, databaseType, TypedSPILoader.getService(DatabaseType.class,
"PostgreSQL"), sources,
+ new
ShardingSpherePipelineDataSourceConfiguration(createRootConfiguration(4, 5)),
Collections.singletonList("T_ORDER"),
+ Collections.singletonMap("T_ORDER", "UPPER_SCHEMA"),
jobDataNodeLine, Collections.singletonList(jobDataNodeLine), 1, 3);
+ }
+
+ private MigrationJobConfiguration createMySQLJobConfiguration() {
+ JobDataNodeLine jobDataNodeLine = new
JobDataNodeLine(Collections.singletonList(
+ new JobDataNodeEntry("T_ORDER", Collections.singletonList(new
DataNode(SOURCE_DATA_SOURCE_NAME + ".UPPER_TABLE")))));
+ Map<String, PipelineDataSourceConfiguration> sources = new
LinkedHashMap<>(1, 1F);
+ sources.put(SOURCE_DATA_SOURCE_NAME, new
StandardPipelineDataSourceConfiguration(createDataSourceProperties(SOURCE_JDBC_URL,
2, 3)));
+ return new MigrationJobConfiguration(createJobId(jobDataNodeLine),
DATABASE_NAME, databaseType, TypedSPILoader.getService(DatabaseType.class,
"MySQL"), sources,
+ new
ShardingSpherePipelineDataSourceConfiguration(createRootConfiguration(4, 5)),
Collections.singletonList("T_ORDER"),
+ Collections.singletonMap("T_ORDER", null), jobDataNodeLine,
Collections.singletonList(jobDataNodeLine), 1, 3);
}
private String createJobId() {
- return PipelineJobIdUtils.marshal(new MigrationJobId(new
PipelineContextKey(InstanceType.PROXY),
Collections.singletonList(createJobDataNodeLine().marshal())));
+ return createJobId(createJobDataNodeLine());
+ }
+
+ private String createJobId(final JobDataNodeLine jobDataNodeLine) {
+ return PipelineJobIdUtils.marshal(new MigrationJobId(new
PipelineContextKey(InstanceType.PROXY),
Collections.singletonList(jobDataNodeLine.marshal())));
}
private JobDataNodeLine createJobDataNodeLine() {
diff --git
a/kernel/data-pipeline/scenario/migration/distsql/handler/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/distsql/handler/update/MigrateTableExecutor.java
b/kernel/data-pipeline/scenario/migration/distsql/handler/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/distsql/handler/update/MigrateTableExecutor.java
index c78dd42c3a3..4ff7dfc3ee6 100644
---
a/kernel/data-pipeline/scenario/migration/distsql/handler/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/distsql/handler/update/MigrateTableExecutor.java
+++
b/kernel/data-pipeline/scenario/migration/distsql/handler/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/distsql/handler/update/MigrateTableExecutor.java
@@ -42,6 +42,7 @@ import
org.apache.shardingsphere.infra.metadata.database.ShardingSphereDatabase;
import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
import
org.apache.shardingsphere.infra.yaml.config.swapper.resource.YamlDataSourceConfigurationSwapper;
import org.apache.shardingsphere.mode.manager.ContextManager;
+import
org.apache.shardingsphere.sql.parser.statement.core.value.identifier.IdentifierValue;
import java.util.Collection;
import java.util.LinkedList;
@@ -70,26 +71,33 @@ public final class MigrateTableExecutor implements
DistSQLUpdateExecutor<Migrate
private Collection<MigrationSourceTargetEntry>
getMigrationSourceTargetEntries(final PipelineContextKey contextKey, final
MigrateTableStatement sqlStatement) {
Collection<MigrationSourceTargetEntry> result = new LinkedList<>();
for (MigrationSourceTargetSegment each :
sqlStatement.getSourceTargetEntries()) {
- String schemaName = getSchemaName(contextKey,
each.getSourceDatabaseName(), each.getSourceSchemaName()).orElse(null);
- result.add(new MigrationSourceTargetEntry(new
DataNode(each.getSourceDatabaseName(), schemaName, each.getSourceTableName()),
each.getTargetTableName()));
+ result.add(new
MigrationSourceTargetEntry(getSourceDataNode(contextKey, each),
each.getTargetTableName()));
}
return result;
}
- private Optional<String> getSchemaName(final PipelineContextKey
contextKey, final String sourceDatabaseName, final String sourceSchemaName) {
+ private DataNode getSourceDataNode(final PipelineContextKey contextKey,
final MigrationSourceTargetSegment segment) {
Map<String, DataSourcePoolProperties> metaDataDataSource = new
PipelineDataSourcePersistService().load(contextKey, "MIGRATION");
- DataSourcePoolProperties sourceDataSourcePoolProps =
metaDataDataSource.get(sourceDatabaseName);
+ DataSourcePoolProperties sourceDataSourcePoolProps =
metaDataDataSource.get(segment.getSourceDatabaseName());
if (null == sourceDataSourcePoolProps) {
- return Optional.ofNullable(sourceSchemaName);
+ return new DataNode(segment.getSourceDatabaseName(),
segment.getSourceSchemaName(), segment.getSourceTableName());
}
Map<String, Object> sourceDataSourceProps = new
YamlDataSourceConfigurationSwapper().swapToMap(sourceDataSourcePoolProps);
StandardPipelineDataSourceConfiguration sourceDataSourceConfig = new
StandardPipelineDataSourceConfiguration(sourceDataSourceProps);
if (!new
DatabaseTypeRegistry(sourceDataSourceConfig.getDatabaseType()).getDialectDatabaseMetaData().getSchemaOption().isSchemaAvailable())
{
- return Optional.ofNullable(sourceSchemaName);
+ return new DataNode(segment.getSourceDatabaseName(),
segment.getSourceSchemaName(), segment.getSourceTableName());
}
- return Optional.ofNullable(null == sourceSchemaName
+ String schemaName = getSchemaName(sourceDataSourceConfig,
segment.getSourceSchemaIdentifier()).orElse(null);
+ String tableName =
IdentifierNormalizeEngine.normalize(IdentifierNormalizeEngine.resolvePolicy(sourceDataSourceConfig.getDatabaseType(),
null, IdentifierScope.TABLE),
+
segment.getSourceTableIdentifier().getValueWithQuoteCharacters());
+ return new DataNode(segment.getSourceDatabaseName(), schemaName,
tableName);
+ }
+
+ private Optional<String> getSchemaName(final
StandardPipelineDataSourceConfiguration sourceDataSourceConfig, final
IdentifierValue sourceSchemaIdentifier) {
+ return Optional.ofNullable(null == sourceSchemaIdentifier
? PipelineSchemaUtils.getDefaultSchema(sourceDataSourceConfig)
- :
IdentifierNormalizeEngine.normalize(IdentifierNormalizeEngine.resolvePolicy(sourceDataSourceConfig.getDatabaseType(),
null, IdentifierScope.SCHEMA), sourceSchemaName));
+ :
IdentifierNormalizeEngine.normalize(IdentifierNormalizeEngine.resolvePolicy(sourceDataSourceConfig.getDatabaseType(),
null, IdentifierScope.SCHEMA),
+ sourceSchemaIdentifier.getValueWithQuoteCharacters()));
}
@Override
diff --git
a/kernel/data-pipeline/scenario/migration/distsql/handler/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/migration/distsql/handler/update/MigrateTableExecutorTest.java
b/kernel/data-pipeline/scenario/migration/distsql/handler/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/migration/distsql/handler/update/MigrateTableExecutorTest.java
index 6f4184fea51..fac1bab0117 100644
---
a/kernel/data-pipeline/scenario/migration/distsql/handler/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/migration/distsql/handler/update/MigrateTableExecutorTest.java
+++
b/kernel/data-pipeline/scenario/migration/distsql/handler/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/migration/distsql/handler/update/MigrateTableExecutorTest.java
@@ -27,6 +27,7 @@ import
org.apache.shardingsphere.data.pipeline.scenario.migration.api.MigrationS
import
org.apache.shardingsphere.data.pipeline.scenario.migration.distsql.segment.MigrationSourceTargetSegment;
import
org.apache.shardingsphere.data.pipeline.scenario.migration.distsql.statement.updatable.MigrateTableStatement;
import
org.apache.shardingsphere.database.connector.core.metadata.database.metadata.DialectDatabaseMetaData;
+import
org.apache.shardingsphere.database.connector.core.metadata.database.enums.QuoteCharacter;
import
org.apache.shardingsphere.database.connector.core.metadata.identifier.IdentifierCasePolicy;
import
org.apache.shardingsphere.database.connector.core.metadata.identifier.IdentifierCasePolicyFactory;
import
org.apache.shardingsphere.database.connector.core.metadata.identifier.IdentifierNormalizeEngine;
@@ -38,6 +39,7 @@ import
org.apache.shardingsphere.infra.metadata.database.ShardingSphereDatabase;
import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
import
org.apache.shardingsphere.infra.yaml.config.swapper.resource.YamlDataSourceConfigurationSwapper;
import org.apache.shardingsphere.mode.manager.ContextManager;
+import
org.apache.shardingsphere.sql.parser.statement.core.value.identifier.IdentifierValue;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
@@ -71,7 +73,7 @@ import static org.mockito.Mockito.when;
@ExtendWith(MockitoExtension.class)
class MigrateTableExecutorTest {
- private static final String SOURCE_DATABASE_NAME = "foo_source";
+ private static final String SOURCE_DATABASE_NAME = "FOO_SOURCE";
private static final String TARGET_DATABASE_NAME = "foo_target";
@@ -140,16 +142,20 @@ class MigrateTableExecutorTest {
schemaUtils.when(() ->
PipelineSchemaUtils.getDefaultSchema(any(StandardPipelineDataSourceConfiguration.class))).thenReturn("public");
executor.executeUpdate(createStatement(null), contextManager);
verify(jobAPI).schedule(any(PipelineContextKey.class),
entriesCaptor.capture(), eq(TARGET_DATABASE_NAME));
-
assertNull(entriesCaptor.getValue().iterator().next().getSource().getSchemaName());
+ MigrationSourceTargetEntry actual =
entriesCaptor.getValue().iterator().next();
+ assertNull(actual.getSource().getSchemaName());
+ assertThat(actual.getSource().getTableName(), is("FOO_TBL"));
schemaUtils.verify(() ->
PipelineSchemaUtils.getDefaultSchema(any(StandardPipelineDataSourceConfiguration.class)),
never());
verify(database, never()).getProtocolType();
}
}
@ParameterizedTest(name = "{0}")
- @MethodSource("provideExplicitSchemaNames")
- void assertExecuteUpdateCanonicalizesExplicitSchema(final String name,
final String sourceSchemaName, final String expectedSchemaName) {
- IdentifierCasePolicy identifierCasePolicy =
IdentifierCasePolicyFactory.newLowerCasePolicySet().getPolicy(IdentifierScope.SCHEMA);
+ @MethodSource("provideExplicitSourceIdentifiers")
+ void assertExecuteUpdateCanonicalizesExplicitSourceIdentifiers(final
String name, final String sourceSchemaName, final String expectedSchemaName,
+ final
IdentifierValue sourceTableIdentifier, final String expectedTableName) {
+ IdentifierCasePolicy schemaIdentifierCasePolicy =
IdentifierCasePolicyFactory.newLowerCasePolicySet().getPolicy(IdentifierScope.SCHEMA);
+ IdentifierCasePolicy tableIdentifierCasePolicy =
IdentifierCasePolicyFactory.newLowerCasePolicySet().getPolicy(IdentifierScope.TABLE);
try (
MockedConstruction<PipelineDataSourcePersistService> ignored =
mockConstruction(PipelineDataSourcePersistService.class,
(mock, context) ->
when(mock.load(any(PipelineContextKey.class),
eq("MIGRATION"))).thenReturn(Collections.singletonMap(SOURCE_DATABASE_NAME,
sourceProps)));
@@ -162,20 +168,26 @@ class MigrateTableExecutorTest {
MockedStatic<PipelineSchemaUtils> schemaUtils =
mockStatic(PipelineSchemaUtils.class);
MockedStatic<IdentifierNormalizeEngine> normalizeEngine =
mockStatic(IdentifierNormalizeEngine.class, Answers.CALLS_REAL_METHODS)) {
spiLoader.when(() ->
TypedSPILoader.getService(TransmissionJobAPI.class,
"MIGRATION")).thenReturn(jobAPI);
- normalizeEngine.when(() ->
IdentifierNormalizeEngine.resolvePolicy(schemaCapableDatabaseType, null,
IdentifierScope.SCHEMA)).thenReturn(identifierCasePolicy);
- executor.executeUpdate(createStatement(sourceSchemaName),
contextManager);
+ normalizeEngine.when(() ->
IdentifierNormalizeEngine.resolvePolicy(schemaCapableDatabaseType, null,
IdentifierScope.SCHEMA)).thenReturn(schemaIdentifierCasePolicy);
+ normalizeEngine.when(() ->
IdentifierNormalizeEngine.resolvePolicy(schemaCapableDatabaseType, null,
IdentifierScope.TABLE)).thenReturn(tableIdentifierCasePolicy);
+ executor.executeUpdate(createStatement(sourceSchemaName,
sourceTableIdentifier), contextManager);
verify(jobAPI).schedule(any(PipelineContextKey.class),
entriesCaptor.capture(), eq(TARGET_DATABASE_NAME));
-
assertThat(entriesCaptor.getValue().iterator().next().getSource().getSchemaName(),
is(expectedSchemaName));
+ MigrationSourceTargetEntry actual =
entriesCaptor.getValue().iterator().next();
+ assertThat(actual.getSource().getDataSourceName(),
is(SOURCE_DATABASE_NAME));
+ assertThat(actual.getSource().getSchemaName(),
is(expectedSchemaName));
+ assertThat(actual.getSource().getTableName(),
is(expectedTableName));
+ assertThat(actual.getTargetTableName(), is("foo_tbl"));
schemaUtils.verify(() ->
PipelineSchemaUtils.getDefaultSchema(any(StandardPipelineDataSourceConfiguration.class)),
never());
}
}
- private static Stream<Arguments> provideExplicitSchemaNames() {
+ private static Stream<Arguments> provideExplicitSourceIdentifiers() {
return Stream.of(
- Arguments.of("lower-case schema", "foo_schema", "foo_schema"),
- Arguments.of("unquoted upper-case schema", "FOO_SCHEMA",
"foo_schema"),
- Arguments.of("quoted upper-case schema", "\"FOO_SCHEMA\"",
"FOO_SCHEMA"),
- Arguments.of("quoted mixed-case schema", "\"CaseSchema\"",
"CaseSchema"));
+ Arguments.of("lower-case schema", "foo_schema", "foo_schema",
new IdentifierValue("FOO_TBL", QuoteCharacter.BACK_QUOTE), "FOO_TBL"),
+ Arguments.of("unquoted upper-case schema", "FOO_SCHEMA",
"foo_schema", new IdentifierValue("FOO_TBL", QuoteCharacter.BACK_QUOTE),
"FOO_TBL"),
+ Arguments.of("quoted upper-case schema", "\"FOO_SCHEMA\"",
"FOO_SCHEMA", new IdentifierValue("FOO_TBL", QuoteCharacter.BACK_QUOTE),
"FOO_TBL"),
+ Arguments.of("quoted mixed-case schema", "\"CaseSchema\"",
"CaseSchema", new IdentifierValue("FOO_TBL", QuoteCharacter.BACK_QUOTE),
"FOO_TBL"),
+ Arguments.of("unquoted upper-case table", "foo_schema",
"foo_schema", new IdentifierValue("FOO_TBL", QuoteCharacter.NONE), "foo_tbl"));
}
@Test
@@ -195,7 +207,13 @@ class MigrateTableExecutorTest {
}
private MigrateTableStatement createStatement(final String
sourceSchemaName) {
- MigrationSourceTargetSegment entry = new
MigrationSourceTargetSegment(SOURCE_DATABASE_NAME, sourceSchemaName, "foo_tbl",
"foo_tbl");
+ return createStatement(sourceSchemaName, new
IdentifierValue("FOO_TBL", QuoteCharacter.BACK_QUOTE));
+ }
+
+ private MigrateTableStatement createStatement(final String
sourceSchemaName, final IdentifierValue sourceTableIdentifier) {
+ MigrationSourceTargetSegment entry = new
MigrationSourceTargetSegment(new IdentifierValue(SOURCE_DATABASE_NAME,
QuoteCharacter.BACK_QUOTE),
+ null == sourceSchemaName ? null : new
IdentifierValue(sourceSchemaName), sourceTableIdentifier,
+ new IdentifierValue("foo_tbl", QuoteCharacter.NONE));
return new MigrateTableStatement(null, Collections.singleton(entry));
}
diff --git
a/kernel/data-pipeline/scenario/migration/distsql/parser/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/distsql/parser/core/MigrationDistSQLStatementVisitor.java
b/kernel/data-pipeline/scenario/migration/distsql/parser/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/distsql/parser/core/MigrationDistSQLStatementVisitor.java
index ae2f0e12b4b..d93db27b26c 100644
---
a/kernel/data-pipeline/scenario/migration/distsql/parser/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/distsql/parser/core/MigrationDistSQLStatementVisitor.java
+++
b/kernel/data-pipeline/scenario/migration/distsql/parser/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/distsql/parser/core/MigrationDistSQLStatementVisitor.java
@@ -17,7 +17,6 @@
package
org.apache.shardingsphere.data.pipeline.scenario.migration.distsql.parser.core;
-import com.google.common.base.Splitter;
import org.antlr.v4.runtime.tree.ParseTree;
import
org.apache.shardingsphere.data.pipeline.distsql.statement.updatable.AlterTransmissionRuleStatement;
import
org.apache.shardingsphere.data.pipeline.scenario.migration.distsql.segment.MigrationSourceTargetSegment;
@@ -87,7 +86,6 @@ import
org.apache.shardingsphere.sql.parser.statement.core.value.identifier.Iden
import java.util.Collection;
import java.util.Collections;
import java.util.LinkedList;
-import java.util.List;
import java.util.Properties;
import java.util.stream.Collectors;
@@ -165,18 +163,15 @@ public final class MigrationDistSQLStatementVisitor
extends MigrationDistSQLStat
}
private String getTargetDatabaseName(final TargetTableNameContext
targetContext) {
- List<String> targetDatabaseNames =
Splitter.on('.').splitToList(getRequiredIdentifierValue(targetContext));
- return targetDatabaseNames.size() > 1 ? targetDatabaseNames.get(0) :
null;
+ return null == targetContext.owner() ? null :
getRequiredIdentifierValue(targetContext.owner());
}
private MigrationSourceTargetSegment getMigrationSourceTargetSegment(final
SourceTableNameContext sourceContext, final TargetTableNameContext
targetContext) {
- List<String> source =
Splitter.on('.').splitToList(getRequiredIdentifierValue(sourceContext));
- List<String> target =
Splitter.on('.').splitToList(getRequiredIdentifierValue(targetContext));
- String sourceDatabaseName = source.get(0);
- String sourceSchemaName = 3 == source.size() ? source.get(1) : null;
- String sourceTableName = source.get(source.size() - 1);
- String targetTableName = target.get(target.size() - 1);
- return new MigrationSourceTargetSegment(sourceDatabaseName,
sourceSchemaName, sourceTableName, targetTableName);
+ IdentifierValue sourceDatabaseIdentifier = new
IdentifierValue(sourceContext.owner().getText());
+ IdentifierValue sourceSchemaIdentifier = null ==
sourceContext.schema() ? null : new
IdentifierValue(sourceContext.schema().getText());
+ IdentifierValue sourceTableIdentifier = new
IdentifierValue(sourceContext.name().getText());
+ IdentifierValue targetTableIdentifier = new
IdentifierValue(targetContext.name().getText());
+ return new MigrationSourceTargetSegment(sourceDatabaseIdentifier,
sourceSchemaIdentifier, sourceTableIdentifier, targetTableIdentifier);
}
private String getRequiredIdentifierValue(final ParseTree context) {
diff --git
a/kernel/data-pipeline/scenario/migration/distsql/parser/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/migration/distsql/parser/core/MigrationDistSQLStatementVisitorTest.java
b/kernel/data-pipeline/scenario/migration/distsql/parser/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/migration/distsql/parser/core/MigrationDistSQLStatementVisitorTest.java
new file mode 100644
index 00000000000..d729fc44865
--- /dev/null
+++
b/kernel/data-pipeline/scenario/migration/distsql/parser/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/migration/distsql/parser/core/MigrationDistSQLStatementVisitorTest.java
@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package
org.apache.shardingsphere.data.pipeline.scenario.migration.distsql.parser.core;
+
+import org.antlr.v4.runtime.CharStreams;
+import org.antlr.v4.runtime.CommonTokenStream;
+import
org.apache.shardingsphere.data.pipeline.scenario.migration.distsql.segment.MigrationSourceTargetSegment;
+import
org.apache.shardingsphere.data.pipeline.scenario.migration.distsql.statement.updatable.MigrateTableStatement;
+import
org.apache.shardingsphere.database.connector.core.metadata.database.enums.QuoteCharacter;
+import org.apache.shardingsphere.sql.parser.engine.core.ParseASTNode;
+import
org.apache.shardingsphere.sql.parser.statement.core.value.identifier.IdentifierValue;
+import org.junit.jupiter.api.Test;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+
+class MigrationDistSQLStatementVisitorTest {
+
+ @Test
+ void assertVisitMigrateTablePreservesQualifiedIdentifiers() {
+ MigrationDistSQLLexer lexer = new
MigrationDistSQLLexer(CharStreams.fromString(
+ "MIGRATE TABLE `FOO_DS`.`FOO_SCHEMA`.`FOO_TABLE` INTO
target_db.FOO_TABLE"));
+ MigrationDistSQLParser parser = new MigrationDistSQLParser(new
CommonTokenStream(lexer));
+ ParseASTNode parseASTNode = (ParseASTNode) parser.parse();
+ MigrateTableStatement statement = (MigrateTableStatement) new
MigrationDistSQLStatementVisitor().visit(parseASTNode.getRootNode());
+ MigrationSourceTargetSegment actual =
statement.getSourceTargetEntries().iterator().next();
+ assertThat(actual.getSourceDatabaseIdentifier(), is(new
IdentifierValue("FOO_DS", QuoteCharacter.BACK_QUOTE)));
+ assertThat(actual.getSourceSchemaIdentifier(), is(new
IdentifierValue("FOO_SCHEMA", QuoteCharacter.BACK_QUOTE)));
+ assertThat(actual.getSourceTableIdentifier(), is(new
IdentifierValue("FOO_TABLE", QuoteCharacter.BACK_QUOTE)));
+ assertThat(actual.getTargetTableIdentifier(), is(new
IdentifierValue("FOO_TABLE", QuoteCharacter.NONE)));
+ assertThat(statement.getTargetDatabaseName(), is("target_db"));
+ }
+}
diff --git
a/kernel/data-pipeline/scenario/migration/distsql/statement/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/distsql/segment/MigrationSourceTargetSegment.java
b/kernel/data-pipeline/scenario/migration/distsql/statement/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/distsql/segment/MigrationSourceTargetSegment.java
index 91fd3cb0ad1..6015f90914b 100644
---
a/kernel/data-pipeline/scenario/migration/distsql/statement/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/distsql/segment/MigrationSourceTargetSegment.java
+++
b/kernel/data-pipeline/scenario/migration/distsql/statement/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/distsql/segment/MigrationSourceTargetSegment.java
@@ -20,6 +20,7 @@ package
org.apache.shardingsphere.data.pipeline.scenario.migration.distsql.segme
import lombok.Getter;
import lombok.RequiredArgsConstructor;
import org.apache.shardingsphere.distsql.segment.DistSQLSegment;
+import
org.apache.shardingsphere.sql.parser.statement.core.value.identifier.IdentifierValue;
/**
* Migration source target segment.
@@ -28,11 +29,60 @@ import
org.apache.shardingsphere.distsql.segment.DistSQLSegment;
@Getter
public final class MigrationSourceTargetSegment implements DistSQLSegment {
- private final String sourceDatabaseName;
+ private final IdentifierValue sourceDatabaseIdentifier;
- private final String sourceSchemaName;
+ private final IdentifierValue sourceSchemaIdentifier;
- private final String sourceTableName;
+ private final IdentifierValue sourceTableIdentifier;
- private final String targetTableName;
+ private final IdentifierValue targetTableIdentifier;
+
+ /**
+ * Construct a migration source target segment from textual identifiers.
+ *
+ * @param sourceDatabaseName source database name
+ * @param sourceSchemaName source schema name
+ * @param sourceTableName source table name
+ * @param targetTableName target table name
+ */
+ public MigrationSourceTargetSegment(final String sourceDatabaseName, final
String sourceSchemaName, final String sourceTableName, final String
targetTableName) {
+ this(new IdentifierValue(sourceDatabaseName), null == sourceSchemaName
? null : new IdentifierValue(sourceSchemaName),
+ new IdentifierValue(sourceTableName), new
IdentifierValue(targetTableName));
+ }
+
+ /**
+ * Get source database name.
+ *
+ * @return source database name
+ */
+ public String getSourceDatabaseName() {
+ return sourceDatabaseIdentifier.getValue();
+ }
+
+ /**
+ * Get source schema name.
+ *
+ * @return source schema name
+ */
+ public String getSourceSchemaName() {
+ return null == sourceSchemaIdentifier ? null :
sourceSchemaIdentifier.getValue();
+ }
+
+ /**
+ * Get source table name.
+ *
+ * @return source table name
+ */
+ public String getSourceTableName() {
+ return sourceTableIdentifier.getValue();
+ }
+
+ /**
+ * Get target table name.
+ *
+ * @return target table name
+ */
+ public String getTargetTableName() {
+ return targetTableIdentifier.getValue();
+ }
}
diff --git
a/test/e2e/operation/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/operation/pipeline/cases/migration/general/MySQLMigrationIdentifierCaseE2EIT.java
b/test/e2e/operation/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/operation/pipeline/cases/migration/general/MySQLMigrationIdentifierCaseE2EIT.java
new file mode 100644
index 00000000000..c4dc39e327f
--- /dev/null
+++
b/test/e2e/operation/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/operation/pipeline/cases/migration/general/MySQLMigrationIdentifierCaseE2EIT.java
@@ -0,0 +1,114 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package
org.apache.shardingsphere.test.e2e.operation.pipeline.cases.migration.general;
+
+import com.google.common.collect.ImmutableMap;
+import
org.apache.shardingsphere.data.pipeline.scenario.migration.MigrationJobType;
+import
org.apache.shardingsphere.test.e2e.operation.pipeline.cases.PipelineContainerComposer;
+import
org.apache.shardingsphere.test.e2e.operation.pipeline.cases.migration.AbstractMigrationE2EIT;
+import
org.apache.shardingsphere.test.e2e.operation.pipeline.framework.param.PipelineE2ECondition;
+import
org.apache.shardingsphere.test.e2e.operation.pipeline.framework.param.PipelineE2ESettings;
+import
org.apache.shardingsphere.test.e2e.operation.pipeline.framework.param.PipelineE2ESettings.PipelineE2EDatabaseSettings;
+import
org.apache.shardingsphere.test.e2e.operation.pipeline.framework.param.PipelineE2ETestCaseArgumentsProvider;
+import
org.apache.shardingsphere.test.e2e.operation.pipeline.framework.param.PipelineTestParameter;
+import
org.apache.shardingsphere.test.e2e.operation.pipeline.util.PipelineE2EDistSQLFacade;
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.condition.EnabledIf;
+import org.junit.jupiter.api.extension.ExtensionContext;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ArgumentsSource;
+
+import javax.sql.DataSource;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.concurrent.TimeUnit;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+@PipelineE2ESettings(database = @PipelineE2EDatabaseSettings(type = "MySQL"))
+class MySQLMigrationIdentifierCaseE2EIT extends AbstractMigrationE2EIT {
+
+ private static final String SOURCE_TABLE_NAME = "t_order_copy";
+
+ private static final String TARGET_TABLE_NAME = "T_ORDER";
+
+ private static final long INVENTORY_ORDER_ID = 1L;
+
+ private static final long INCREMENTAL_ORDER_ID = 2L;
+
+ @ParameterizedTest(name = "{0}")
+ @EnabledIf("isEnabled")
+ @ArgumentsSource(PipelineE2ETestCaseArgumentsProvider.class)
+ void assertMigrationToUpperCaseTargetTable(final PipelineTestParameter
testParam) throws SQLException {
+ try (PipelineContainerComposer containerComposer = new
PipelineContainerComposer(testParam)) {
+ assertCaseSensitiveTableNames(containerComposer);
+ PipelineE2EDistSQLFacade distSQLFacade = new
PipelineE2EDistSQLFacade(containerComposer, new MigrationJobType());
+ distSQLFacade.alterPipelineRule();
+ containerComposer.sourceExecuteWithLog(String.format(
+ "CREATE TABLE %s (order_id BIGINT PRIMARY KEY, user_id INT
NOT NULL, status VARCHAR(32) NOT NULL)", SOURCE_TABLE_NAME));
+ addMigrationSourceResource(containerComposer);
+ addMigrationTargetResource(containerComposer);
+ createTargetTableRule(containerComposer);
+ insertSourceRecord(containerComposer, INVENTORY_ORDER_ID);
+ containerComposer.proxyExecuteWithLog(String.format("MIGRATE TABLE
ds_0.%s INTO sharding_DB.%s", SOURCE_TABLE_NAME, TARGET_TABLE_NAME), 5);
+ Awaitility.waitAtMost(10L, TimeUnit.SECONDS).pollInterval(1L,
TimeUnit.SECONDS).until(() -> !distSQLFacade.listJobIds().isEmpty());
+ String jobId = distSQLFacade.getJobIdByTableName("ds_0." +
SOURCE_TABLE_NAME);
+ distSQLFacade.waitJobPreparingStageFinished(jobId);
+ distSQLFacade.waitJobIncrementalStageStarted(jobId);
+ insertSourceRecord(containerComposer, INCREMENTAL_ORDER_ID);
+
assertFalse(distSQLFacade.waitJobIncrementalStageFinished(jobId).isEmpty(),
"Migration incremental stage did not become idle");
+ DataSource targetDataSource =
containerComposer.generateShardingSphereDataSourceFromProxy();
+ containerComposer.assertRecordExists(targetDataSource,
TARGET_TABLE_NAME, INVENTORY_ORDER_ID);
+ containerComposer.assertRecordExists(targetDataSource,
TARGET_TABLE_NAME, INCREMENTAL_ORDER_ID);
+ distSQLFacade.startCheck(jobId, "DATA_MATCH",
ImmutableMap.of("chunk-size", "300", "streaming-range-type", "SMALL"));
+ distSQLFacade.verifyCheck(jobId);
+ distSQLFacade.commit(jobId);
+ assertTrue(distSQLFacade.listJobIds().isEmpty());
+ }
+ }
+
+ private void assertCaseSensitiveTableNames(final PipelineContainerComposer
containerComposer) throws SQLException {
+ try (
+ Connection connection =
containerComposer.getSourceDataSource().getConnection();
+ Statement statement = connection.createStatement();
+ ResultSet resultSet = statement.executeQuery("SELECT
@@lower_case_table_names")) {
+ assertTrue(resultSet.next());
+ assertThat(resultSet.getInt(1), is(0));
+ }
+ }
+
+ private void createTargetTableRule(final PipelineContainerComposer
containerComposer) throws SQLException {
+ containerComposer.proxyExecuteWithLog(String.format("CREATE SHARDING
TABLE RULE %s(STORAGE_UNITS(ds_2,ds_3,ds_4),SHARDING_COLUMN=user_id,"
+ + "TYPE(NAME='hash_mod',PROPERTIES('sharding-count'='6')))",
TARGET_TABLE_NAME), 0);
+ Awaitility.waitAtMost(10L, TimeUnit.SECONDS).pollInterval(1L,
TimeUnit.SECONDS)
+ .until(() -> !containerComposer.queryForListWithLog("SHOW
SHARDING TABLE RULE " + TARGET_TABLE_NAME).isEmpty());
+ }
+
+ private void insertSourceRecord(final PipelineContainerComposer
containerComposer, final long orderId) throws SQLException {
+ containerComposer.sourceExecuteWithLog(String.format("INSERT INTO %s
VALUES (%d, 1, 'OK')", SOURCE_TABLE_NAME, orderId));
+ }
+
+ private static boolean isEnabled(final ExtensionContext context) {
+ return PipelineE2ECondition.isEnabled(context);
+ }
+}
diff --git
a/test/e2e/operation/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/operation/pipeline/cases/migration/general/PostgreSQLMigrationIdentifierCaseE2EIT.java
b/test/e2e/operation/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/operation/pipeline/cases/migration/general/PostgreSQLMigrationIdentifierCaseE2EIT.java
new file mode 100644
index 00000000000..025715ebcea
--- /dev/null
+++
b/test/e2e/operation/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/operation/pipeline/cases/migration/general/PostgreSQLMigrationIdentifierCaseE2EIT.java
@@ -0,0 +1,142 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package
org.apache.shardingsphere.test.e2e.operation.pipeline.cases.migration.general;
+
+import lombok.extern.slf4j.Slf4j;
+import
org.apache.shardingsphere.data.pipeline.scenario.migration.MigrationJobType;
+import
org.apache.shardingsphere.test.e2e.operation.pipeline.cases.PipelineContainerComposer;
+import
org.apache.shardingsphere.test.e2e.operation.pipeline.cases.migration.AbstractMigrationE2EIT;
+import
org.apache.shardingsphere.test.e2e.operation.pipeline.framework.param.PipelineE2ECondition;
+import
org.apache.shardingsphere.test.e2e.operation.pipeline.framework.param.PipelineE2ESettings;
+import
org.apache.shardingsphere.test.e2e.operation.pipeline.framework.param.PipelineE2ESettings.PipelineE2EDatabaseSettings;
+import
org.apache.shardingsphere.test.e2e.operation.pipeline.framework.param.PipelineE2ETestCaseArgumentsProvider;
+import
org.apache.shardingsphere.test.e2e.operation.pipeline.framework.param.PipelineTestParameter;
+import
org.apache.shardingsphere.test.e2e.operation.pipeline.util.PipelineE2EDistSQLFacade;
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.condition.EnabledIf;
+import org.junit.jupiter.api.extension.ExtensionContext;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ArgumentsSource;
+
+import javax.sql.DataSource;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.time.LocalDateTime;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+@Slf4j
+@PipelineE2ESettings(database = @PipelineE2EDatabaseSettings(type =
"PostgreSQL"))
+class PostgreSQLMigrationIdentifierCaseE2EIT extends AbstractMigrationE2EIT {
+
+ private static final String SOURCE_SCHEMA_NAME = "UPPER_SCHEMA";
+
+ private static final String SOURCE_TABLE_NAME = "UPPER_TABLE";
+
+ private static final String TARGET_TABLE_NAME = "upper_schema.t_order";
+
+ private static final long INVENTORY_ORDER_ID = 1L;
+
+ private static final long INCREMENTAL_ORDER_ID = 2L;
+
+ private static final long LOWER_CASE_CONFLICT_ORDER_ID = 20001L;
+
+ private static final long CROSS_SCHEMA_CONFLICT_ORDER_ID = 20002L;
+
+ @ParameterizedTest(name = "{0}")
+ @EnabledIf("isEnabled")
+ @ArgumentsSource(PipelineE2ETestCaseArgumentsProvider.class)
+ void assertMigrationWithUpperCaseSourceIdentifiers(final
PipelineTestParameter testParam) throws SQLException {
+ try (PipelineContainerComposer containerComposer = new
PipelineContainerComposer(testParam)) {
+ PipelineE2EDistSQLFacade distSQLFacade = new
PipelineE2EDistSQLFacade(containerComposer, new MigrationJobType());
+ distSQLFacade.alterPipelineRule();
+ createSourceTables(containerComposer);
+ addMigrationSourceResource(containerComposer);
+ addMigrationTargetResource(containerComposer);
+ createTargetOrderTableRule(containerComposer);
+ insertSourceRecord(containerComposer, qualifiedUpperCaseTable(),
INVENTORY_ORDER_ID);
+ int replicationSlotsCount =
getReplicationSlotsCount(containerComposer);
+ log.info("Start migration at {}, replication slots count: {}",
LocalDateTime.now(), replicationSlotsCount);
+ containerComposer.proxyExecuteWithLog("MIGRATE TABLE
ds_0.`UPPER_SCHEMA`.`UPPER_TABLE` INTO sharding_DB.T_ORDER", 5);
+ Awaitility.waitAtMost(10L, TimeUnit.SECONDS).pollInterval(1L,
TimeUnit.SECONDS).until(() -> !distSQLFacade.listJobIds().isEmpty());
+ String jobId =
distSQLFacade.getJobIdByTableName("ds_0.UPPER_SCHEMA.UPPER_TABLE");
+ distSQLFacade.waitJobPreparingStageFinished(jobId);
+ distSQLFacade.waitJobIncrementalStageStarted(jobId);
+ insertSourceRecord(containerComposer, qualifiedUpperCaseTable(),
INCREMENTAL_ORDER_ID);
+ insertSourceRecord(containerComposer, "upper_schema.upper_table",
LOWER_CASE_CONFLICT_ORDER_ID);
+ insertSourceRecord(containerComposer, "public.\"UPPER_TABLE\"",
CROSS_SCHEMA_CONFLICT_ORDER_ID);
+
assertFalse(distSQLFacade.waitJobIncrementalStageFinished(jobId).isEmpty(),
"Migration incremental stage did not become idle");
+ DataSource targetDataSource =
containerComposer.generateShardingSphereDataSourceFromProxy();
+ containerComposer.assertRecordExists(targetDataSource,
TARGET_TABLE_NAME, INVENTORY_ORDER_ID);
+ containerComposer.assertRecordExists(targetDataSource,
TARGET_TABLE_NAME, INCREMENTAL_ORDER_ID);
+ assertRecordAbsent(targetDataSource, LOWER_CASE_CONFLICT_ORDER_ID);
+ assertRecordAbsent(targetDataSource,
CROSS_SCHEMA_CONFLICT_ORDER_ID);
+ distSQLFacade.startCheckAndVerify(jobId, "DATA_MATCH");
+ distSQLFacade.commit(jobId);
+ assertTrue(distSQLFacade.listJobIds().isEmpty());
+ Awaitility.waitAtMost(30L, TimeUnit.SECONDS).pollInterval(1L,
TimeUnit.SECONDS)
+ .until(() -> getReplicationSlotsCount(containerComposer)
== replicationSlotsCount);
+ }
+ }
+
+ private void createSourceTables(final PipelineContainerComposer
containerComposer) throws SQLException {
+ containerComposer.sourceExecuteWithLog("CREATE SCHEMA IF NOT EXISTS
\"UPPER_SCHEMA\"");
+ containerComposer.sourceExecuteWithLog("CREATE SCHEMA IF NOT EXISTS
upper_schema");
+
containerComposer.sourceExecuteWithLog(createTableSQL(qualifiedUpperCaseTable(),
"\"ORDER_ID\"", "\"USER_ID\"", "\"STATUS\""));
+
containerComposer.sourceExecuteWithLog(createTableSQL("upper_schema.upper_table",
"order_id", "user_id", "status"));
+
containerComposer.sourceExecuteWithLog(createTableSQL("public.\"UPPER_TABLE\"",
"\"ORDER_ID\"", "\"USER_ID\"", "\"STATUS\""));
+ }
+
+ private String createTableSQL(final String tableName, final String
orderIdColumn, final String userIdColumn, final String statusColumn) {
+ return String.format("CREATE TABLE %s (%s BIGINT PRIMARY KEY, %s INT
NOT NULL, %s VARCHAR(32) NOT NULL)", tableName, orderIdColumn, userIdColumn,
statusColumn);
+ }
+
+ private void insertSourceRecord(final PipelineContainerComposer
containerComposer, final String tableName, final long orderId) throws
SQLException {
+ containerComposer.sourceExecuteWithLog(String.format("INSERT INTO %s
VALUES (%d, 1, 'OK')", tableName, orderId));
+ }
+
+ private void assertRecordAbsent(final DataSource dataSource, final long
orderId) throws SQLException {
+ try (
+ Connection connection = dataSource.getConnection();
+ Statement statement = connection.createStatement();
+ ResultSet resultSet =
statement.executeQuery(String.format("SELECT 1 FROM %s WHERE order_id = %d",
TARGET_TABLE_NAME, orderId))) {
+ assertFalse(resultSet.next(), "A record from a conflicting source
table must not be migrated");
+ }
+ }
+
+ private int getReplicationSlotsCount(final PipelineContainerComposer
containerComposer) throws SQLException {
+ try (
+ Connection connection =
containerComposer.getSourceDataSource().getConnection();
+ Statement statement = connection.createStatement();
+ ResultSet resultSet = statement.executeQuery("SELECT COUNT(1)
FROM pg_replication_slots")) {
+ return resultSet.next() ? resultSet.getInt(1) : 0;
+ }
+ }
+
+ private String qualifiedUpperCaseTable() {
+ return String.format("\"%s\".\"%s\"", SOURCE_SCHEMA_NAME,
SOURCE_TABLE_NAME);
+ }
+
+ private static boolean isEnabled(final ExtensionContext context) {
+ return PipelineE2ECondition.isEnabled(context);
+ }
+}
diff --git
a/test/e2e/operation/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/operation/pipeline/framework/container/compose/natived/PipelineNativeContainerComposer.java
b/test/e2e/operation/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/operation/pipeline/framework/container/compose/natived/PipelineNativeContainerComposer.java
index 570ff171322..24d8aeb048b 100644
---
a/test/e2e/operation/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/operation/pipeline/framework/container/compose/natived/PipelineNativeContainerComposer.java
+++
b/test/e2e/operation/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/operation/pipeline/framework/container/compose/natived/PipelineNativeContainerComposer.java
@@ -18,8 +18,10 @@
package
org.apache.shardingsphere.test.e2e.operation.pipeline.framework.container.compose.natived;
import lombok.SneakyThrows;
+import
org.apache.shardingsphere.database.connector.core.metadata.database.enums.QuoteCharacter;
import
org.apache.shardingsphere.database.connector.core.spi.DatabaseTypedSPILoader;
import org.apache.shardingsphere.database.connector.core.type.DatabaseType;
+import
org.apache.shardingsphere.database.connector.core.type.DatabaseTypeRegistry;
import
org.apache.shardingsphere.test.e2e.env.container.storage.option.StorageContainerOption;
import org.apache.shardingsphere.test.e2e.env.runtime.E2ETestEnvironment;
import
org.apache.shardingsphere.test.e2e.operation.pipeline.framework.container.compose.PipelineBaseContainerComposer;
@@ -67,10 +69,11 @@ public final class PipelineNativeContainerComposer extends
PipelineBaseContainer
private void dropTable(final Connection connection, final String
databaseName) throws SQLException {
Map<String, List<String>> schemaAndTableMapper =
getSchemaAndTableMapper(connection, databaseName);
+ QuoteCharacter quoteCharacter = new
DatabaseTypeRegistry(databaseType).getDialectDatabaseMetaData().getQuoteCharacter();
try (Statement statement = connection.createStatement()) {
for (Entry<String, List<String>> entry :
schemaAndTableMapper.entrySet()) {
for (String each : entry.getValue()) {
- statement.executeUpdate(String.format("DROP TABLE %s.%s",
entry.getKey(), each));
+ statement.executeUpdate(String.format("DROP TABLE %s.%s",
quoteCharacter.wrap(entry.getKey()), quoteCharacter.wrap(each)));
}
}
Optional<String> dropSchemaSQL =
dropTableOption.getDropSchemaSQL();
diff --git
a/test/e2e/operation/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/operation/pipeline/framework/container/compose/natived/dialect/PostgreSQLPipelineNativeContainerDropTableOption.java
b/test/e2e/operation/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/operation/pipeline/framework/container/compose/natived/dialect/PostgreSQLPipelineNativeContainerDropTableOption.java
index 5ea3ce0e33a..8be12eec2be 100644
---
a/test/e2e/operation/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/operation/pipeline/framework/container/compose/natived/dialect/PostgreSQLPipelineNativeContainerDropTableOption.java
+++
b/test/e2e/operation/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/operation/pipeline/framework/container/compose/natived/dialect/PostgreSQLPipelineNativeContainerDropTableOption.java
@@ -34,7 +34,7 @@ public final class
PostgreSQLPipelineNativeContainerDropTableOption implements D
@Override
public String getQueryAllSchemaAndTableMapperSQL(final String
databaseName) {
- return "SELECT schemaname, tablename FROM pg_tables WHERE
schemaname='public' OR schemaname='test'";
+ return "SELECT schemaname, tablename FROM pg_tables WHERE schemaname
NOT LIKE 'pg_%' AND schemaname <> 'information_schema'";
}
@Override
diff --git
a/test/e2e/operation/pipeline/src/test/resources/env/container/mysql/cnf/5/my.cnf
b/test/e2e/operation/pipeline/src/test/resources/env/container/mysql/cnf/5/my.cnf
index 5e034f5c393..d7a0a560c92 100644
---
a/test/e2e/operation/pipeline/src/test/resources/env/container/mysql/cnf/5/my.cnf
+++
b/test/e2e/operation/pipeline/src/test/resources/env/container/mysql/cnf/5/my.cnf
@@ -24,7 +24,7 @@ log-bin=mysql-bin
binlog-format=row
binlog-row-image=full
max_connections=600
-lower_case_table_names=1
+lower_case_table_names=0
innodb_lock_wait_timeout=5
default-time-zone='+00:00'
default-authentication-plugin=mysql_native_password
diff --git
a/test/e2e/operation/pipeline/src/test/resources/env/container/mysql/cnf/8/my.cnf
b/test/e2e/operation/pipeline/src/test/resources/env/container/mysql/cnf/8/my.cnf
index 7bfabebe18a..4266f8c269c 100644
---
a/test/e2e/operation/pipeline/src/test/resources/env/container/mysql/cnf/8/my.cnf
+++
b/test/e2e/operation/pipeline/src/test/resources/env/container/mysql/cnf/8/my.cnf
@@ -24,7 +24,7 @@ log-bin=mysql-bin
binlog-format=row
binlog-row-image=full
max_connections=600
-lower_case_table_names=1
+lower_case_table_names=0
innodb_lock_wait_timeout=5
default-time-zone='+00:00'
default-authentication-plugin=caching_sha2_password
