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 96cd926eef0 Use endpoint identifier policies in pipeline SQL
generation and E2E cleanup (#39845)
96cd926eef0 is described below
commit 96cd926eef0f1e2821ed7ce9ac6bf6956628e111
Author: Hongsheng Zhong <[email protected]>
AuthorDate: Tue Sep 15 13:39:48 2026 +0800
Use endpoint identifier policies in pipeline SQL generation and E2E cleanup
(#39845)
* Use endpoint identifier policies in pipeline SQL builders
* Use storage identifier policies in pipeline E2E cleanup
* Normalize migration target identifiers before scheduling and rollback
* Fix index metadata loading on openGauss 3.1
---
.../data/loader/OpenGaussMetaDataLoader.java | 9 +--
.../data/loader/OpenGaussMetaDataLoaderTest.java | 9 +--
.../CRC32TableInventoryCheckCalculator.java | 2 +-
.../core/datasource/PipelineDataSource.java | 27 +++++++
.../importer/sink/type/PipelineDataSourceSink.java | 8 ++-
.../datasource/PipelineJobDataSourcePreparer.java | 3 +-
.../dialect/DialectPipelineSQLBuilder.java | 4 +-
.../segment/PipelineSQLSegmentBuilder.java | 56 ++++++++++++---
...PipelineDataConsistencyCalculateSQLBuilder.java | 8 ++-
.../sqlbuilder/sql/PipelineImportSQLBuilder.java | 83 ++++++++++++++++------
.../sqlbuilder/sql/PipelinePrepareSQLBuilder.java | 30 ++++++--
.../CRC32TableInventoryCheckCalculatorTest.java | 2 +
.../core/datasource/PipelineDataSourceTest.java | 60 ++++++++++++++++
.../segment/PipelineSQLSegmentBuilderTest.java | 56 +++++++++++++--
...lineDataConsistencyCalculateSQLBuilderTest.java | 5 +-
.../sql/PipelineImportSQLBuilderTest.java | 73 ++++++++++++++++++-
.../sql/PipelinePrepareSQLBuilderTest.java | 25 ++++++-
.../mysql/sqlbuilder/MySQLPipelineSQLBuilder.java | 7 +-
.../sqlbuilder/MySQLPipelineSQLBuilderTest.java | 18 ++++-
.../sqlbuilder/OpenGaussPipelineSQLBuilder.java | 8 ++-
.../OpenGaussPipelineSQLBuilderTest.java | 18 ++++-
.../sqlbuilder/PostgreSQLPipelineSQLBuilder.java | 11 +--
.../sqlbuilder/PipelinePrepareSQLBuilderTest.java | 5 +-
.../PostgreSQLPipelineSQLBuilderTest.java | 17 ++++-
.../scenario/migration/api/MigrationJobAPI.java | 7 +-
.../migration/api/MigrationJobAPITest.java | 57 +++++++++++++++
.../handler/update/MigrateTableExecutor.java | 9 ++-
.../handler/update/MigrateTableExecutorTest.java | 35 ++++++++-
.../pipeline/cases/PipelineContainerComposer.java | 9 ++-
29 files changed, 579 insertions(+), 82 deletions(-)
diff --git
a/database/connector/dialect/opengauss/src/main/java/org/apache/shardingsphere/database/connector/opengauss/metadata/data/loader/OpenGaussMetaDataLoader.java
b/database/connector/dialect/opengauss/src/main/java/org/apache/shardingsphere/database/connector/opengauss/metadata/data/loader/OpenGaussMetaDataLoader.java
index d4d60df35e6..fbaf43af22b 100644
---
a/database/connector/dialect/opengauss/src/main/java/org/apache/shardingsphere/database/connector/opengauss/metadata/data/loader/OpenGaussMetaDataLoader.java
+++
b/database/connector/dialect/opengauss/src/main/java/org/apache/shardingsphere/database/connector/opengauss/metadata/data/loader/OpenGaussMetaDataLoader.java
@@ -68,10 +68,11 @@ public final class OpenGaussMetaDataLoader implements
DialectMetaDataLoader {
private static final String ADVANCE_INDEX_META_DATA_SQL =
"SELECT idx.relname as index_name, insp.nspname as index_schema,
tbl.relname as table_name, att.attname AS column_name, pgi.indisunique as
is_unique"
- + " FROM pg_index pgi JOIN pg_class idx ON idx.oid =
pgi.indexrelid JOIN pg_namespace insp ON insp.oid = idx.relnamespace JOIN
pg_class tbl ON tbl.oid = pgi.indrelid"
- + " JOIN pg_namespace tnsp ON tnsp.oid = tbl.relnamespace
JOIN generate_subscripts(pgi.indkey, 1) AS index_position(position) ON TRUE"
- + " JOIN pg_attribute att ON att.attrelid = tbl.oid AND
att.attnum = pgi.indkey[index_position.position] WHERE tnsp.nspname IN (%s)"
- + " ORDER BY insp.nspname, idx.relname,
index_position.position";
+ + " FROM (SELECT indexrelid, indrelid, indisunique,
indkey, generate_subscripts(indkey, 1) AS position FROM pg_index) pgi"
+ + " JOIN pg_class idx ON idx.oid = pgi.indexrelid JOIN
pg_namespace insp ON insp.oid = idx.relnamespace JOIN pg_class tbl ON tbl.oid =
pgi.indrelid"
+ + " JOIN pg_namespace tnsp ON tnsp.oid = tbl.relnamespace"
+ + " JOIN pg_attribute att ON att.attrelid = tbl.oid AND
att.attnum = pgi.indkey[pgi.position] WHERE tnsp.nspname IN (%s)"
+ + " ORDER BY insp.nspname, idx.relname, pgi.position";
private static final String BASIC_VIEW_META_DATA_SQL = "SELECT
table_schema, table_name FROM information_schema.views WHERE table_schema IN
(%s)";
diff --git
a/database/connector/dialect/opengauss/src/test/java/org/apache/shardingsphere/database/connector/opengauss/metadata/data/loader/OpenGaussMetaDataLoaderTest.java
b/database/connector/dialect/opengauss/src/test/java/org/apache/shardingsphere/database/connector/opengauss/metadata/data/loader/OpenGaussMetaDataLoaderTest.java
index 705a8c755cd..99ce47fe6cc 100644
---
a/database/connector/dialect/opengauss/src/test/java/org/apache/shardingsphere/database/connector/opengauss/metadata/data/loader/OpenGaussMetaDataLoaderTest.java
+++
b/database/connector/dialect/opengauss/src/test/java/org/apache/shardingsphere/database/connector/opengauss/metadata/data/loader/OpenGaussMetaDataLoaderTest.java
@@ -67,10 +67,11 @@ class OpenGaussMetaDataLoaderTest {
private static final String ADVANCE_INDEX_META_DATA_SQL =
"SELECT idx.relname as index_name, insp.nspname as index_schema,
tbl.relname as table_name, att.attname AS column_name, pgi.indisunique as
is_unique"
- + " FROM pg_index pgi JOIN pg_class idx ON idx.oid =
pgi.indexrelid JOIN pg_namespace insp ON insp.oid = idx.relnamespace JOIN
pg_class tbl ON tbl.oid = pgi.indrelid"
- + " JOIN pg_namespace tnsp ON tnsp.oid = tbl.relnamespace
JOIN generate_subscripts(pgi.indkey, 1) AS index_position(position) ON TRUE"
- + " JOIN pg_attribute att ON att.attrelid = tbl.oid AND
att.attnum = pgi.indkey[index_position.position] WHERE tnsp.nspname IN
('public')"
- + " ORDER BY insp.nspname, idx.relname,
index_position.position";
+ + " FROM (SELECT indexrelid, indrelid, indisunique,
indkey, generate_subscripts(indkey, 1) AS position FROM pg_index) pgi"
+ + " JOIN pg_class idx ON idx.oid = pgi.indexrelid JOIN
pg_namespace insp ON insp.oid = idx.relnamespace JOIN pg_class tbl ON tbl.oid =
pgi.indrelid"
+ + " JOIN pg_namespace tnsp ON tnsp.oid = tbl.relnamespace"
+ + " JOIN pg_attribute att ON att.attrelid = tbl.oid AND
att.attnum = pgi.indkey[pgi.position] WHERE tnsp.nspname IN ('public')"
+ + " ORDER BY insp.nspname, idx.relname, pgi.position";
private static final String VIEW_META_DATA_SQL_WITHOUT_TABLES = "SELECT
table_schema, table_name FROM information_schema.views WHERE table_schema IN
('public')";
diff --git
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/calculator/CRC32TableInventoryCheckCalculator.java
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/calculator/CRC32TableInventoryCheckCalculator.java
index a83639fb780..65770eb59e5 100644
---
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/calculator/CRC32TableInventoryCheckCalculator.java
+++
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/calculator/CRC32TableInventoryCheckCalculator.java
@@ -45,7 +45,7 @@ public final class CRC32TableInventoryCheckCalculator extends
AbstractTableInven
@Override
public Iterable<TableInventoryCheckCalculatedResult> calculate(final
TableInventoryCalculateParameter param) {
- PipelineDataConsistencyCalculateSQLBuilder pipelineSQLBuilder = new
PipelineDataConsistencyCalculateSQLBuilder(param.getDatabaseType());
+ PipelineDataConsistencyCalculateSQLBuilder pipelineSQLBuilder = new
PipelineDataConsistencyCalculateSQLBuilder(param.getDatabaseType(),
param.getDataSource().getIdentifierContext());
List<CalculatedItem> calculatedItems =
param.getColumnNames().stream().map(each -> calculateCRC32(pipelineSQLBuilder,
param, each)).collect(Collectors.toList());
return Collections.singletonList(new
CalculatedResult(calculatedItems.get(0).getRecordsCount(),
calculatedItems.stream().map(CalculatedItem::getCrc32).collect(Collectors.toList())));
}
diff --git
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/datasource/PipelineDataSource.java
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/datasource/PipelineDataSource.java
index 20408f7fea3..70038a9b237 100644
---
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/datasource/PipelineDataSource.java
+++
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/datasource/PipelineDataSource.java
@@ -21,10 +21,16 @@ import lombok.Getter;
import lombok.RequiredArgsConstructor;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.concurrent.ConcurrentException;
import
org.apache.shardingsphere.data.pipeline.api.PipelineDataSourceConfiguration;
+import
org.apache.shardingsphere.data.pipeline.core.util.PipelineLazyInitializer;
import org.apache.shardingsphere.data.pipeline.spi.PipelineDataSourceCreator;
import org.apache.shardingsphere.database.connector.core.type.DatabaseType;
+import
org.apache.shardingsphere.driver.jdbc.core.datasource.ShardingSphereDataSource;
import
org.apache.shardingsphere.infra.datasource.pool.destroyer.DataSourcePoolDestroyer;
+import
org.apache.shardingsphere.infra.metadata.identifier.DatabaseIdentifierContext;
+import
org.apache.shardingsphere.infra.metadata.identifier.DatabaseIdentifierContextFactory;
+import
org.apache.shardingsphere.infra.metadata.identifier.IdentifierCasePolicyResolver;
import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
import javax.sql.DataSource;
@@ -49,12 +55,33 @@ public final class PipelineDataSource implements
DataSource, AutoCloseable {
private final AtomicBoolean closed = new AtomicBoolean(false);
+ private final PipelineLazyInitializer<DatabaseIdentifierContext>
identifierContext = new PipelineLazyInitializer<DatabaseIdentifierContext>() {
+
+ @Override
+ protected DatabaseIdentifierContext doInitialize() {
+ return dataSource instanceof ShardingSphereDataSource
+ ? new
DatabaseIdentifierContext(IdentifierCasePolicyResolver.resolveProtocol(databaseType))
+ : DatabaseIdentifierContextFactory.create(databaseType,
dataSource);
+ }
+ };
+
@SneakyThrows(SQLException.class)
public PipelineDataSource(final PipelineDataSourceConfiguration
pipelineDataSourceConfig) {
dataSource =
TypedSPILoader.getService(PipelineDataSourceCreator.class,
pipelineDataSourceConfig.getType()).create(pipelineDataSourceConfig.getDataSourceConfiguration());
databaseType = pipelineDataSourceConfig.getDatabaseType();
}
+ /**
+ * Get the SQL endpoint's identifier context, resolved once for this data
source.
+ * ShardingSphere JDBC receives protocol identifiers; native data sources
use their storage policy.
+ *
+ * @return identifier context
+ */
+ @SneakyThrows(ConcurrentException.class)
+ public DatabaseIdentifierContext getIdentifierContext() {
+ return identifierContext.get();
+ }
+
/**
* Whether underlying data source is closed or not.
*
diff --git
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/importer/sink/type/PipelineDataSourceSink.java
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/importer/sink/type/PipelineDataSourceSink.java
index 6a6e299dc32..c82216fe582 100644
---
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/importer/sink/type/PipelineDataSourceSink.java
+++
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/importer/sink/type/PipelineDataSourceSink.java
@@ -19,7 +19,9 @@ package
org.apache.shardingsphere.data.pipeline.core.importer.sink.type;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
+import
org.apache.shardingsphere.data.pipeline.api.type.ShardingSpherePipelineDataSourceConfiguration;
import
org.apache.shardingsphere.data.pipeline.core.constant.PipelineSQLOperationType;
+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.exception.job.PipelineImporterJobWriteException;
import
org.apache.shardingsphere.data.pipeline.core.importer.ImporterConfiguration;
@@ -66,8 +68,10 @@ public final class PipelineDataSourceSink implements
PipelineSink {
public PipelineDataSourceSink(final ImporterConfiguration importerConfig,
final PipelineDataSourceManager dataSourceManager) {
this.importerConfig = importerConfig;
- dataSource =
dataSourceManager.getDataSource(importerConfig.getDataSourceConfig());
- importSQLBuilder = new
PipelineImportSQLBuilder(importerConfig.getDataSourceConfig().getDatabaseType());
+ PipelineDataSource pipelineDataSource =
dataSourceManager.getDataSource(importerConfig.getDataSourceConfig());
+ dataSource = pipelineDataSource;
+ importSQLBuilder = new
PipelineImportSQLBuilder(importerConfig.getDataSourceConfig().getDatabaseType(),
pipelineDataSource.getIdentifierContext(),
+ importerConfig.getDataSourceConfig() instanceof
ShardingSpherePipelineDataSourceConfiguration);
groupEngine = new DataRecordGroupEngine();
runningStatement = new AtomicReference<>();
}
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 d460ba2fe7d..1c79abf32c0 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
@@ -74,13 +74,14 @@ public final class PipelineJobDataSourcePreparer {
return;
}
String defaultSchema =
dialectDatabaseMetaData.getSchemaOption().getDefaultSchema().orElse(null);
- PipelinePrepareSQLBuilder pipelineSQLBuilder = new
PipelinePrepareSQLBuilder(targetDatabaseType);
Collection<String> createdSchemaNames = new
HashSet<>(param.getCreateTableConfigurations().size(), 1F);
for (CreateTableConfiguration each :
param.getCreateTableConfigurations()) {
String targetSchemaName = each.getTargetName().getSchemaName();
if (null == targetSchemaName ||
targetSchemaName.equalsIgnoreCase(defaultSchema) ||
createdSchemaNames.contains(targetSchemaName)) {
continue;
}
+ PipelinePrepareSQLBuilder pipelineSQLBuilder = new
PipelinePrepareSQLBuilder(
+ targetDatabaseType,
param.getDataSourceManager().getDataSource(each.getTargetDataSourceConfig()).getIdentifierContext());
Optional<String> sql =
pipelineSQLBuilder.buildCreateSchemaSQL(targetSchemaName);
if (sql.isPresent()) {
executeCreateSchema(param.getDataSourceManager(),
each.getTargetDataSourceConfig(), sql.get());
diff --git
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/dialect/DialectPipelineSQLBuilder.java
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/dialect/DialectPipelineSQLBuilder.java
index 6c37474ed05..5258199c333 100644
---
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/dialect/DialectPipelineSQLBuilder.java
+++
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/dialect/DialectPipelineSQLBuilder.java
@@ -18,6 +18,7 @@
package org.apache.shardingsphere.data.pipeline.core.sqlbuilder.dialect;
import org.apache.shardingsphere.data.pipeline.core.ingest.record.DataRecord;
+import
org.apache.shardingsphere.data.pipeline.core.sqlbuilder.segment.PipelineSQLSegmentBuilder;
import org.apache.shardingsphere.database.connector.core.spi.DatabaseTypedSPI;
import org.apache.shardingsphere.infra.spi.annotation.SingletonSPI;
@@ -46,9 +47,10 @@ public interface DialectPipelineSQLBuilder extends
DatabaseTypedSPI {
* Build on duplicate clause of insert SQL.
*
* @param dataRecord data record
+ * @param sqlSegmentBuilder SQL endpoint identifier builder
* @return built SQL clause
*/
- default Optional<String> buildInsertOnDuplicateClause(final DataRecord
dataRecord) {
+ default Optional<String> buildInsertOnDuplicateClause(final DataRecord
dataRecord, final PipelineSQLSegmentBuilder sqlSegmentBuilder) {
return Optional.empty();
}
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 cdca225b028..86c5364188f 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
@@ -19,9 +19,14 @@ package
org.apache.shardingsphere.data.pipeline.core.sqlbuilder.segment;
import com.google.common.base.Strings;
import
org.apache.shardingsphere.database.connector.core.metadata.database.metadata.DialectDatabaseMetaData;
+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.DatabaseTypeRegistry;
+import org.apache.shardingsphere.infra.annotation.HighFrequencyInvocation;
import org.apache.shardingsphere.infra.metadata.database.schema.QualifiedTable;
+import
org.apache.shardingsphere.infra.metadata.identifier.DatabaseIdentifierContext;
+import
org.apache.shardingsphere.sql.parser.statement.core.value.identifier.IdentifierValue;
+import org.jspecify.annotations.Nullable;
import java.util.function.Function;
@@ -30,23 +35,53 @@ import java.util.function.Function;
*/
public final class PipelineSQLSegmentBuilder {
- private final DatabaseTypeRegistry databaseTypeRegistry;
-
private final DialectDatabaseMetaData dialectDatabaseMetaData;
+ @Nullable
+ private final DatabaseIdentifierContext identifierContext;
+
+ /**
+ * Create a builder for resolved actual identifiers without resolving a
storage policy.
+ *
+ * @param databaseType database type
+ */
public PipelineSQLSegmentBuilder(final DatabaseType databaseType) {
- databaseTypeRegistry = new DatabaseTypeRegistry(databaseType);
- dialectDatabaseMetaData =
databaseTypeRegistry.getDialectDatabaseMetaData();
+ this(databaseType, null);
+ }
+
+ /**
+ * Create a builder using the SQL endpoint's identifier context.
+ *
+ * @param databaseType database type
+ * @param identifierContext endpoint context, or null for actual
identifiers only
+ */
+ public PipelineSQLSegmentBuilder(final DatabaseType databaseType,
@Nullable final DatabaseIdentifierContext identifierContext) {
+ dialectDatabaseMetaData = new
DatabaseTypeRegistry(databaseType).getDialectDatabaseMetaData();
+ this.identifierContext = identifierContext;
}
/**
* Get escaped identifier.
*
- * @param identifier identifier to be processed
+ * @param identifierScope identifier scope
+ * @param identifier reference identifier, retaining its quote characters
* @return escaped identifier
*/
- public String getEscapedIdentifier(final String identifier) {
- return "*".equals(identifier) ? identifier :
dialectDatabaseMetaData.getQuoteCharacter().wrap(databaseTypeRegistry.formatIdentifierPattern(identifier));
+ @HighFrequencyInvocation
+ public String getEscapedIdentifier(final IdentifierScope identifierScope,
final String identifier) {
+ return "*".equals(identifier) ? identifier :
getEscapedActualIdentifier(normalizeStorageIdentifier(identifierScope,
identifier));
+ }
+
+ /**
+ * Normalize a reference for storage, including names persisted for later
cleanup.
+ *
+ * @param identifierScope identifier scope
+ * @param identifier reference identifier
+ * @return storage identifier
+ */
+ @HighFrequencyInvocation
+ public String normalizeStorageIdentifier(final IdentifierScope
identifierScope, final String identifier) {
+ return identifierContext.normalizeStorage(identifierScope, new
IdentifierValue(identifier));
}
/**
@@ -55,6 +90,7 @@ public final class PipelineSQLSegmentBuilder {
* @param identifier actual identifier to be processed
* @return escaped actual identifier
*/
+ @HighFrequencyInvocation
public String getEscapedActualIdentifier(final String identifier) {
return "*".equals(identifier) ? identifier :
dialectDatabaseMetaData.getQuoteCharacter().wrap(identifier);
}
@@ -66,8 +102,12 @@ public final class PipelineSQLSegmentBuilder {
* @param tableName table name
* @return qualified table name
*/
+ @HighFrequencyInvocation
public String getQualifiedTableName(final String schemaName, final String
tableName) {
- return buildQualifiedTableName(schemaName, tableName,
this::getEscapedIdentifier);
+ String escapedTableName = getEscapedIdentifier(IdentifierScope.TABLE,
tableName);
+ return dialectDatabaseMetaData.getSchemaOption().isSchemaAvailable()
&& !Strings.isNullOrEmpty(schemaName)
+ ? String.join(".",
getEscapedIdentifier(IdentifierScope.SCHEMA, schemaName), escapedTableName)
+ : escapedTableName;
}
/**
diff --git
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/sql/PipelineDataConsistencyCalculateSQLBuilder.java
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/sql/PipelineDataConsistencyCalculateSQLBuilder.java
index 67501bf4d83..472835fab98 100644
---
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/sql/PipelineDataConsistencyCalculateSQLBuilder.java
+++
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/sql/PipelineDataConsistencyCalculateSQLBuilder.java
@@ -19,9 +19,11 @@ package
org.apache.shardingsphere.data.pipeline.core.sqlbuilder.sql;
import
org.apache.shardingsphere.data.pipeline.core.sqlbuilder.dialect.DialectPipelineSQLBuilder;
import
org.apache.shardingsphere.data.pipeline.core.sqlbuilder.segment.PipelineSQLSegmentBuilder;
+import
org.apache.shardingsphere.database.connector.core.metadata.identifier.IdentifierScope;
import
org.apache.shardingsphere.database.connector.core.spi.DatabaseTypedSPILoader;
import org.apache.shardingsphere.database.connector.core.type.DatabaseType;
import org.apache.shardingsphere.infra.metadata.database.schema.QualifiedTable;
+import
org.apache.shardingsphere.infra.metadata.identifier.DatabaseIdentifierContext;
import java.util.Optional;
@@ -34,9 +36,9 @@ public final class PipelineDataConsistencyCalculateSQLBuilder
{
private final PipelineSQLSegmentBuilder sqlSegmentBuilder;
- public PipelineDataConsistencyCalculateSQLBuilder(final DatabaseType
databaseType) {
+ public PipelineDataConsistencyCalculateSQLBuilder(final DatabaseType
databaseType, final DatabaseIdentifierContext identifierContext) {
dialectSQLBuilder =
DatabaseTypedSPILoader.getService(DialectPipelineSQLBuilder.class,
databaseType);
- sqlSegmentBuilder = new PipelineSQLSegmentBuilder(databaseType);
+ sqlSegmentBuilder = new PipelineSQLSegmentBuilder(databaseType,
identifierContext);
}
/**
@@ -47,6 +49,6 @@ public final class PipelineDataConsistencyCalculateSQLBuilder
{
* @return built SQL
*/
public Optional<String> buildCRC32SQL(final QualifiedTable qualifiedTable,
final String columnName) {
- return
dialectSQLBuilder.buildCRC32SQL(sqlSegmentBuilder.getQualifiedTableName(qualifiedTable),
sqlSegmentBuilder.getEscapedIdentifier(columnName));
+ return
dialectSQLBuilder.buildCRC32SQL(sqlSegmentBuilder.getQualifiedTableName(qualifiedTable),
sqlSegmentBuilder.getEscapedIdentifier(IdentifierScope.COLUMN, columnName));
}
}
diff --git
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/sql/PipelineImportSQLBuilder.java
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/sql/PipelineImportSQLBuilder.java
index a6671af6c79..487ac37fcb1 100644
---
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/sql/PipelineImportSQLBuilder.java
+++
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/sql/PipelineImportSQLBuilder.java
@@ -19,21 +19,27 @@ package
org.apache.shardingsphere.data.pipeline.core.sqlbuilder.sql;
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
+import com.google.common.base.Strings;
import org.apache.shardingsphere.data.pipeline.core.ingest.record.Column;
import org.apache.shardingsphere.data.pipeline.core.ingest.record.DataRecord;
import
org.apache.shardingsphere.data.pipeline.core.sqlbuilder.dialect.DialectPipelineSQLBuilder;
import
org.apache.shardingsphere.data.pipeline.core.sqlbuilder.segment.PipelineSQLSegmentBuilder;
+import
org.apache.shardingsphere.database.connector.core.metadata.identifier.IdentifierScope;
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.infra.annotation.HighFrequencyInvocation;
+import
org.apache.shardingsphere.infra.metadata.identifier.DatabaseIdentifierContext;
import java.util.Collection;
import java.util.Objects;
import java.util.Optional;
-import java.util.stream.Collectors;
+import java.util.StringJoiner;
/**
* Pipeline import SQL builder engine.
*/
+@HighFrequencyInvocation
public final class PipelineImportSQLBuilder {
private static final String INSERT_SQL_CACHE_KEY_PREFIX = "INSERT_";
@@ -48,9 +54,22 @@ public final class PipelineImportSQLBuilder {
private final Cache<String, String> sqlCache;
- public PipelineImportSQLBuilder(final DatabaseType databaseType) {
+ private final boolean actualTableNames;
+
+ private final boolean schemaAvailable;
+
+ /**
+ * Create an import builder with endpoint identifier policies.
+ *
+ * @param databaseType database type
+ * @param identifierContext endpoint identifier context
+ * @param actualTableNames whether record table names are resolved actual
names; schemas and columns remain references
+ */
+ public PipelineImportSQLBuilder(final DatabaseType databaseType, final
DatabaseIdentifierContext identifierContext, final boolean actualTableNames) {
dialectSQLBuilder =
DatabaseTypedSPILoader.getService(DialectPipelineSQLBuilder.class,
databaseType);
- sqlSegmentBuilder = new PipelineSQLSegmentBuilder(databaseType);
+ sqlSegmentBuilder = new PipelineSQLSegmentBuilder(databaseType,
identifierContext);
+ this.actualTableNames = actualTableNames;
+ schemaAvailable = new
DatabaseTypeRegistry(databaseType).getDialectDatabaseMetaData().getSchemaOption().isSchemaAvailable();
sqlCache =
Caffeine.newBuilder().initialCapacity(16).maximumSize(1024L).build();
}
@@ -62,7 +81,7 @@ public final class PipelineImportSQLBuilder {
* @return insert SQL
*/
public String buildInsertSQL(final String schemaName, final DataRecord
dataRecord) {
- String sqlCacheKey = INSERT_SQL_CACHE_KEY_PREFIX +
dataRecord.getTableName();
+ String sqlCacheKey =
INSERT_SQL_CACHE_KEY_PREFIX.concat(dataRecord.getTableName());
if (null == sqlCache.getIfPresent(sqlCacheKey)) {
sqlCache.put(sqlCacheKey, buildInsertSQL0(schemaName, dataRecord));
}
@@ -71,13 +90,25 @@ public final class PipelineImportSQLBuilder {
private String buildInsertSQL0(final String schemaName, final DataRecord
dataRecord) {
String insertMainClause = buildInsertMainClause(schemaName,
dataRecord);
- return
dialectSQLBuilder.buildInsertOnDuplicateClause(dataRecord).map(optional ->
insertMainClause + " " + optional).orElse(insertMainClause);
+ return dialectSQLBuilder.buildInsertOnDuplicateClause(dataRecord,
sqlSegmentBuilder).map(optional -> String.join(" ", insertMainClause,
optional)).orElse(insertMainClause);
}
private String buildInsertMainClause(final String schemaName, final
DataRecord dataRecord) {
- String columnsLiteral = dataRecord.getColumns().stream().map(each ->
sqlSegmentBuilder.getEscapedIdentifier(each.getName())).collect(Collectors.joining(","));
- String valuesLiteral = dataRecord.getColumns().stream().map(each ->
"?").collect(Collectors.joining(","));
- return String.format("INSERT INTO %s(%s) VALUES(%s)",
sqlSegmentBuilder.getQualifiedTableName(schemaName, dataRecord.getTableName()),
columnsLiteral, valuesLiteral);
+ StringJoiner columnsLiteral = new StringJoiner(",");
+ StringJoiner valuesLiteral = new StringJoiner(",");
+ for (Column each : dataRecord.getColumns()) {
+
columnsLiteral.add(sqlSegmentBuilder.getEscapedIdentifier(IdentifierScope.COLUMN,
each.getName()));
+ valuesLiteral.add("?");
+ }
+ return String.format("INSERT INTO %s(%s) VALUES(%s)",
getQualifiedTableName(schemaName, dataRecord.getTableName()), columnsLiteral,
valuesLiteral);
+ }
+
+ private String getQualifiedTableName(final String schemaName, final String
tableName) {
+ if (!actualTableNames) {
+ return sqlSegmentBuilder.getQualifiedTableName(schemaName,
tableName);
+ }
+ String actualSchemaName = schemaAvailable &&
!Strings.isNullOrEmpty(schemaName) ?
sqlSegmentBuilder.normalizeStorageIdentifier(IdentifierScope.SCHEMA,
schemaName) : schemaName;
+ return sqlSegmentBuilder.getQualifiedActualTableName(actualSchemaName,
tableName);
}
/**
@@ -89,18 +120,33 @@ public final class PipelineImportSQLBuilder {
* @return update SQL
*/
public String buildUpdateSQL(final String schemaName, final DataRecord
dataRecord, final Collection<Column> conditionColumns) {
- String sqlCacheKey = UPDATE_SQL_CACHE_KEY_PREFIX +
dataRecord.getTableName();
+ String sqlCacheKey =
UPDATE_SQL_CACHE_KEY_PREFIX.concat(dataRecord.getTableName());
if (null == sqlCache.getIfPresent(sqlCacheKey)) {
sqlCache.put(sqlCacheKey, buildUpdateSQL0(schemaName, dataRecord,
conditionColumns));
}
- Collection<Column> setColumns =
dataRecord.getColumns().stream().filter(Column::isUpdated).collect(Collectors.toList());
- String updateSetClause = setColumns.stream().map(each ->
sqlSegmentBuilder.getEscapedIdentifier(each.getName()) + " =
?").collect(Collectors.joining(","));
+ StringJoiner updateSetClause = new StringJoiner(",");
+ for (Column each : dataRecord.getColumns()) {
+ if (each.isUpdated()) {
+
updateSetClause.add(sqlSegmentBuilder.getEscapedIdentifier(IdentifierScope.COLUMN,
each.getName()).concat(" = ?"));
+ }
+ }
return
String.format(Objects.requireNonNull(sqlCache.getIfPresent(sqlCacheKey)),
updateSetClause);
}
private String buildUpdateSQL0(final String schemaName, final DataRecord
dataRecord, final Collection<Column> conditionColumns) {
- String updateMainClause = String.format("UPDATE %s SET %%s",
sqlSegmentBuilder.getQualifiedTableName(schemaName, dataRecord.getTableName()));
- return buildWhereClause(conditionColumns).map(optional ->
updateMainClause + optional).orElse(updateMainClause);
+ String updateMainClause = String.format("UPDATE %s SET %%s",
getQualifiedTableName(schemaName, dataRecord.getTableName()));
+ return buildWhereClause(conditionColumns).map(optional ->
updateMainClause.concat(optional)).orElse(updateMainClause);
+ }
+
+ private Optional<String> buildWhereClause(final Collection<Column>
conditionColumns) {
+ if (conditionColumns.isEmpty()) {
+ return Optional.empty();
+ }
+ StringJoiner result = new StringJoiner(" AND ", " WHERE ", "");
+ for (Column each : conditionColumns) {
+
result.add(sqlSegmentBuilder.getEscapedIdentifier(IdentifierScope.COLUMN,
each.getName()).concat(" = ?"));
+ }
+ return Optional.of(result.toString());
}
/**
@@ -112,7 +158,7 @@ public final class PipelineImportSQLBuilder {
* @return delete SQL
*/
public String buildDeleteSQL(final String schemaName, final DataRecord
dataRecord, final Collection<Column> conditionColumns) {
- String sqlCacheKey = DELETE_SQL_CACHE_KEY_PREFIX +
dataRecord.getTableName();
+ String sqlCacheKey =
DELETE_SQL_CACHE_KEY_PREFIX.concat(dataRecord.getTableName());
if (null == sqlCache.getIfPresent(sqlCacheKey)) {
sqlCache.put(sqlCacheKey, buildDeleteSQL0(schemaName, dataRecord,
conditionColumns));
}
@@ -121,16 +167,11 @@ public final class PipelineImportSQLBuilder {
private String buildDeleteSQL0(final String schemaName, final DataRecord
dataRecord, final Collection<Column> conditionColumns) {
String deleteMainClause = buildDeleteMainClause(schemaName,
dataRecord);
- return buildWhereClause(conditionColumns).map(optional ->
deleteMainClause + optional).orElse(deleteMainClause);
+ return buildWhereClause(conditionColumns).map(optional ->
deleteMainClause.concat(optional)).orElse(deleteMainClause);
}
private String buildDeleteMainClause(final String schemaName, final
DataRecord dataRecord) {
- return String.format("DELETE FROM %s",
sqlSegmentBuilder.getQualifiedTableName(schemaName, dataRecord.getTableName()));
+ return String.format("DELETE FROM %s",
getQualifiedTableName(schemaName, dataRecord.getTableName()));
}
- private Optional<String> buildWhereClause(final Collection<Column>
conditionColumns) {
- return conditionColumns.isEmpty()
- ? Optional.empty()
- : Optional.of(" WHERE " + conditionColumns.stream().map(each
-> sqlSegmentBuilder.getEscapedIdentifier(each.getName()) + " =
?").collect(Collectors.joining(" AND ")));
- }
}
diff --git
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/sql/PipelinePrepareSQLBuilder.java
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/sql/PipelinePrepareSQLBuilder.java
index b33809e8547..7cc95e40117 100644
---
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/sql/PipelinePrepareSQLBuilder.java
+++
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/sql/PipelinePrepareSQLBuilder.java
@@ -19,8 +19,11 @@ package
org.apache.shardingsphere.data.pipeline.core.sqlbuilder.sql;
import
org.apache.shardingsphere.data.pipeline.core.sqlbuilder.dialect.DialectPipelineSQLBuilder;
import
org.apache.shardingsphere.data.pipeline.core.sqlbuilder.segment.PipelineSQLSegmentBuilder;
+import
org.apache.shardingsphere.database.connector.core.metadata.identifier.IdentifierScope;
import
org.apache.shardingsphere.database.connector.core.spi.DatabaseTypedSPILoader;
import org.apache.shardingsphere.database.connector.core.type.DatabaseType;
+import
org.apache.shardingsphere.infra.metadata.identifier.DatabaseIdentifierContext;
+import org.jspecify.annotations.Nullable;
import java.util.Optional;
@@ -33,9 +36,24 @@ public final class PipelinePrepareSQLBuilder {
private final PipelineSQLSegmentBuilder sqlSegmentBuilder;
+ /**
+ * Create a builder for actual identifiers only. Creating a schema
requires an endpoint context.
+ *
+ * @param databaseType database type
+ */
public PipelinePrepareSQLBuilder(final DatabaseType databaseType) {
+ this(databaseType, null);
+ }
+
+ /**
+ * Create a builder using the SQL endpoint identifier context.
+ *
+ * @param databaseType database type
+ * @param identifierContext endpoint context, or null for actual
identifiers only
+ */
+ public PipelinePrepareSQLBuilder(final DatabaseType databaseType,
@Nullable final DatabaseIdentifierContext identifierContext) {
dialectSQLBuilder =
DatabaseTypedSPILoader.getService(DialectPipelineSQLBuilder.class,
databaseType);
- sqlSegmentBuilder = new PipelineSQLSegmentBuilder(databaseType);
+ sqlSegmentBuilder = new PipelineSQLSegmentBuilder(databaseType,
identifierContext);
}
/**
@@ -45,18 +63,18 @@ public final class PipelinePrepareSQLBuilder {
* @return create schema SQL
*/
public Optional<String> buildCreateSchemaSQL(final String schemaName) {
- return
dialectSQLBuilder.buildCreateSchemaSQL(sqlSegmentBuilder.getEscapedIdentifier(schemaName));
+ return
dialectSQLBuilder.buildCreateSchemaSQL(sqlSegmentBuilder.getEscapedIdentifier(IdentifierScope.SCHEMA,
schemaName));
}
/**
- * Build drop SQL.
+ * Build drop SQL for resolved actual identifiers.
*
- * @param schemaName schema name
- * @param tableName table name
+ * @param schemaName actual schema name
+ * @param tableName actual table name
* @return drop SQL
*/
public String buildDropSQL(final String schemaName, final String
tableName) {
- return String.format("DROP TABLE IF EXISTS %s",
sqlSegmentBuilder.getQualifiedTableName(schemaName, tableName));
+ return String.format("DROP TABLE IF EXISTS %s",
sqlSegmentBuilder.getQualifiedActualTableName(schemaName, tableName));
}
/**
diff --git
a/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/calculator/CRC32TableInventoryCheckCalculatorTest.java
b/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/calculator/CRC32TableInventoryCheckCalculatorTest.java
index 0dc4230f694..7e52368c098 100644
---
a/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/calculator/CRC32TableInventoryCheckCalculatorTest.java
+++
b/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/calculator/CRC32TableInventoryCheckCalculatorTest.java
@@ -25,6 +25,7 @@ import
org.apache.shardingsphere.data.pipeline.core.ingest.dumper.inventory.quer
import
org.apache.shardingsphere.data.pipeline.core.metadata.model.PipelineColumnMetaData;
import org.apache.shardingsphere.database.connector.core.type.DatabaseType;
import org.apache.shardingsphere.infra.metadata.database.schema.QualifiedTable;
+import
org.apache.shardingsphere.infra.metadata.identifier.DatabaseIdentifierContextFactory;
import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -70,6 +71,7 @@ class CRC32TableInventoryCheckCalculatorTest {
parameter = new TableInventoryCalculateParameter(pipelineDataSource,
new QualifiedTable(null, "foo_tbl"),
Arrays.asList("foo_col", "bar_col"), uniqueKeys,
QueryType.RANGE_QUERY, null);
when(pipelineDataSource.getDatabaseType()).thenReturn(databaseType);
+
when(pipelineDataSource.getIdentifierContext()).thenReturn(DatabaseIdentifierContextFactory.createDefault());
when(pipelineDataSource.getConnection()).thenReturn(connection);
}
diff --git
a/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/datasource/PipelineDataSourceTest.java
b/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/datasource/PipelineDataSourceTest.java
index ada947f6ec9..133b9093300 100644
---
a/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/datasource/PipelineDataSourceTest.java
+++
b/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/datasource/PipelineDataSourceTest.java
@@ -17,8 +17,12 @@
package org.apache.shardingsphere.data.pipeline.core.datasource;
+import
org.apache.shardingsphere.database.connector.core.metadata.identifier.IdentifierScope;
import org.apache.shardingsphere.database.connector.core.type.DatabaseType;
+import
org.apache.shardingsphere.driver.jdbc.core.datasource.ShardingSphereDataSource;
+import
org.apache.shardingsphere.infra.metadata.identifier.DatabaseIdentifierContext;
import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
+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;
@@ -30,16 +34,20 @@ import org.mockito.quality.Strictness;
import javax.sql.DataSource;
import java.io.PrintWriter;
import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.logging.Logger;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.sameInstance;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
import static org.mockito.Mockito.when;
@ExtendWith(MockitoExtension.class)
@@ -121,4 +129,56 @@ class PipelineDataSourceTest {
pipelineDataSource.close();
assertFalse(pipelineDataSource.isClosed());
}
+
+ @Test
+ void assertIdentifierContextIsResolvedLazilyAndReused() throws
SQLException {
+ DataSource nativeDataSource = mockNativeDataSource(1);
+ PipelineDataSource pipelineDataSource = new
PipelineDataSource(nativeDataSource,
TypedSPILoader.getService(DatabaseType.class, "MySQL"));
+ verifyNoInteractions(nativeDataSource);
+ DatabaseIdentifierContext actual =
pipelineDataSource.getIdentifierContext();
+ assertThat(pipelineDataSource.getIdentifierContext(),
sameInstance(actual));
+ verify(nativeDataSource).getConnection();
+ }
+
+ private DataSource mockNativeDataSource(final int lowerCaseTableNames)
throws SQLException {
+ DataSource result = mock(DataSource.class);
+ Connection connection = mock(Connection.class);
+ PreparedStatement statement = mock(PreparedStatement.class);
+ ResultSet resultSet = mock(ResultSet.class);
+ when(result.getConnection()).thenReturn(connection);
+ when(connection.prepareStatement("SELECT
@@lower_case_table_names")).thenReturn(statement);
+ when(statement.executeQuery()).thenReturn(resultSet);
+ when(resultSet.next()).thenReturn(true);
+ when(resultSet.getInt(1)).thenReturn(lowerCaseTableNames);
+ return result;
+ }
+
+ @Test
+ void assertIdentifierContextsAreIsolatedByDataSource() throws SQLException
{
+ DatabaseType databaseType =
TypedSPILoader.getService(DatabaseType.class, "MySQL");
+ PipelineDataSource preservingDataSource = new
PipelineDataSource(mockNativeDataSource(0), databaseType);
+ PipelineDataSource foldingDataSource = new
PipelineDataSource(mockNativeDataSource(1), databaseType);
+
assertThat(preservingDataSource.getIdentifierContext().normalizeStorage(IdentifierScope.TABLE,
new IdentifierValue("T_User")), is("T_User"));
+
assertThat(foldingDataSource.getIdentifierContext().normalizeStorage(IdentifierScope.TABLE,
new IdentifierValue("T_User")), is("t_user"));
+ }
+
+ @Test
+ void assertEmbeddedDataSourceUsesProtocolWithoutStorageConnection() {
+ ShardingSphereDataSource embeddedDataSource =
mock(ShardingSphereDataSource.class);
+ PipelineDataSource pipelineDataSource = new
PipelineDataSource(embeddedDataSource,
TypedSPILoader.getService(DatabaseType.class, "MySQL"));
+
assertThat(pipelineDataSource.getIdentifierContext().normalizeStorage(IdentifierScope.TABLE,
new IdentifierValue("T_User")), is("t_user"));
+ verifyNoInteractions(embeddedDataSource);
+ }
+
+ @Test
+ void assertIdentifierContextRetainsProviderFallback() throws SQLException {
+ DataSource unavailableDataSource = mock(DataSource.class);
+ when(unavailableDataSource.getConnection()).thenThrow(new
SQLException("unavailable"));
+ PipelineDataSource pipelineDataSource = new
PipelineDataSource(unavailableDataSource,
TypedSPILoader.getService(DatabaseType.class, "MySQL"));
+ DatabaseIdentifierContext actual =
pipelineDataSource.getIdentifierContext();
+ assertThat(actual.normalizeStorage(IdentifierScope.TABLE, new
IdentifierValue("T_User")), is("t_user"));
+ assertThat(pipelineDataSource.getIdentifierContext(),
sameInstance(actual));
+ verify(unavailableDataSource).getConnection();
+ }
+
}
diff --git
a/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/segment/PipelineSQLSegmentBuilderTest.java
b/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/segment/PipelineSQLSegmentBuilderTest.java
index 2932375b12f..8370c0af19b 100644
---
a/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/segment/PipelineSQLSegmentBuilderTest.java
+++
b/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/segment/PipelineSQLSegmentBuilderTest.java
@@ -17,28 +17,43 @@
package org.apache.shardingsphere.data.pipeline.core.sqlbuilder.segment;
+import
org.apache.shardingsphere.database.connector.core.metadata.identifier.IdentifierCasePolicyFactory;
+import
org.apache.shardingsphere.database.connector.core.metadata.identifier.IdentifierScope;
import org.apache.shardingsphere.database.connector.core.type.DatabaseType;
import org.apache.shardingsphere.infra.metadata.database.schema.QualifiedTable;
+import
org.apache.shardingsphere.infra.metadata.identifier.DatabaseIdentifierContext;
+import
org.apache.shardingsphere.infra.metadata.identifier.IdentifierCasePolicyResolver;
import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.util.stream.Stream;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.is;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
class PipelineSQLSegmentBuilderTest {
- private final PipelineSQLSegmentBuilder mysqlBuilder = new
PipelineSQLSegmentBuilder(TypedSPILoader.getService(DatabaseType.class,
"MySQL"));
+ private final PipelineSQLSegmentBuilder mysqlBuilder = new
PipelineSQLSegmentBuilder(TypedSPILoader.getService(DatabaseType.class,
"MySQL"),
+ new
DatabaseIdentifierContext(IdentifierCasePolicyResolver.resolveProtocol(TypedSPILoader.getService(DatabaseType.class,
"MySQL"))));
- private final PipelineSQLSegmentBuilder postgresqlBuilder = new
PipelineSQLSegmentBuilder(TypedSPILoader.getService(DatabaseType.class,
"PostgreSQL"));
+ private final PipelineSQLSegmentBuilder postgresqlBuilder = new
PipelineSQLSegmentBuilder(TypedSPILoader.getService(DatabaseType.class,
"PostgreSQL"),
+ new
DatabaseIdentifierContext(IdentifierCasePolicyResolver.resolveProtocol(TypedSPILoader.getService(DatabaseType.class,
"PostgreSQL"))));
@Test
void assertGetEscapedIdentifier() {
- assertThat(mysqlBuilder.getEscapedIdentifier("SELECT"),
is("`SELECT`"));
+ assertThat(mysqlBuilder.getEscapedIdentifier(IdentifierScope.COLUMN,
"SELECT"), is("`SELECT`"));
}
@Test
void assertGetUnescapedIdentifier() {
- assertThat(mysqlBuilder.getEscapedIdentifier("SELECT1"),
is("`SELECT1`"));
+ assertThat(mysqlBuilder.getEscapedIdentifier(IdentifierScope.COLUMN,
"SELECT1"), is("`SELECT1`"));
}
@Test
@@ -70,4 +85,37 @@ class PipelineSQLSegmentBuilderTest {
assertThat(postgresqlBuilder.getQualifiedTableName(null, "foo_tbl"),
is("\"foo_tbl\""));
assertThat(postgresqlBuilder.getQualifiedTableName(new
QualifiedTable(null, "foo_tbl")), is("\"foo_tbl\""));
}
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("getScopedIdentifiers")
+ void assertGetEscapedIdentifierUsesScope(final String name, final
IdentifierScope scope, final String identifier, final String expected) {
+ DatabaseIdentifierContext context =
mock(DatabaseIdentifierContext.class);
+ when(context.normalizeStorage(eq(scope), any())).thenReturn(expected);
+ PipelineSQLSegmentBuilder builder = new
PipelineSQLSegmentBuilder(TypedSPILoader.getService(DatabaseType.class,
"PostgreSQL"), context);
+ assertThat(builder.getEscapedIdentifier(scope, identifier), is("\"" +
expected + "\""));
+ }
+
+ private static Stream<Arguments> getScopedIdentifiers() {
+ return Stream.of(Arguments.of("schema", IdentifierScope.SCHEMA,
"Tenant", "TENANT"),
+ Arguments.of("table", IdentifierScope.TABLE, "Orders",
"orders"), Arguments.of("column", IdentifierScope.COLUMN, "OrderId",
"OrderId"));
+ }
+
+ @Test
+ void assertGetEscapedIdentifierPreservesQuotedCase() {
+ PipelineSQLSegmentBuilder builder = new
PipelineSQLSegmentBuilder(TypedSPILoader.getService(DatabaseType.class,
"PostgreSQL"),
+ new
DatabaseIdentifierContext(IdentifierCasePolicyFactory.newLowerCasePolicySet()));
+ assertThat(builder.getEscapedIdentifier(IdentifierScope.COLUMN,
"\"OrderId\""), is("\"OrderId\""));
+ }
+
+ @Test
+ void assertGetEscapedIdentifierPreservesWildcardWithoutContext() {
+ PipelineSQLSegmentBuilder builder = new
PipelineSQLSegmentBuilder(TypedSPILoader.getService(DatabaseType.class,
"PostgreSQL"));
+ assertThat(builder.getEscapedIdentifier(IdentifierScope.COLUMN, "*"),
is("*"));
+ }
+
+ @Test
+ void assertGetQualifiedTableNameWithEmptySchema() {
+ assertThat(postgresqlBuilder.getQualifiedTableName("", "Orders"),
is("\"orders\""));
+ }
+
}
diff --git
a/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/sql/PipelineDataConsistencyCalculateSQLBuilderTest.java
b/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/sql/PipelineDataConsistencyCalculateSQLBuilderTest.java
index 536093c6180..ede411d9f80 100644
---
a/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/sql/PipelineDataConsistencyCalculateSQLBuilderTest.java
+++
b/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/sql/PipelineDataConsistencyCalculateSQLBuilderTest.java
@@ -19,6 +19,8 @@ package
org.apache.shardingsphere.data.pipeline.core.sqlbuilder.sql;
import org.apache.shardingsphere.database.connector.core.type.DatabaseType;
import org.apache.shardingsphere.infra.metadata.database.schema.QualifiedTable;
+import
org.apache.shardingsphere.infra.metadata.identifier.DatabaseIdentifierContext;
+import
org.apache.shardingsphere.infra.metadata.identifier.IdentifierCasePolicyResolver;
import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
import org.junit.jupiter.api.Test;
@@ -29,7 +31,8 @@ import static org.hamcrest.Matchers.is;
class PipelineDataConsistencyCalculateSQLBuilderTest {
- private final PipelineDataConsistencyCalculateSQLBuilder sqlBuilder = new
PipelineDataConsistencyCalculateSQLBuilder(TypedSPILoader.getService(DatabaseType.class,
"FIXTURE"));
+ private final PipelineDataConsistencyCalculateSQLBuilder sqlBuilder = new
PipelineDataConsistencyCalculateSQLBuilder(TypedSPILoader.getService(DatabaseType.class,
"FIXTURE"),
+ new
DatabaseIdentifierContext(IdentifierCasePolicyResolver.resolveProtocol(TypedSPILoader.getService(DatabaseType.class,
"FIXTURE"))));
@Test
void assertBuildCRC32SQL() {
diff --git
a/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/sql/PipelineImportSQLBuilderTest.java
b/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/sql/PipelineImportSQLBuilderTest.java
index 1044d402b03..7de8831b2fe 100644
---
a/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/sql/PipelineImportSQLBuilderTest.java
+++
b/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/sql/PipelineImportSQLBuilderTest.java
@@ -23,19 +23,40 @@ import
org.apache.shardingsphere.data.pipeline.core.ingest.record.Column;
import org.apache.shardingsphere.data.pipeline.core.ingest.record.DataRecord;
import org.apache.shardingsphere.data.pipeline.core.ingest.record.NormalColumn;
import org.apache.shardingsphere.data.pipeline.core.ingest.record.RecordUtils;
+import
org.apache.shardingsphere.data.pipeline.core.sqlbuilder.dialect.DialectPipelineSQLBuilder;
+import
org.apache.shardingsphere.database.connector.core.metadata.database.metadata.DialectDatabaseMetaData;
+import
org.apache.shardingsphere.database.connector.core.metadata.identifier.IdentifierCasePolicyFactory;
+import
org.apache.shardingsphere.database.connector.core.spi.DatabaseTypedSPILoader;
import org.apache.shardingsphere.database.connector.core.type.DatabaseType;
+import
org.apache.shardingsphere.infra.metadata.identifier.DatabaseIdentifierContext;
+import
org.apache.shardingsphere.infra.metadata.identifier.IdentifierCasePolicyResolver;
import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
+import
org.apache.shardingsphere.test.infra.framework.extension.mock.AutoMockExtension;
+import
org.apache.shardingsphere.test.infra.framework.extension.mock.StaticMockSettings;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
import java.util.Collection;
import java.util.Collections;
+import java.util.stream.Stream;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.is;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+@ExtendWith(AutoMockExtension.class)
+@StaticMockSettings(DatabaseTypedSPILoader.class)
class PipelineImportSQLBuilderTest {
- private final PipelineImportSQLBuilder sqlBuilder = new
PipelineImportSQLBuilder(TypedSPILoader.getService(DatabaseType.class,
"FIXTURE"));
+ private final PipelineImportSQLBuilder sqlBuilder = new
PipelineImportSQLBuilder(TypedSPILoader.getService(DatabaseType.class,
"FIXTURE"),
+ new
DatabaseIdentifierContext(IdentifierCasePolicyResolver.resolveProtocol(TypedSPILoader.getService(DatabaseType.class,
"FIXTURE"))), false);
+
+ private final PipelineImportSQLBuilder actualTableSQLBuilder = new
PipelineImportSQLBuilder(TypedSPILoader.getService(DatabaseType.class,
"FIXTURE"),
+ new
DatabaseIdentifierContext(IdentifierCasePolicyFactory.newLowerCasePolicySet()),
true);
@Test
void assertBuildInsertSQL() {
@@ -102,4 +123,54 @@ class PipelineImportSQLBuilderTest {
result.addColumn(new NormalColumn("foo_col", "", true, false));
return result;
}
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("getActualTableSchemas")
+ void assertBuildInsertSQLWithActualTableSchema(final String name, final
String schemaName, final String expectedTableName) {
+ DatabaseType databaseType =
TypedSPILoader.getService(DatabaseType.class, "PostgreSQL");
+ when((Object)
DatabaseTypedSPILoader.getService(DialectDatabaseMetaData.class, databaseType))
+
.thenReturn(TypedSPILoader.getService(DialectDatabaseMetaData.class,
databaseType));
+ when((Object)
DatabaseTypedSPILoader.getService(DialectPipelineSQLBuilder.class,
databaseType)).thenReturn(mock(DialectPipelineSQLBuilder.class));
+ PipelineImportSQLBuilder builder = new
PipelineImportSQLBuilder(TypedSPILoader.getService(DatabaseType.class,
"PostgreSQL"),
+ new
DatabaseIdentifierContext(IdentifierCasePolicyFactory.newLowerCasePolicySet()),
true);
+ DataRecord dataRecord = new
DataRecord(PipelineSQLOperationType.INSERT, "T_Order", new
IngestPlaceholderPosition(), 1);
+ dataRecord.addColumn(new NormalColumn("Value", "foo", true, false));
+ assertThat(builder.buildInsertSQL(schemaName, dataRecord), is("INSERT
INTO " + expectedTableName + "(\"value\") VALUES(?)"));
+ }
+
+ private static Stream<Arguments> getActualTableSchemas() {
+ return Stream.of(
+ Arguments.of("unquoted schema", "Foo_SCHEMA",
"\"foo_schema\".\"T_Order\""),
+ Arguments.of("quoted schema", "\"Foo_SCHEMA\"",
"\"Foo_SCHEMA\".\"T_Order\""),
+ Arguments.of("empty schema", "", "\"T_Order\""),
+ Arguments.of("absent schema", null, "\"T_Order\""));
+ }
+
+ @Test
+ void assertBuildInsertSQLWithActualTableName() {
+ String actual = actualTableSQLBuilder.buildInsertSQL("UnusedSchema",
createDataRecordWithActualTableName());
+ assertThat(actual, is("INSERT INTO T_Order(id,value) VALUES(?,?)"));
+ }
+
+ private DataRecord createDataRecordWithActualTableName() {
+ DataRecord result = new DataRecord(PipelineSQLOperationType.INSERT,
"T_Order", new IngestPlaceholderPosition(), 2);
+ result.addColumn(new NormalColumn("Id", 1, false, true));
+ result.addColumn(new NormalColumn("Value", "foo", true, false));
+ return result;
+ }
+
+ @Test
+ void assertBuildUpdateSQLWithActualTableName() {
+ DataRecord dataRecord = createDataRecordWithActualTableName();
+ String actual = actualTableSQLBuilder.buildUpdateSQL(null, dataRecord,
Collections.singleton(dataRecord.getColumns().get(0)));
+ assertThat(actual, is("UPDATE T_Order SET value = ? WHERE id = ?"));
+ }
+
+ @Test
+ void assertBuildDeleteSQLWithActualTableName() {
+ DataRecord dataRecord = createDataRecordWithActualTableName();
+ String actual = actualTableSQLBuilder.buildDeleteSQL(null, dataRecord,
Collections.singleton(dataRecord.getColumns().get(0)));
+ assertThat(actual, is("DELETE FROM T_Order WHERE id = ?"));
+ }
+
}
diff --git
a/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/sql/PipelinePrepareSQLBuilderTest.java
b/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/sql/PipelinePrepareSQLBuilderTest.java
index 650a25b4008..8eb29056fba 100644
---
a/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/sql/PipelinePrepareSQLBuilderTest.java
+++
b/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/sql/PipelinePrepareSQLBuilderTest.java
@@ -17,17 +17,30 @@
package org.apache.shardingsphere.data.pipeline.core.sqlbuilder.sql;
+import
org.apache.shardingsphere.data.pipeline.core.sqlbuilder.dialect.DialectPipelineSQLBuilder;
+import
org.apache.shardingsphere.database.connector.core.metadata.database.metadata.DialectDatabaseMetaData;
+import
org.apache.shardingsphere.database.connector.core.spi.DatabaseTypedSPILoader;
import org.apache.shardingsphere.database.connector.core.type.DatabaseType;
+import
org.apache.shardingsphere.infra.metadata.identifier.DatabaseIdentifierContext;
+import
org.apache.shardingsphere.infra.metadata.identifier.IdentifierCasePolicyResolver;
import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
+import
org.apache.shardingsphere.test.infra.framework.extension.mock.AutoMockExtension;
+import
org.apache.shardingsphere.test.infra.framework.extension.mock.StaticMockSettings;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.is;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+@ExtendWith(AutoMockExtension.class)
+@StaticMockSettings(DatabaseTypedSPILoader.class)
class PipelinePrepareSQLBuilderTest {
- private final PipelinePrepareSQLBuilder sqlBuilder = new
PipelinePrepareSQLBuilder(TypedSPILoader.getService(DatabaseType.class,
"FIXTURE"));
+ private final PipelinePrepareSQLBuilder sqlBuilder = new
PipelinePrepareSQLBuilder(TypedSPILoader.getService(DatabaseType.class,
"FIXTURE"),
+ new
DatabaseIdentifierContext(IdentifierCasePolicyResolver.resolveProtocol(TypedSPILoader.getService(DatabaseType.class,
"FIXTURE"))));
@Test
void assertBuildCreateSchemaSQL() {
@@ -39,6 +52,16 @@ class PipelinePrepareSQLBuilderTest {
assertThat(sqlBuilder.buildDropSQL("foo_schema", "foo_tbl"), is("DROP
TABLE IF EXISTS foo_tbl"));
}
+ @Test
+ void assertBuildDropSQLWithActualIdentifiers() {
+ DatabaseType databaseType =
TypedSPILoader.getService(DatabaseType.class, "PostgreSQL");
+ when((Object)
DatabaseTypedSPILoader.getService(DialectDatabaseMetaData.class, databaseType))
+
.thenReturn(TypedSPILoader.getService(DialectDatabaseMetaData.class,
databaseType));
+ when((Object)
DatabaseTypedSPILoader.getService(DialectPipelineSQLBuilder.class,
databaseType)).thenReturn(mock(DialectPipelineSQLBuilder.class));
+ PipelinePrepareSQLBuilder builder = new
PipelinePrepareSQLBuilder(TypedSPILoader.getService(DatabaseType.class,
"PostgreSQL"));
+ assertThat(builder.buildDropSQL("Foo_SCHEMA", "T_Order"), is("DROP
TABLE IF EXISTS \"Foo_SCHEMA\".\"T_Order\""));
+ }
+
@Test
void assertBuildCountSQL() {
assertThat(sqlBuilder.buildCountSQL("foo_schema", "foo_tbl"),
is("SELECT COUNT(*) FROM foo_tbl"));
diff --git
a/kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/sqlbuilder/MySQLPipelineSQLBuilder.java
b/kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/sqlbuilder/MySQLPipelineSQLBuilder.java
index 966cf49d134..fd2e786e5a6 100644
---
a/kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/sqlbuilder/MySQLPipelineSQLBuilder.java
+++
b/kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/sqlbuilder/MySQLPipelineSQLBuilder.java
@@ -22,6 +22,7 @@ import
org.apache.shardingsphere.data.pipeline.core.ingest.record.Column;
import org.apache.shardingsphere.data.pipeline.core.ingest.record.DataRecord;
import
org.apache.shardingsphere.data.pipeline.core.sqlbuilder.dialect.DialectPipelineSQLBuilder;
import
org.apache.shardingsphere.data.pipeline.core.sqlbuilder.segment.PipelineSQLSegmentBuilder;
+import
org.apache.shardingsphere.database.connector.core.metadata.identifier.IdentifierScope;
import javax.sql.DataSource;
import java.sql.Connection;
@@ -38,12 +39,12 @@ import java.util.Optional;
public final class MySQLPipelineSQLBuilder implements
DialectPipelineSQLBuilder {
@Override
- public Optional<String> buildInsertOnDuplicateClause(final DataRecord
dataRecord) {
+ public Optional<String> buildInsertOnDuplicateClause(final DataRecord
dataRecord, final PipelineSQLSegmentBuilder sqlSegmentBuilder) {
StringBuilder result = new StringBuilder("ON DUPLICATE KEY UPDATE ");
- PipelineSQLSegmentBuilder sqlSegmentBuilder = new
PipelineSQLSegmentBuilder(getType());
for (int i = 0; i < dataRecord.getColumnCount(); i++) {
Column column = dataRecord.getColumn(i);
-
result.append(sqlSegmentBuilder.getEscapedIdentifier(column.getName())).append("=VALUES(").append(sqlSegmentBuilder.getEscapedIdentifier(column.getName())).append("),");
+
result.append(sqlSegmentBuilder.getEscapedIdentifier(IdentifierScope.COLUMN,
column.getName())).append("=VALUES(")
+
.append(sqlSegmentBuilder.getEscapedIdentifier(IdentifierScope.COLUMN,
column.getName())).append("),");
}
result.setLength(result.length() - 1);
return Optional.of(result.toString());
diff --git
a/kernel/data-pipeline/dialect/mysql/src/test/java/org/apache/shardingsphere/data/pipeline/mysql/sqlbuilder/MySQLPipelineSQLBuilderTest.java
b/kernel/data-pipeline/dialect/mysql/src/test/java/org/apache/shardingsphere/data/pipeline/mysql/sqlbuilder/MySQLPipelineSQLBuilderTest.java
index 22c1402a461..2fa0f7618cf 100644
---
a/kernel/data-pipeline/dialect/mysql/src/test/java/org/apache/shardingsphere/data/pipeline/mysql/sqlbuilder/MySQLPipelineSQLBuilderTest.java
+++
b/kernel/data-pipeline/dialect/mysql/src/test/java/org/apache/shardingsphere/data/pipeline/mysql/sqlbuilder/MySQLPipelineSQLBuilderTest.java
@@ -23,8 +23,12 @@ import
org.apache.shardingsphere.data.pipeline.core.ingest.position.type.placeho
import org.apache.shardingsphere.data.pipeline.core.ingest.record.DataRecord;
import org.apache.shardingsphere.data.pipeline.core.ingest.record.NormalColumn;
import
org.apache.shardingsphere.data.pipeline.core.sqlbuilder.dialect.DialectPipelineSQLBuilder;
+import
org.apache.shardingsphere.data.pipeline.core.sqlbuilder.segment.PipelineSQLSegmentBuilder;
+import
org.apache.shardingsphere.database.connector.core.metadata.identifier.IdentifierCasePolicyFactory;
import
org.apache.shardingsphere.database.connector.core.spi.DatabaseTypedSPILoader;
import org.apache.shardingsphere.database.connector.core.type.DatabaseType;
+import
org.apache.shardingsphere.infra.metadata.identifier.DatabaseIdentifierContext;
+import
org.apache.shardingsphere.infra.metadata.identifier.IdentifierCasePolicyResolver;
import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
import org.apache.shardingsphere.test.infra.fixture.jdbc.MockedDataSource;
import org.junit.jupiter.api.Test;
@@ -45,11 +49,14 @@ import static org.mockito.Mockito.when;
class MySQLPipelineSQLBuilderTest {
- private final DialectPipelineSQLBuilder sqlBuilder =
DatabaseTypedSPILoader.getService(DialectPipelineSQLBuilder.class,
TypedSPILoader.getService(DatabaseType.class, "MySQL"));
+ private final DatabaseType databaseType =
TypedSPILoader.getService(DatabaseType.class, "MySQL");
+
+ private final DialectPipelineSQLBuilder sqlBuilder =
DatabaseTypedSPILoader.getService(DialectPipelineSQLBuilder.class,
databaseType);
@Test
void assertBuildInsertSQLOnDuplicateClause() {
- Optional<String> actual =
sqlBuilder.buildInsertOnDuplicateClause(createDataRecord());
+ Optional<String> actual =
sqlBuilder.buildInsertOnDuplicateClause(createDataRecord(),
+ new PipelineSQLSegmentBuilder(databaseType, new
DatabaseIdentifierContext(IdentifierCasePolicyResolver.resolveProtocol(databaseType))));
assertTrue(actual.isPresent());
assertThat(actual.get(), is("ON DUPLICATE KEY UPDATE
`id`=VALUES(`id`),`sc`=VALUES(`sc`),`c1`=VALUES(`c1`),`c2`=VALUES(`c2`),`c3`=VALUES(`c3`)"));
}
@@ -64,6 +71,13 @@ class MySQLPipelineSQLBuilderTest {
return result;
}
+ @Test
+ void assertBuildInsertOnDuplicateClauseWithStorageColumnPolicy() {
+ PipelineSQLSegmentBuilder segmentBuilder = new
PipelineSQLSegmentBuilder(databaseType, new
DatabaseIdentifierContext(IdentifierCasePolicyFactory.newUpperCasePolicySet()));
+ Optional<String> actual =
sqlBuilder.buildInsertOnDuplicateClause(createDataRecord(), segmentBuilder);
+ assertThat(actual, is(Optional.of("ON DUPLICATE KEY UPDATE
`ID`=VALUES(`ID`),`SC`=VALUES(`SC`),`C1`=VALUES(`C1`),`C2`=VALUES(`C2`),`C3`=VALUES(`C3`)")));
+ }
+
@Test
void assertBuildCheckEmptyTableSQL() {
assertThat(sqlBuilder.buildCheckEmptyTableSQL("foo_tbl"), is("SELECT *
FROM foo_tbl LIMIT 1"));
diff --git
a/kernel/data-pipeline/dialect/opengauss/src/main/java/org/apache/shardingsphere/data/pipeline/opengauss/sqlbuilder/OpenGaussPipelineSQLBuilder.java
b/kernel/data-pipeline/dialect/opengauss/src/main/java/org/apache/shardingsphere/data/pipeline/opengauss/sqlbuilder/OpenGaussPipelineSQLBuilder.java
index bb6a72ec797..6778e4c77e1 100644
---
a/kernel/data-pipeline/dialect/opengauss/src/main/java/org/apache/shardingsphere/data/pipeline/opengauss/sqlbuilder/OpenGaussPipelineSQLBuilder.java
+++
b/kernel/data-pipeline/dialect/opengauss/src/main/java/org/apache/shardingsphere/data/pipeline/opengauss/sqlbuilder/OpenGaussPipelineSQLBuilder.java
@@ -22,6 +22,7 @@ import
org.apache.shardingsphere.data.pipeline.core.exception.job.CreateTableSQL
import org.apache.shardingsphere.data.pipeline.core.ingest.record.DataRecord;
import
org.apache.shardingsphere.data.pipeline.core.sqlbuilder.dialect.DialectPipelineSQLBuilder;
import
org.apache.shardingsphere.data.pipeline.core.sqlbuilder.segment.PipelineSQLSegmentBuilder;
+import
org.apache.shardingsphere.database.connector.core.metadata.identifier.IdentifierScope;
import javax.sql.DataSource;
import java.sql.Connection;
@@ -45,11 +46,12 @@ public final class OpenGaussPipelineSQLBuilder implements
DialectPipelineSQLBuil
}
@Override
- public Optional<String> buildInsertOnDuplicateClause(final DataRecord
dataRecord) {
+ public Optional<String> buildInsertOnDuplicateClause(final DataRecord
dataRecord, final PipelineSQLSegmentBuilder sqlSegmentBuilder) {
StringBuilder result = new StringBuilder("ON DUPLICATE KEY UPDATE ");
- PipelineSQLSegmentBuilder sqlSegmentBuilder = new
PipelineSQLSegmentBuilder(getType());
result.append(dataRecord.getColumns().stream()
- .filter(each -> !each.isUniqueKey()).map(each ->
sqlSegmentBuilder.getEscapedIdentifier(each.getName()) + "=EXCLUDED." +
sqlSegmentBuilder.getEscapedIdentifier(each.getName()))
+ .filter(each -> !each.isUniqueKey())
+ .map(each ->
sqlSegmentBuilder.getEscapedIdentifier(IdentifierScope.COLUMN, each.getName())
+ "=EXCLUDED."
+ +
sqlSegmentBuilder.getEscapedIdentifier(IdentifierScope.COLUMN, each.getName()))
.collect(Collectors.joining(",")));
return Optional.of(result.toString());
}
diff --git
a/kernel/data-pipeline/dialect/opengauss/src/test/java/org/apache/shardingsphere/data/pipeline/opengauss/sqlbuilder/OpenGaussPipelineSQLBuilderTest.java
b/kernel/data-pipeline/dialect/opengauss/src/test/java/org/apache/shardingsphere/data/pipeline/opengauss/sqlbuilder/OpenGaussPipelineSQLBuilderTest.java
index 4a46fadf865..a83504b19ef 100644
---
a/kernel/data-pipeline/dialect/opengauss/src/test/java/org/apache/shardingsphere/data/pipeline/opengauss/sqlbuilder/OpenGaussPipelineSQLBuilderTest.java
+++
b/kernel/data-pipeline/dialect/opengauss/src/test/java/org/apache/shardingsphere/data/pipeline/opengauss/sqlbuilder/OpenGaussPipelineSQLBuilderTest.java
@@ -23,8 +23,12 @@ import
org.apache.shardingsphere.data.pipeline.core.ingest.position.type.placeho
import org.apache.shardingsphere.data.pipeline.core.ingest.record.DataRecord;
import org.apache.shardingsphere.data.pipeline.core.ingest.record.NormalColumn;
import
org.apache.shardingsphere.data.pipeline.core.sqlbuilder.dialect.DialectPipelineSQLBuilder;
+import
org.apache.shardingsphere.data.pipeline.core.sqlbuilder.segment.PipelineSQLSegmentBuilder;
+import
org.apache.shardingsphere.database.connector.core.metadata.identifier.IdentifierCasePolicyFactory;
import
org.apache.shardingsphere.database.connector.core.spi.DatabaseTypedSPILoader;
import org.apache.shardingsphere.database.connector.core.type.DatabaseType;
+import
org.apache.shardingsphere.infra.metadata.identifier.DatabaseIdentifierContext;
+import
org.apache.shardingsphere.infra.metadata.identifier.IdentifierCasePolicyResolver;
import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
import org.apache.shardingsphere.test.infra.fixture.jdbc.MockedDataSource;
import org.junit.jupiter.api.Test;
@@ -45,7 +49,9 @@ import static org.mockito.Mockito.when;
class OpenGaussPipelineSQLBuilderTest {
- private final DialectPipelineSQLBuilder sqlBuilder =
DatabaseTypedSPILoader.getService(DialectPipelineSQLBuilder.class,
TypedSPILoader.getService(DatabaseType.class, "openGauss"));
+ private final DatabaseType databaseType =
TypedSPILoader.getService(DatabaseType.class, "openGauss");
+
+ private final DialectPipelineSQLBuilder sqlBuilder =
DatabaseTypedSPILoader.getService(DialectPipelineSQLBuilder.class,
databaseType);
@Test
void assertBuildCreateSchemaSQL() {
@@ -54,7 +60,8 @@ class OpenGaussPipelineSQLBuilderTest {
@Test
void assertBuildInsertOnDuplicateClause() {
- Optional<String> actual =
sqlBuilder.buildInsertOnDuplicateClause(createDataRecord());
+ Optional<String> actual =
sqlBuilder.buildInsertOnDuplicateClause(createDataRecord(),
+ new PipelineSQLSegmentBuilder(databaseType, new
DatabaseIdentifierContext(IdentifierCasePolicyResolver.resolveProtocol(databaseType))));
assertTrue(actual.isPresent());
assertThat(actual.get(), is("ON DUPLICATE KEY UPDATE
\"c0\"=EXCLUDED.\"c0\",\"c1\"=EXCLUDED.\"c1\",\"c2\"=EXCLUDED.\"c2\",\"c3\"=EXCLUDED.\"c3\""));
}
@@ -69,6 +76,13 @@ class OpenGaussPipelineSQLBuilderTest {
return result;
}
+ @Test
+ void assertBuildInsertOnDuplicateClauseWithStorageColumnPolicy() {
+ PipelineSQLSegmentBuilder segmentBuilder = new
PipelineSQLSegmentBuilder(databaseType, new
DatabaseIdentifierContext(IdentifierCasePolicyFactory.newUpperCasePolicySet()));
+ Optional<String> actual =
sqlBuilder.buildInsertOnDuplicateClause(createDataRecord(), segmentBuilder);
+ assertThat(actual, is(Optional.of("ON DUPLICATE KEY UPDATE
\"C0\"=EXCLUDED.\"C0\",\"C1\"=EXCLUDED.\"C1\",\"C2\"=EXCLUDED.\"C2\",\"C3\"=EXCLUDED.\"C3\"")));
+ }
+
@Test
void assertBuildCheckEmptyTableSQL() {
assertThat(sqlBuilder.buildCheckEmptyTableSQL("foo_tbl"), is("SELECT *
FROM foo_tbl LIMIT 1"));
diff --git
a/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/sqlbuilder/PostgreSQLPipelineSQLBuilder.java
b/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/sqlbuilder/PostgreSQLPipelineSQLBuilder.java
index 963c88c4cfb..aa7eed4d35f 100644
---
a/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/sqlbuilder/PostgreSQLPipelineSQLBuilder.java
+++
b/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/sqlbuilder/PostgreSQLPipelineSQLBuilder.java
@@ -26,6 +26,7 @@ import
org.apache.shardingsphere.data.pipeline.postgresql.sqlbuilder.ddl.constra
import
org.apache.shardingsphere.data.pipeline.postgresql.sqlbuilder.ddl.index.PostgreSQLIndexSQLGenerator;
import
org.apache.shardingsphere.data.pipeline.postgresql.sqlbuilder.ddl.table.PostgreSQLTablePropertiesLoader;
import
org.apache.shardingsphere.data.pipeline.postgresql.sqlbuilder.template.PostgreSQLPipelineFreemarkerManager;
+import
org.apache.shardingsphere.database.connector.core.metadata.identifier.IdentifierScope;
import javax.sql.DataSource;
import java.sql.Connection;
@@ -47,17 +48,19 @@ public final class PostgreSQLPipelineSQLBuilder implements
DialectPipelineSQLBui
}
@Override
- public Optional<String> buildInsertOnDuplicateClause(final DataRecord
dataRecord) {
+ public Optional<String> buildInsertOnDuplicateClause(final DataRecord
dataRecord, final PipelineSQLSegmentBuilder sqlSegmentBuilder) {
// TODO without unique key, job has been interrupted, which may lead
to data duplication
if (dataRecord.getUniqueKeyValue().isEmpty()) {
return Optional.empty();
}
StringBuilder result = new StringBuilder("ON CONFLICT (");
- PipelineSQLSegmentBuilder sqlSegmentBuilder = new
PipelineSQLSegmentBuilder(getType());
-
result.append(dataRecord.getColumns().stream().filter(Column::isUniqueKey).map(each
->
sqlSegmentBuilder.getEscapedIdentifier(each.getName())).collect(Collectors.joining(",")));
+
result.append(dataRecord.getColumns().stream().filter(Column::isUniqueKey).map(each
-> sqlSegmentBuilder.getEscapedIdentifier(IdentifierScope.COLUMN,
each.getName()))
+ .collect(Collectors.joining(",")));
result.append(") DO UPDATE SET ");
result.append(dataRecord.getColumns().stream()
- .filter(each -> !each.isUniqueKey()).map(each ->
sqlSegmentBuilder.getEscapedIdentifier(each.getName()) + "=EXCLUDED." +
sqlSegmentBuilder.getEscapedIdentifier(each.getName()))
+ .filter(each -> !each.isUniqueKey())
+ .map(each ->
sqlSegmentBuilder.getEscapedIdentifier(IdentifierScope.COLUMN, each.getName())
+ "=EXCLUDED."
+ +
sqlSegmentBuilder.getEscapedIdentifier(IdentifierScope.COLUMN, each.getName()))
.collect(Collectors.joining(",")));
return Optional.of(result.toString());
}
diff --git
a/kernel/data-pipeline/dialect/postgresql/src/test/java/org/apache/shardingsphere/data/pipeline/postgresql/sqlbuilder/PipelinePrepareSQLBuilderTest.java
b/kernel/data-pipeline/dialect/postgresql/src/test/java/org/apache/shardingsphere/data/pipeline/postgresql/sqlbuilder/PipelinePrepareSQLBuilderTest.java
index 54cb2b97ca7..f8c343d33af 100644
---
a/kernel/data-pipeline/dialect/postgresql/src/test/java/org/apache/shardingsphere/data/pipeline/postgresql/sqlbuilder/PipelinePrepareSQLBuilderTest.java
+++
b/kernel/data-pipeline/dialect/postgresql/src/test/java/org/apache/shardingsphere/data/pipeline/postgresql/sqlbuilder/PipelinePrepareSQLBuilderTest.java
@@ -19,6 +19,8 @@ package
org.apache.shardingsphere.data.pipeline.postgresql.sqlbuilder;
import
org.apache.shardingsphere.data.pipeline.core.sqlbuilder.sql.PipelinePrepareSQLBuilder;
import org.apache.shardingsphere.database.connector.core.type.DatabaseType;
+import
org.apache.shardingsphere.infra.metadata.identifier.DatabaseIdentifierContext;
+import
org.apache.shardingsphere.infra.metadata.identifier.IdentifierCasePolicyResolver;
import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
import org.junit.jupiter.api.Test;
@@ -29,7 +31,8 @@ import static org.hamcrest.Matchers.is;
class PipelinePrepareSQLBuilderTest {
- private final PipelinePrepareSQLBuilder sqlBuilder = new
PipelinePrepareSQLBuilder(TypedSPILoader.getService(DatabaseType.class,
"PostgreSQL"));
+ private final PipelinePrepareSQLBuilder sqlBuilder = new
PipelinePrepareSQLBuilder(TypedSPILoader.getService(DatabaseType.class,
"PostgreSQL"),
+ new
DatabaseIdentifierContext(IdentifierCasePolicyResolver.resolveProtocol(TypedSPILoader.getService(DatabaseType.class,
"PostgreSQL"))));
@Test
void assertBuildCountSQLWithActualIdentifiers() {
diff --git
a/kernel/data-pipeline/dialect/postgresql/src/test/java/org/apache/shardingsphere/data/pipeline/postgresql/sqlbuilder/PostgreSQLPipelineSQLBuilderTest.java
b/kernel/data-pipeline/dialect/postgresql/src/test/java/org/apache/shardingsphere/data/pipeline/postgresql/sqlbuilder/PostgreSQLPipelineSQLBuilderTest.java
index a3735175c77..cc73dabaaa4 100644
---
a/kernel/data-pipeline/dialect/postgresql/src/test/java/org/apache/shardingsphere/data/pipeline/postgresql/sqlbuilder/PostgreSQLPipelineSQLBuilderTest.java
+++
b/kernel/data-pipeline/dialect/postgresql/src/test/java/org/apache/shardingsphere/data/pipeline/postgresql/sqlbuilder/PostgreSQLPipelineSQLBuilderTest.java
@@ -21,6 +21,7 @@ import
org.apache.shardingsphere.data.pipeline.core.constant.PipelineSQLOperatio
import org.apache.shardingsphere.data.pipeline.core.ingest.record.DataRecord;
import org.apache.shardingsphere.data.pipeline.core.ingest.record.NormalColumn;
import
org.apache.shardingsphere.data.pipeline.core.sqlbuilder.dialect.DialectPipelineSQLBuilder;
+import
org.apache.shardingsphere.data.pipeline.core.sqlbuilder.segment.PipelineSQLSegmentBuilder;
import
org.apache.shardingsphere.data.pipeline.postgresql.ingest.incremental.wal.WALPosition;
import
org.apache.shardingsphere.data.pipeline.postgresql.ingest.incremental.wal.decode.PostgreSQLLogSequenceNumber;
import
org.apache.shardingsphere.data.pipeline.postgresql.sqlbuilder.ddl.column.PostgreSQLColumnPropertiesAppender;
@@ -28,8 +29,11 @@ import
org.apache.shardingsphere.data.pipeline.postgresql.sqlbuilder.ddl.constra
import
org.apache.shardingsphere.data.pipeline.postgresql.sqlbuilder.ddl.index.PostgreSQLIndexSQLGenerator;
import
org.apache.shardingsphere.data.pipeline.postgresql.sqlbuilder.ddl.table.PostgreSQLTablePropertiesLoader;
import
org.apache.shardingsphere.data.pipeline.postgresql.sqlbuilder.template.PostgreSQLPipelineFreemarkerManager;
+import
org.apache.shardingsphere.database.connector.core.metadata.identifier.IdentifierCasePolicyFactory;
import
org.apache.shardingsphere.database.connector.core.spi.DatabaseTypedSPILoader;
import org.apache.shardingsphere.database.connector.core.type.DatabaseType;
+import
org.apache.shardingsphere.infra.metadata.identifier.DatabaseIdentifierContext;
+import
org.apache.shardingsphere.infra.metadata.identifier.IdentifierCasePolicyResolver;
import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
import org.apache.shardingsphere.test.infra.fixture.jdbc.MockedDataSource;
import org.junit.jupiter.api.Test;
@@ -74,13 +78,15 @@ class PostgreSQLPipelineSQLBuilderTest {
@Test
void assertBuildInsertSQLOnDuplicateClauseWithEmptyUniqueKey() {
Optional<String> actual = sqlBuilder.buildInsertOnDuplicateClause(
- new DataRecord(PipelineSQLOperationType.INSERT, "foo_tbl", new
WALPosition(new PostgreSQLLogSequenceNumber(LogSequenceNumber.valueOf(100L))),
2));
+ new DataRecord(PipelineSQLOperationType.INSERT, "foo_tbl", new
WALPosition(new PostgreSQLLogSequenceNumber(LogSequenceNumber.valueOf(100L))),
2),
+ new PipelineSQLSegmentBuilder(databaseType, new
DatabaseIdentifierContext(IdentifierCasePolicyResolver.resolveProtocol(databaseType))));
assertFalse(actual.isPresent());
}
@Test
void assertBuildInsertSQLOnDuplicateClause() {
- Optional<String> actual =
sqlBuilder.buildInsertOnDuplicateClause(createDataRecord());
+ Optional<String> actual =
sqlBuilder.buildInsertOnDuplicateClause(createDataRecord(),
+ new PipelineSQLSegmentBuilder(databaseType, new
DatabaseIdentifierContext(IdentifierCasePolicyResolver.resolveProtocol(databaseType))));
assertTrue(actual.isPresent());
assertThat(actual.get(), is("ON CONFLICT (\"order_id\") DO UPDATE SET
\"user_id\"=EXCLUDED.\"user_id\",\"status\"=EXCLUDED.\"status\""));
}
@@ -93,6 +99,13 @@ class PostgreSQLPipelineSQLBuilderTest {
return result;
}
+ @Test
+ void assertBuildInsertOnDuplicateClauseWithStorageColumnPolicy() {
+ PipelineSQLSegmentBuilder segmentBuilder = new
PipelineSQLSegmentBuilder(databaseType, new
DatabaseIdentifierContext(IdentifierCasePolicyFactory.newUpperCasePolicySet()));
+ Optional<String> actual =
sqlBuilder.buildInsertOnDuplicateClause(createDataRecord(), segmentBuilder);
+ assertThat(actual, is(Optional.of("ON CONFLICT (\"ORDER_ID\") DO
UPDATE SET \"USER_ID\"=EXCLUDED.\"USER_ID\",\"STATUS\"=EXCLUDED.\"STATUS\"")));
+ }
+
@Test
void assertBuildCheckEmptyTableSQL() {
assertThat(sqlBuilder.buildCheckEmptyTableSQL("foo_tbl"), is("SELECT *
FROM foo_tbl LIMIT 1"));
diff --git
a/kernel/data-pipeline/scenario/migration/core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/api/MigrationJobAPI.java
b/kernel/data-pipeline/scenario/migration/core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/api/MigrationJobAPI.java
index 32cc5055c00..a7640709de5 100644
---
a/kernel/data-pipeline/scenario/migration/core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/api/MigrationJobAPI.java
+++
b/kernel/data-pipeline/scenario/migration/core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/api/MigrationJobAPI.java
@@ -45,6 +45,7 @@ import
org.apache.shardingsphere.data.pipeline.scenario.migration.config.yaml.co
import
org.apache.shardingsphere.data.pipeline.scenario.migration.config.yaml.swapper.YamlMigrationJobConfigurationSwapper;
import
org.apache.shardingsphere.database.connector.core.jdbcurl.parser.ConnectionProperties;
import
org.apache.shardingsphere.database.connector.core.jdbcurl.parser.ConnectionPropertiesParser;
+import
org.apache.shardingsphere.database.connector.core.metadata.identifier.IdentifierScope;
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.DatabaseTypeFactory;
@@ -57,6 +58,7 @@ import
org.apache.shardingsphere.infra.exception.kernel.metadata.resource.storag
import
org.apache.shardingsphere.infra.exception.kernel.metadata.rule.EmptyRuleException;
import
org.apache.shardingsphere.infra.metadata.database.ShardingSphereDatabase;
import
org.apache.shardingsphere.infra.metadata.database.resource.unit.StorageUnit;
+import
org.apache.shardingsphere.infra.metadata.identifier.DatabaseIdentifierContext;
import
org.apache.shardingsphere.infra.metadata.identifier.ShardingSphereIdentifier;
import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
import org.apache.shardingsphere.infra.util.json.JsonEngine;
@@ -67,6 +69,7 @@ import
org.apache.shardingsphere.infra.yaml.config.swapper.rule.YamlRuleConfigur
import org.apache.shardingsphere.mode.manager.ContextManager;
import org.apache.shardingsphere.single.constant.SingleTableConstants;
import
org.apache.shardingsphere.single.yaml.config.YamlSingleRuleConfiguration;
+import
org.apache.shardingsphere.sql.parser.statement.core.value.identifier.IdentifierValue;
import java.sql.Connection;
import java.sql.SQLException;
@@ -354,9 +357,11 @@ public final class MigrationJobAPI implements
TransmissionJobAPI {
try (
PipelineDataSource dataSource = new
PipelineDataSource(jobConfig.getTarget());
Connection connection = dataSource.getConnection()) {
+ DatabaseIdentifierContext identifierContext =
dataSource.getIdentifierContext();
for (String each : jobConfig.getTargetTableNames()) {
String targetSchemaName = mapping.getSchemaName(each);
- String sql = pipelineSQLBuilder.buildDropSQL(targetSchemaName,
each);
+ String actualSchemaName = null == targetSchemaName ? null :
identifierContext.normalizeStorage(IdentifierScope.SCHEMA, new
IdentifierValue(targetSchemaName));
+ String sql = pipelineSQLBuilder.buildDropSQL(actualSchemaName,
each);
log.info("cleanTempTableOnRollback, targetSchemaName={},
targetTableName={}, sql={}", targetSchemaName, each, sql);
try (Statement statement = connection.createStatement()) {
statement.execute(sql);
diff --git
a/kernel/data-pipeline/scenario/migration/core/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/migration/api/MigrationJobAPITest.java
b/kernel/data-pipeline/scenario/migration/core/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/migration/api/MigrationJobAPITest.java
index 380b123f7f0..c208bdf925a 100644
---
a/kernel/data-pipeline/scenario/migration/core/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/migration/api/MigrationJobAPITest.java
+++
b/kernel/data-pipeline/scenario/migration/core/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/migration/api/MigrationJobAPITest.java
@@ -18,26 +18,49 @@
package org.apache.shardingsphere.data.pipeline.scenario.migration.api;
import org.apache.shardingsphere.data.pipeline.core.context.PipelineContextKey;
+import
org.apache.shardingsphere.data.pipeline.core.datasource.PipelineDataSource;
import
org.apache.shardingsphere.data.pipeline.core.exception.param.PipelineInvalidParameterException;
+import org.apache.shardingsphere.data.pipeline.core.job.api.PipelineAPIFactory;
+import org.apache.shardingsphere.data.pipeline.core.job.id.PipelineJobIdUtils;
+import
org.apache.shardingsphere.data.pipeline.core.job.service.PipelineJobConfigurationManager;
import
org.apache.shardingsphere.data.pipeline.core.job.service.PipelineJobManager;
import
org.apache.shardingsphere.data.pipeline.core.metadata.PipelineDataSourcePersistService;
+import
org.apache.shardingsphere.data.pipeline.core.sqlbuilder.sql.PipelinePrepareSQLBuilder;
+import
org.apache.shardingsphere.data.pipeline.scenario.migration.MigrationJobId;
+import
org.apache.shardingsphere.data.pipeline.scenario.migration.config.MigrationJobConfiguration;
+import
org.apache.shardingsphere.database.connector.core.metadata.identifier.IdentifierCasePolicyFactory;
import org.apache.shardingsphere.infra.datanode.DataNode;
import org.apache.shardingsphere.infra.instance.metadata.InstanceType;
+import
org.apache.shardingsphere.infra.metadata.identifier.DatabaseIdentifierContext;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+import org.mockito.Answers;
import org.mockito.Mock;
+import org.mockito.MockedConstruction;
+import org.mockito.MockedStatic;
import org.mockito.internal.configuration.plugins.Plugins;
import org.mockito.junit.jupiter.MockitoExtension;
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.sql.Statement;
import java.util.Arrays;
import java.util.Collection;
+import java.util.Collections;
+import java.util.stream.Stream;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.is;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockConstruction;
+import static org.mockito.Mockito.mockStatic;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -97,4 +120,38 @@ class MigrationJobAPITest {
verify(dataSourcePersistService).load(contextKey, "MIGRATION");
verify(jobManager, never()).start(any());
}
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("provideRollbackSchemas")
+ void assertRollbackNormalizesSchemaAndPreservesActualTable(final String
name, final String schemaName, final String expectedSchemaName) throws
SQLException {
+ String jobId = PipelineJobIdUtils.marshal(new
MigrationJobId(contextKey, Collections.emptyList()));
+ MigrationJobConfiguration jobConfig =
mock(MigrationJobConfiguration.class);
+
when(jobConfig.getTargetTableNames()).thenReturn(Collections.singletonList("T_Order"));
+
when(jobConfig.getTargetTableSchemaMap()).thenReturn(Collections.singletonMap("T_Order",
schemaName));
+ Connection connection = mock(Connection.class);
+ Statement statement = mock(Statement.class);
+ when(connection.createStatement()).thenReturn(statement);
+ try (
+ MockedStatic<PipelineAPIFactory> ignoredFactory =
mockStatic(PipelineAPIFactory.class, Answers.RETURNS_DEEP_STUBS);
+ MockedConstruction<PipelineJobConfigurationManager>
ignoredManager = mockConstruction(PipelineJobConfigurationManager.class,
+ (mock, context) ->
when(mock.getJobConfiguration(jobId)).thenReturn(jobConfig));
+ MockedConstruction<PipelineDataSource> ignoredDataSource =
mockConstruction(PipelineDataSource.class, (mock, context) -> {
+ when(mock.getConnection()).thenReturn(connection);
+ when(mock.getIdentifierContext()).thenReturn(new
DatabaseIdentifierContext(IdentifierCasePolicyFactory.newLowerCasePolicySet()));
+ });
+ MockedConstruction<PipelinePrepareSQLBuilder> builders =
mockConstruction(PipelinePrepareSQLBuilder.class,
+ (mock, context) ->
when(mock.buildDropSQL(expectedSchemaName, "T_Order")).thenReturn("DROP TABLE
IF EXISTS target_table"))) {
+ jobAPI.rollback(jobId);
+
verify(builders.constructed().get(0)).buildDropSQL(expectedSchemaName,
"T_Order");
+ verify(statement).execute("DROP TABLE IF EXISTS target_table");
+ verify(jobManager).drop(jobId);
+ }
+ }
+
+ private static Stream<Arguments> provideRollbackSchemas() {
+ return Stream.of(
+ Arguments.of("uppercase schema reference", "UPPER_SCHEMA",
"upper_schema"),
+ Arguments.of("absent schema", null, null),
+ Arguments.of("empty schema", "", ""));
+ }
}
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 4ff7dfc3ee6..eb31fd2dfb2 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
@@ -65,13 +65,16 @@ public final class MigrateTableExecutor implements
DistSQLUpdateExecutor<Migrate
() -> new
MissingRequiredTargetDatabaseException(targetDatabaseName));
MigrationJobAPI jobAPI = (MigrationJobAPI)
TypedSPILoader.getService(TransmissionJobAPI.class, "MIGRATION");
PipelineContextKey contextKey = new
PipelineContextKey(InstanceType.PROXY);
- jobAPI.schedule(contextKey,
getMigrationSourceTargetEntries(contextKey, sqlStatement), targetDatabaseName);
+ jobAPI.schedule(contextKey,
getMigrationSourceTargetEntries(contextKey, sqlStatement,
+
contextManager.getMetaDataContexts().getMetaData().getDatabase(targetDatabaseName)),
targetDatabaseName);
}
- private Collection<MigrationSourceTargetEntry>
getMigrationSourceTargetEntries(final PipelineContextKey contextKey, final
MigrateTableStatement sqlStatement) {
+ private Collection<MigrationSourceTargetEntry>
getMigrationSourceTargetEntries(final PipelineContextKey contextKey, final
MigrateTableStatement sqlStatement,
+
final ShardingSphereDatabase targetDatabase) {
Collection<MigrationSourceTargetEntry> result = new LinkedList<>();
for (MigrationSourceTargetSegment each :
sqlStatement.getSourceTargetEntries()) {
- result.add(new
MigrationSourceTargetEntry(getSourceDataNode(contextKey, each),
each.getTargetTableName()));
+ String targetTableName =
targetDatabase.getIdentifierContext().normalizeStorage(IdentifierScope.TABLE,
each.getTargetTableIdentifier());
+ result.add(new
MigrationSourceTargetEntry(getSourceDataNode(contextKey, each),
targetTableName));
}
return result;
}
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 b0c53d040a9..efacedbe7f0 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
@@ -30,12 +30,14 @@ import
org.apache.shardingsphere.database.connector.core.metadata.database.enums
import
org.apache.shardingsphere.database.connector.core.metadata.database.metadata.DialectDatabaseMetaData;
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.IdentifierCasePolicySet;
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.DatabaseTypeRegistry;
import
org.apache.shardingsphere.infra.datasource.pool.props.domain.DataSourcePoolProperties;
import
org.apache.shardingsphere.infra.metadata.database.ShardingSphereDatabase;
+import
org.apache.shardingsphere.infra.metadata.identifier.DatabaseIdentifierContext;
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;
@@ -103,7 +105,6 @@ class MigrateTableExecutorTest {
@BeforeEach
void setUp() {
executor.setDatabase(database);
- when(database.getName()).thenReturn(TARGET_DATABASE_NAME);
when(contextManager.getMetaDataContexts().getMetaData().containsDatabase(TARGET_DATABASE_NAME)).thenReturn(true);
}
@@ -206,11 +207,43 @@ class MigrateTableExecutorTest {
}
}
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("provideTargetIdentifiers")
+ void assertExecuteUpdateNormalizesTargetIdentifier(final String name,
final IdentifierCasePolicySet policySet,
+ final IdentifierValue
targetIdentifier, final String expectedTableName) {
+ ShardingSphereDatabase targetDatabase =
mock(ShardingSphereDatabase.class);
+
when(contextManager.getMetaDataContexts().getMetaData().getDatabase(TARGET_DATABASE_NAME)).thenReturn(targetDatabase);
+ when(targetDatabase.getIdentifierContext()).thenReturn(new
DatabaseIdentifierContext(policySet));
+ MigrationSourceTargetSegment entry = new
MigrationSourceTargetSegment(new IdentifierValue(SOURCE_DATABASE_NAME), null,
+ new IdentifierValue("source_table"), targetIdentifier);
+ try (
+ MockedConstruction<PipelineDataSourcePersistService> ignored =
mockConstruction(PipelineDataSourcePersistService.class,
+ (mock, context) ->
when(mock.load(any(PipelineContextKey.class),
eq("MIGRATION"))).thenReturn(Collections.emptyMap()));
+ MockedStatic<TypedSPILoader> spiLoader =
mockStatic(TypedSPILoader.class, Answers.CALLS_REAL_METHODS)) {
+ spiLoader.when(() ->
TypedSPILoader.getService(TransmissionJobAPI.class,
"MIGRATION")).thenReturn(jobAPI);
+ executor.executeUpdate(new
MigrateTableStatement(TARGET_DATABASE_NAME, Collections.singleton(entry)),
contextManager);
+ verify(jobAPI).schedule(any(PipelineContextKey.class),
entriesCaptor.capture(), eq(TARGET_DATABASE_NAME));
+
assertThat(entriesCaptor.getValue().iterator().next().getTargetTableName(),
is(expectedTableName));
+ verify(database, never()).getIdentifierContext();
+ }
+ }
+
+ private static Stream<Arguments> provideTargetIdentifiers() {
+ return Stream.of(
+ Arguments.of("unquoted uppercase target",
IdentifierCasePolicyFactory.newLowerCasePolicySet(), new
IdentifierValue("T_ORDER"), "t_order"),
+ Arguments.of("quoted uppercase target",
IdentifierCasePolicyFactory.newLowerCasePolicySet(), new
IdentifierValue("T_ORDER", QuoteCharacter.QUOTE), "T_ORDER"),
+ Arguments.of("quoted mixed-case target",
IdentifierCasePolicyFactory.newLowerCasePolicySet(), new
IdentifierValue("T_Order", QuoteCharacter.BACK_QUOTE), "T_Order"),
+ Arguments.of("case-sensitive storage target",
IdentifierCasePolicyFactory.newSensitivePolicySet(), new
IdentifierValue("T_ORDER"), "T_ORDER"));
+ }
+
private MigrateTableStatement createStatement(final String
sourceSchemaName) {
return createStatement(sourceSchemaName, new
IdentifierValue("FOO_TBL", QuoteCharacter.BACK_QUOTE));
}
private MigrateTableStatement createStatement(final String
sourceSchemaName, final IdentifierValue sourceTableIdentifier) {
+ when(database.getName()).thenReturn(TARGET_DATABASE_NAME);
+
when(contextManager.getMetaDataContexts().getMetaData().getDatabase(TARGET_DATABASE_NAME)).thenReturn(database);
+ when(database.getIdentifierContext()).thenReturn(new
DatabaseIdentifierContext(IdentifierCasePolicyFactory.newLowerCasePolicySet()));
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));
diff --git
a/test/e2e/operation/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/operation/pipeline/cases/PipelineContainerComposer.java
b/test/e2e/operation/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/operation/pipeline/cases/PipelineContainerComposer.java
index 95ca3b51f63..4084615e155 100644
---
a/test/e2e/operation/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/operation/pipeline/cases/PipelineContainerComposer.java
+++
b/test/e2e/operation/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/operation/pipeline/cases/PipelineContainerComposer.java
@@ -28,6 +28,7 @@ import
org.apache.shardingsphere.data.pipeline.core.datasource.PipelineDataSourc
import org.apache.shardingsphere.data.pipeline.core.job.JobStatus;
import org.apache.shardingsphere.data.pipeline.core.job.type.PipelineJobType;
import
org.apache.shardingsphere.database.connector.core.jdbcurl.appender.JdbcUrlAppender;
+import
org.apache.shardingsphere.database.connector.core.metadata.identifier.IdentifierScope;
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;
@@ -36,12 +37,15 @@ import
org.apache.shardingsphere.database.connector.opengauss.type.OpenGaussData
import
org.apache.shardingsphere.database.connector.postgresql.type.PostgreSQLDatabaseType;
import org.apache.shardingsphere.infra.exception.ShardingSpherePreconditions;
import org.apache.shardingsphere.infra.metadata.database.schema.QualifiedTable;
+import
org.apache.shardingsphere.infra.metadata.identifier.DatabaseIdentifierContext;
+import
org.apache.shardingsphere.infra.metadata.identifier.DatabaseIdentifierContextFactory;
import org.apache.shardingsphere.infra.spi.ShardingSphereServiceLoader;
import org.apache.shardingsphere.infra.util.props.PropertiesBuilder;
import org.apache.shardingsphere.infra.util.props.PropertiesBuilder.Property;
import org.apache.shardingsphere.infra.util.yaml.YamlEngine;
import org.apache.shardingsphere.infra.yaml.config.pojo.YamlRootConfiguration;
import
org.apache.shardingsphere.single.yaml.config.YamlSingleRuleConfiguration;
+import
org.apache.shardingsphere.sql.parser.statement.core.value.identifier.IdentifierValue;
import
org.apache.shardingsphere.test.e2e.env.container.constants.ProxyContainerConstants;
import
org.apache.shardingsphere.test.e2e.env.container.constants.StorageContainerConstants;
import
org.apache.shardingsphere.test.e2e.env.container.storage.option.StorageContainerOption;
@@ -219,9 +223,10 @@ public final class PipelineContainerComposer implements
AutoCloseable {
if (Type.NATIVE !=
E2ETestEnvironment.getInstance().getRunEnvironment().getType()) {
return;
}
- DatabaseTypeRegistry databaseTypeRegistry = new
DatabaseTypeRegistry(databaseType);
+ DatabaseIdentifierContext identifierContext =
DatabaseIdentifierContextFactory.create(databaseType, sourceDataSource);
+ IdentifierScope identifierScope =
ProxyDatabaseTypeUtils.isOracleBranch(databaseType) ? IdentifierScope.SCHEMA :
IdentifierScope.DATABASE;
for (String each : Arrays.asList(DS_0, DS_1, DS_2, DS_3, DS_4)) {
-
containerComposer.cleanUpDatabase(databaseTypeRegistry.formatIdentifierPattern(each));
+
containerComposer.cleanUpDatabase(identifierContext.normalizeStorage(identifierScope,
new IdentifierValue(each)));
}
}
