This is an automated email from the ASF dual-hosted git repository.
terrymanu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shardingsphere.git
The following commit(s) were added to refs/heads/master by this push:
new f016aa5cd90 Fix Pipeline schema identity handling (#39598)
f016aa5cd90 is described below
commit f016aa5cd908b3de8d001b9a5c7d16246667e9b9
Author: Liang Zhang <[email protected]>
AuthorDate: Tue Aug 25 00:49:01 2026 +0800
Fix Pipeline schema identity handling (#39598)
- resolve Migration default schemas from source database types
- reject ambiguous same-name tables before job creation
- preserve schema-aware CDC data node selection
- filter PostgreSQL and openGauss WAL events by schema
---
.../pipeline/core/util/PipelineDataNodeUtils.java | 36 +++-
.../core/util/PipelineDataNodeUtilsTest.java | 124 +++++++++++-
.../ingest/incremental/wal/WALEventConverter.java | 81 +++++---
.../incremental/wal/WALEventConverterTest.java | 97 ++++++++--
.../data/pipeline/cdc/api/CDCJobAPI.java | 13 ++
.../pipeline/cdc/handler/CDCBackendHandler.java | 24 +--
.../pipeline/cdc/util/CDCSchemaTableUtils.java | 53 +++--
.../data/pipeline/cdc/api/CDCJobAPITest.java | 35 +++-
.../cdc/handler/CDCBackendHandlerTest.java | 38 ++--
.../pipeline/cdc/util/CDCSchemaTableUtilsTest.java | 128 +++++++++---
.../scenario/migration/api/MigrationJobAPI.java | 18 +-
.../migration/api/MigrationJobAPITest.java | 100 ++++++++++
.../handler/update/MigrateTableExecutor.java | 23 ++-
.../handler/update/MigrateTableExecutorTest.java | 214 +++++++++++++++++++++
.../general/PostgreSQLMigrationGeneralE2EIT.java | 17 ++
15 files changed, 852 insertions(+), 149 deletions(-)
diff --git
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/util/PipelineDataNodeUtils.java
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/util/PipelineDataNodeUtils.java
index 81ad3a165f4..22864e704db 100644
---
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/util/PipelineDataNodeUtils.java
+++
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/util/PipelineDataNodeUtils.java
@@ -31,6 +31,9 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.stream.Collectors;
/**
* Pipeline data node utility class.
@@ -42,30 +45,43 @@ public final class PipelineDataNodeUtils {
* Build table and data nodes map.
*
* @param database database
- * @param tableNames table names
+ * @param schemaTableNames resolved schema table names, empty schema name
means schema unavailable
* @return table and data nodes map
* @throws PipelineInvalidParameterException thrown invalid parameter
exception when can't get data nodes.
*/
- public static Map<String, List<DataNode>> buildTableAndDataNodesMap(final
ShardingSphereDatabase database, final Collection<String> tableNames) {
- ShardingSpherePreconditions.checkNotEmpty(tableNames, () -> new
PipelineInvalidParameterException("Table names are empty."));
- Map<String, List<DataNode>> result = new HashMap<>(tableNames.size(),
1F);
+ public static Map<String, List<DataNode>> buildTableAndDataNodesMap(final
ShardingSphereDatabase database, final Map<String, Set<String>>
schemaTableNames) {
+ int tableCount =
schemaTableNames.values().stream().mapToInt(Collection::size).sum();
+ ShardingSpherePreconditions.checkState(0 < tableCount, () -> new
PipelineInvalidParameterException("Table names are empty."));
+ Map<String, List<DataNode>> result = new HashMap<>(tableCount, 1F);
Collection<DataNodeRuleAttribute> attributes =
database.getRuleMetaData().getAttributes(DataNodeRuleAttribute.class);
// TODO support virtual data source name
- for (String each : tableNames) {
- Collection<DataNode> dataNodes = findDataNodes(each, attributes);
- ShardingSpherePreconditions.checkNotEmpty(dataNodes, () -> new
PipelineInvalidParameterException(String.format("Not find actual data nodes of
`%s`", each)));
- result.put(each, new ArrayList<>(dataNodes));
+ for (Entry<String, Set<String>> entry : schemaTableNames.entrySet()) {
+ for (String each : entry.getValue()) {
+
ShardingSpherePreconditions.checkState(!result.containsKey(each),
+ () -> new
PipelineInvalidParameterException(String.format("More than one schema table has
the same table name `%s`.", each)));
+ Collection<DataNode> dataNodes = findDataNodes(entry.getKey(),
each, attributes);
+ ShardingSpherePreconditions.checkNotEmpty(dataNodes, () -> new
PipelineInvalidParameterException(String.format("Not find actual data nodes of
`%s`", each)));
+ result.put(each, new ArrayList<>(dataNodes));
+ }
}
return result;
}
- private static Collection<DataNode> findDataNodes(final String tableName,
final Collection<DataNodeRuleAttribute> attributes) {
+ private static Collection<DataNode> findDataNodes(final String schemaName,
final String tableName, final Collection<DataNodeRuleAttribute> attributes) {
for (DataNodeRuleAttribute each : attributes) {
Collection<DataNode> dataNodes =
each.getDataNodesByTableName(tableName);
if (!dataNodes.isEmpty()) {
- return each.isReplicaBasedDistribution() ?
Collections.singleton(dataNodes.iterator().next()) : dataNodes;
+ Collection<DataNode> matchedDataNodes =
filterDataNodesBySchema(schemaName, each, dataNodes);
+ return matchedDataNodes.isEmpty() ||
!each.isReplicaBasedDistribution() ? matchedDataNodes :
Collections.singleton(matchedDataNodes.iterator().next());
}
}
return Collections.emptyList();
}
+
+ private static Collection<DataNode> filterDataNodesBySchema(final String
schemaName, final DataNodeRuleAttribute ruleAttribute, final
Collection<DataNode> dataNodes) {
+ if (schemaName.isEmpty() ||
!ruleAttribute.isDataNodeTableNameLoadedFromStorage()) {
+ return dataNodes;
+ }
+ return dataNodes.stream().filter(each ->
schemaName.equals(each.getSchemaName())).collect(Collectors.toList());
+ }
}
diff --git
a/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/util/PipelineDataNodeUtilsTest.java
b/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/util/PipelineDataNodeUtilsTest.java
index e3bd5b012d9..236dced5e0c 100644
---
a/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/util/PipelineDataNodeUtilsTest.java
+++
b/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/util/PipelineDataNodeUtilsTest.java
@@ -25,41 +25,145 @@ import
org.apache.shardingsphere.infra.rule.attribute.datanode.DataNodeRuleAttri
import org.junit.jupiter.api.Test;
import java.util.Arrays;
+import java.util.Collection;
import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
+import java.util.Set;
-import static org.hamcrest.Matchers.is;
import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
class PipelineDataNodeUtilsTest {
@Test
void assertBuildTableAndDataNodesMap() {
- RuleMetaData ruleMetaData = mock(RuleMetaData.class);
DataNodeRuleAttribute notReplicaBasedDistributionDataNodeRuleAttribute
= mock(DataNodeRuleAttribute.class);
when(notReplicaBasedDistributionDataNodeRuleAttribute.getDataNodesByTableName("foo_tbl")).thenReturn(Collections.singleton(new
DataNode("foo_ds.foo_tbl")));
DataNodeRuleAttribute replicaBasedDistributionDataNodeRuleAttribute =
mock(DataNodeRuleAttribute.class);
+
when(replicaBasedDistributionDataNodeRuleAttribute.getDataNodesByTableName("foo_tbl")).thenReturn(Collections.singleton(new
DataNode("bar_ds.foo_tbl")));
when(replicaBasedDistributionDataNodeRuleAttribute.getDataNodesByTableName("bar_tbl")).thenReturn(Arrays.asList(new
DataNode("foo_ds.bar_tbl"), new DataNode("bar_ds.bar_tbl")));
+
when(replicaBasedDistributionDataNodeRuleAttribute.isDataNodeTableNameLoadedFromStorage()).thenReturn(true);
when(replicaBasedDistributionDataNodeRuleAttribute.isReplicaBasedDistribution()).thenReturn(true);
-
when(ruleMetaData.getAttributes(DataNodeRuleAttribute.class)).thenReturn(Arrays.asList(notReplicaBasedDistributionDataNodeRuleAttribute,
replicaBasedDistributionDataNodeRuleAttribute));
- ShardingSphereDatabase database = mock(ShardingSphereDatabase.class);
- when(database.getRuleMetaData()).thenReturn(ruleMetaData);
- Map<String, List<DataNode>> actual =
PipelineDataNodeUtils.buildTableAndDataNodesMap(database,
Arrays.asList("foo_tbl", "bar_tbl"));
+ ShardingSphereDatabase database =
mockDatabase(Arrays.asList(notReplicaBasedDistributionDataNodeRuleAttribute,
replicaBasedDistributionDataNodeRuleAttribute));
+ Map<String, Set<String>> schemaTableNames =
Collections.singletonMap("", new LinkedHashSet<>(Arrays.asList("foo_tbl",
"bar_tbl")));
+ Map<String, List<DataNode>> actual =
PipelineDataNodeUtils.buildTableAndDataNodesMap(database, schemaTableNames);
assertThat(actual.size(), is(2));
assertThat(actual.get("foo_tbl"), is(Collections.singletonList(new
DataNode("foo_ds.foo_tbl"))));
assertThat(actual.get("bar_tbl"), is(Collections.singletonList(new
DataNode("foo_ds.bar_tbl"))));
+ verify(replicaBasedDistributionDataNodeRuleAttribute,
never()).getDataNodesByTableName("foo_tbl");
+ }
+
+ @Test
+ void
assertBuildTableAndDataNodesMapFiltersStorageLoadedDataNodesBeforeSelectingReplica()
{
+ DataNode mismatchedDataNode = new DataNode("foo_ds", "caseschema",
"foo_tbl");
+ DataNode matchedDataNode = new DataNode("bar_ds", "CaseSchema",
"foo_tbl");
+ DataNodeRuleAttribute ruleAttribute = mockRuleAttribute("foo_tbl",
Arrays.asList(mismatchedDataNode, matchedDataNode), true, true);
+ ShardingSphereDatabase database =
mockDatabase(Collections.singleton(ruleAttribute));
+ Map<String, List<DataNode>> actual =
PipelineDataNodeUtils.buildTableAndDataNodesMap(
+ database, Collections.singletonMap("CaseSchema",
Collections.singleton("foo_tbl")));
+ assertThat(actual, is(Collections.singletonMap("foo_tbl",
Collections.singletonList(matchedDataNode))));
+ Map<String, List<DataNode>> lowerCaseActual =
PipelineDataNodeUtils.buildTableAndDataNodesMap(
+ database, Collections.singletonMap("caseschema",
Collections.singleton("foo_tbl")));
+ assertThat(lowerCaseActual, is(Collections.singletonMap("foo_tbl",
Collections.singletonList(mismatchedDataNode))));
+ }
+
+ @Test
+ void
assertBuildTableAndDataNodesMapDoesNotFilterOwnerNotLoadedFromStorage() {
+ DataNode dataNode = new DataNode("foo_ds", "bar_schema", "foo_tbl");
+ ShardingSphereDatabase database =
mockDatabase(Collections.singleton(mockRuleAttribute("foo_tbl",
Collections.singleton(dataNode), false, false)));
+ Map<String, List<DataNode>> actual =
PipelineDataNodeUtils.buildTableAndDataNodesMap(
+ database, Collections.singletonMap("foo_schema",
Collections.singleton("foo_tbl")));
+ assertThat(actual, is(Collections.singletonMap("foo_tbl",
Collections.singletonList(dataNode))));
+ }
+
+ @Test
+ void
assertBuildTableAndDataNodesMapDoesNotFallThroughWhenStorageLoadedOwnerHasNoMatchingSchema()
{
+ DataNodeRuleAttribute storageLoadedOwner = mockRuleAttribute(
+ "foo_tbl", Collections.singleton(new DataNode("foo_ds",
"bar_schema", "foo_tbl")), true, false);
+ DataNodeRuleAttribute laterOwner = mock(DataNodeRuleAttribute.class);
+ ShardingSphereDatabase database =
mockDatabase(Arrays.asList(storageLoadedOwner, laterOwner));
+ PipelineInvalidParameterException actual =
assertThrows(PipelineInvalidParameterException.class, () ->
PipelineDataNodeUtils.buildTableAndDataNodesMap(
+ database, Collections.singletonMap("foo_schema",
Collections.singleton("foo_tbl"))));
+ assertThat(actual.getMessage(), is("There is invalid parameter value.
Not find actual data nodes of `foo_tbl`"));
+ verify(laterOwner, never()).getDataNodesByTableName("foo_tbl");
+ }
+
+ @Test
+ void assertBuildTableAndDataNodesMapWithEmptySchemaTableNames() {
+ PipelineInvalidParameterException actual = assertThrows(
+ PipelineInvalidParameterException.class, () ->
PipelineDataNodeUtils.buildTableAndDataNodesMap(mock(ShardingSphereDatabase.class),
Collections.emptyMap()));
+ assertThat(actual.getMessage(), is("There is invalid parameter value.
Table names are empty."));
+ }
+
+ @Test
+ void assertBuildTableAndDataNodesMapWithEmptyTableNames() {
+ PipelineInvalidParameterException actual =
assertThrows(PipelineInvalidParameterException.class, () ->
PipelineDataNodeUtils.buildTableAndDataNodesMap(
+ mock(ShardingSphereDatabase.class),
Collections.singletonMap("foo_schema", Collections.emptySet())));
+ assertThat(actual.getMessage(), is("There is invalid parameter value.
Table names are empty."));
}
@Test
void assertBuildTableAndDataNodesMapWithNotExistedTable() {
+ PipelineInvalidParameterException actual =
assertThrows(PipelineInvalidParameterException.class, () ->
PipelineDataNodeUtils.buildTableAndDataNodesMap(
+ mockDatabase(Collections.emptyList()),
Collections.singletonMap("", Collections.singleton("foo_tbl"))));
+ assertThat(actual.getMessage(), is("There is invalid parameter value.
Not find actual data nodes of `foo_tbl`"));
+ }
+
+ @Test
+ void assertBuildTableAndDataNodesMapWithDuplicateTableNameAcrossSchemas() {
+ DataNodeRuleAttribute ruleAttribute = mockRuleAttribute("foo_tbl",
Collections.singleton(new DataNode("foo_ds.foo_tbl")), false, false);
+ ShardingSphereDatabase database =
mockDatabase(Collections.singleton(ruleAttribute));
+ Map<String, Set<String>> schemaTableNames = new LinkedHashMap<>(2, 1F);
+ schemaTableNames.put("foo_schema", Collections.singleton("foo_tbl"));
+ schemaTableNames.put("bar_schema", Collections.singleton("foo_tbl"));
+ PipelineInvalidParameterException actual = assertThrows(
+ PipelineInvalidParameterException.class, () ->
PipelineDataNodeUtils.buildTableAndDataNodesMap(database, schemaTableNames));
+ assertThat(actual.getMessage(), is("There is invalid parameter value.
More than one schema table has the same table name `foo_tbl`."));
+ verify(ruleAttribute).getDataNodesByTableName("foo_tbl");
+ }
+
+ @Test
+ void
assertBuildTableAndDataNodesMapDoesNotNormalizeTableNamesAcrossSchemas() {
+ DataNodeRuleAttribute ruleAttribute =
mock(DataNodeRuleAttribute.class);
+ DataNode fooDataNode = new DataNode("foo_ds.foo_tbl");
+ DataNode upperCaseFooDataNode = new DataNode("foo_ds.FOO_TBL");
+
when(ruleAttribute.getDataNodesByTableName("foo_tbl")).thenReturn(Collections.singleton(fooDataNode));
+
when(ruleAttribute.getDataNodesByTableName("FOO_TBL")).thenReturn(Collections.singleton(upperCaseFooDataNode));
+ ShardingSphereDatabase database =
mockDatabase(Collections.singleton(ruleAttribute));
+ Map<String, Set<String>> schemaTableNames = new LinkedHashMap<>(2, 1F);
+ schemaTableNames.put("foo_schema", Collections.singleton("foo_tbl"));
+ schemaTableNames.put("bar_schema", Collections.singleton("FOO_TBL"));
+ Map<String, List<DataNode>> actual =
PipelineDataNodeUtils.buildTableAndDataNodesMap(database, schemaTableNames);
+ assertThat(actual.get("foo_tbl"),
is(Collections.singletonList(fooDataNode)));
+ assertThat(actual.get("FOO_TBL"),
is(Collections.singletonList(upperCaseFooDataNode)));
+ }
+
+ private ShardingSphereDatabase mockDatabase(final
Collection<DataNodeRuleAttribute> ruleAttributes) {
RuleMetaData ruleMetaData = mock(RuleMetaData.class);
-
when(ruleMetaData.getAttributes(DataNodeRuleAttribute.class)).thenReturn(Collections.emptyList());
- ShardingSphereDatabase database = mock(ShardingSphereDatabase.class);
- when(database.getRuleMetaData()).thenReturn(ruleMetaData);
- assertThrows(PipelineInvalidParameterException.class, () ->
PipelineDataNodeUtils.buildTableAndDataNodesMap(database,
Collections.singleton("foo_tbl")));
+
when(ruleMetaData.getAttributes(DataNodeRuleAttribute.class)).thenReturn(ruleAttributes);
+ ShardingSphereDatabase result = mock(ShardingSphereDatabase.class);
+ when(result.getRuleMetaData()).thenReturn(ruleMetaData);
+ return result;
+ }
+
+ private DataNodeRuleAttribute mockRuleAttribute(final String tableName,
final Collection<DataNode> dataNodes,
+ final boolean
tableNameLoadedFromStorage, final boolean replicaBasedDistribution) {
+ DataNodeRuleAttribute result = mock(DataNodeRuleAttribute.class);
+ when(result.getDataNodesByTableName(tableName)).thenReturn(dataNodes);
+ if (tableNameLoadedFromStorage) {
+
when(result.isDataNodeTableNameLoadedFromStorage()).thenReturn(true);
+ }
+ if (replicaBasedDistribution) {
+ when(result.isReplicaBasedDistribution()).thenReturn(true);
+ }
+ return result;
}
}
diff --git
a/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/ingest/incremental/wal/WALEventConverter.java
b/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/ingest/incremental/wal/WALEventConverter.java
index 0f48322c2ee..61153669589 100644
---
a/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/ingest/incremental/wal/WALEventConverter.java
+++
b/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/ingest/incremental/wal/WALEventConverter.java
@@ -17,7 +17,6 @@
package
org.apache.shardingsphere.data.pipeline.postgresql.ingest.incremental.wal;
-import lombok.RequiredArgsConstructor;
import
org.apache.shardingsphere.data.pipeline.core.constant.PipelineSQLOperationType;
import
org.apache.shardingsphere.data.pipeline.core.ingest.dumper.incremental.IncrementalDumperContext;
import org.apache.shardingsphere.data.pipeline.core.ingest.record.Column;
@@ -33,6 +32,10 @@ import
org.apache.shardingsphere.data.pipeline.postgresql.ingest.incremental.wal
import
org.apache.shardingsphere.data.pipeline.postgresql.ingest.incremental.wal.event.DeleteRowEvent;
import
org.apache.shardingsphere.data.pipeline.postgresql.ingest.incremental.wal.event.UpdateRowEvent;
import
org.apache.shardingsphere.data.pipeline.postgresql.ingest.incremental.wal.event.WriteRowEvent;
+import
org.apache.shardingsphere.database.connector.core.metadata.database.enums.QuoteCharacter;
+import
org.apache.shardingsphere.database.connector.core.metadata.identifier.IdentifierCasePolicy;
+import
org.apache.shardingsphere.database.connector.core.metadata.identifier.IdentifierNormalizeEngine;
+import
org.apache.shardingsphere.database.connector.core.metadata.identifier.IdentifierScope;
import
org.apache.shardingsphere.infra.exception.generic.UnsupportedSQLOperationException;
import
org.apache.shardingsphere.infra.metadata.identifier.ShardingSphereIdentifier;
@@ -41,13 +44,27 @@ import java.util.List;
/**
* WAL event converter.
*/
-@RequiredArgsConstructor
public final class WALEventConverter {
private final IncrementalDumperContext dumperContext;
private final PipelineTableMetaDataLoader metaDataLoader;
+ private final IdentifierCasePolicy schemaIdentifierCasePolicy;
+
+ /**
+ * Create a WAL event converter.
+ *
+ * @param dumperContext incremental dumper context
+ * @param metaDataLoader pipeline table metadata loader
+ */
+ public WALEventConverter(final IncrementalDumperContext dumperContext,
final PipelineTableMetaDataLoader metaDataLoader) {
+ this.dumperContext = dumperContext;
+ this.metaDataLoader = metaDataLoader;
+ schemaIdentifierCasePolicy = IdentifierNormalizeEngine.resolvePolicy(
+
dumperContext.getCommonContext().getDataSourceConfig().getDatabaseType(), null,
IdentifierScope.SCHEMA);
+ }
+
/**
* Convert WAL event to {@code Record}.
*
@@ -56,57 +73,68 @@ public final class WALEventConverter {
* @throws UnsupportedSQLOperationException unsupported SQL operation
exception
*/
public Record convert(final AbstractWALEvent event) {
- if (filter(event)) {
+ if (!(event instanceof AbstractRowEvent)) {
return createPlaceholderRecord(event);
}
- if (!(event instanceof AbstractRowEvent)) {
+ AbstractRowEvent rowEvent = (AbstractRowEvent) event;
+ String actualTableName = rowEvent.getTableName();
+ ShardingSphereIdentifier logicTableName =
dumperContext.getCommonContext().getTableNameMapper().getLogicTableName(actualTableName);
+ if (null == logicTableName) {
return createPlaceholderRecord(event);
}
- PipelineTableMetaData tableMetaData =
getPipelineTableMetaData(((AbstractRowEvent) event).getTableName());
+ String expectedSchemaName =
dumperContext.getCommonContext().getTableAndSchemaNameMapper().getSchemaName(logicTableName);
+ if (isDifferentSchemaName(rowEvent, expectedSchemaName)) {
+ return createPlaceholderRecord(event);
+ }
+ PipelineTableMetaData tableMetaData =
metaDataLoader.getTableMetaData(expectedSchemaName, actualTableName);
if (event instanceof WriteRowEvent) {
- return handleWriteRowEvent((WriteRowEvent) event, tableMetaData);
+ return handleWriteRowEvent((WriteRowEvent) event, tableMetaData,
logicTableName);
}
if (event instanceof UpdateRowEvent) {
- return handleUpdateRowEvent((UpdateRowEvent) event, tableMetaData);
+ return handleUpdateRowEvent((UpdateRowEvent) event, tableMetaData,
logicTableName);
}
if (event instanceof DeleteRowEvent) {
- return handleDeleteRowEvent((DeleteRowEvent) event, tableMetaData);
+ return handleDeleteRowEvent((DeleteRowEvent) event, tableMetaData,
logicTableName);
}
throw new UnsupportedSQLOperationException("");
}
- private boolean filter(final AbstractWALEvent event) {
- if (event instanceof AbstractRowEvent) {
- AbstractRowEvent rowEvent = (AbstractRowEvent) event;
- return
!dumperContext.getCommonContext().getTableNameMapper().containsTable(rowEvent.getTableName());
+ private boolean isDifferentSchemaName(final AbstractRowEvent event, final
String expectedSchemaName) {
+ if (!isConcreteSchemaName(expectedSchemaName)) {
+ return false;
}
- return false;
+ String actualSchemaName = event.getSchemaName();
+ if (!isConcreteSchemaName(actualSchemaName)) {
+ return false;
+ }
+ // The mapper stores a canonical schema name, while the WAL decoders
preserve the event identifier's quote characters.
+ QuoteCharacter quoteCharacter =
QuoteCharacter.getQuoteCharacter(actualSchemaName);
+ return !schemaIdentifierCasePolicy.matches(expectedSchemaName,
quoteCharacter.unwrap(actualSchemaName), quoteCharacter);
}
- private PlaceholderRecord createPlaceholderRecord(final AbstractWALEvent
event) {
- return new PlaceholderRecord(new
WALPosition(event.getLogSequenceNumber()));
+ private static boolean isConcreteSchemaName(final String schemaName) {
+ return null != schemaName && !schemaName.isEmpty() &&
!"*".equals(schemaName);
}
- private PipelineTableMetaData getPipelineTableMetaData(final String
actualTableName) {
- ShardingSphereIdentifier logicTableName =
dumperContext.getCommonContext().getTableNameMapper().getLogicTableName(actualTableName);
- return
metaDataLoader.getTableMetaData(dumperContext.getCommonContext().getTableAndSchemaNameMapper().getSchemaName(logicTableName),
actualTableName);
+ private PlaceholderRecord createPlaceholderRecord(final AbstractWALEvent
event) {
+ return new PlaceholderRecord(new
WALPosition(event.getLogSequenceNumber()));
}
- private DataRecord handleWriteRowEvent(final WriteRowEvent writeRowEvent,
final PipelineTableMetaData tableMetaData) {
- DataRecord result = createDataRecord(PipelineSQLOperationType.INSERT,
writeRowEvent, writeRowEvent.getAfterRow().size());
+ private DataRecord handleWriteRowEvent(final WriteRowEvent writeRowEvent,
final PipelineTableMetaData tableMetaData, final ShardingSphereIdentifier
logicTableName) {
+ DataRecord result = createDataRecord(PipelineSQLOperationType.INSERT,
writeRowEvent, logicTableName, writeRowEvent.getAfterRow().size());
putColumnsIntoDataRecord(result, tableMetaData,
writeRowEvent.getAfterRow());
return result;
}
- private DataRecord handleUpdateRowEvent(final UpdateRowEvent
updateRowEvent, final PipelineTableMetaData tableMetaData) {
- DataRecord result = createDataRecord(PipelineSQLOperationType.UPDATE,
updateRowEvent, updateRowEvent.getAfterRow().size());
+ private DataRecord handleUpdateRowEvent(final UpdateRowEvent
updateRowEvent, final PipelineTableMetaData tableMetaData, final
ShardingSphereIdentifier logicTableName) {
+ DataRecord result = createDataRecord(PipelineSQLOperationType.UPDATE,
updateRowEvent, logicTableName, updateRowEvent.getAfterRow().size());
putColumnsIntoDataRecord(result, tableMetaData,
updateRowEvent.getAfterRow());
return result;
}
- private DataRecord handleDeleteRowEvent(final DeleteRowEvent event, final
PipelineTableMetaData tableMetaData) {
+ private DataRecord handleDeleteRowEvent(final DeleteRowEvent event, final
PipelineTableMetaData tableMetaData, final ShardingSphereIdentifier
logicTableName) {
// TODO completion columns
- DataRecord result = createDataRecord(PipelineSQLOperationType.DELETE,
event, event.getPrimaryKeys().size());
+ DataRecord result = createDataRecord(PipelineSQLOperationType.DELETE,
event, logicTableName, event.getPrimaryKeys().size());
// TODO Unique key may be a column within unique index
List<String> primaryKeyColumns = tableMetaData.getPrimaryKeyColumns();
for (int i = 0; i < event.getPrimaryKeys().size(); i++) {
@@ -115,9 +143,8 @@ public final class WALEventConverter {
return result;
}
- private DataRecord createDataRecord(final PipelineSQLOperationType type,
final AbstractRowEvent rowsEvent, final int columnCount) {
- String tableName =
dumperContext.getCommonContext().getTableNameMapper().getLogicTableName(rowsEvent.getTableName()).toString();
- DataRecord result = new DataRecord(type, rowsEvent.getSchemaName(),
tableName, new WALPosition(rowsEvent.getLogSequenceNumber()), columnCount);
+ private DataRecord createDataRecord(final PipelineSQLOperationType type,
final AbstractRowEvent rowsEvent, final ShardingSphereIdentifier
logicTableName, final int columnCount) {
+ DataRecord result = new DataRecord(type, rowsEvent.getSchemaName(),
logicTableName.toString(), new WALPosition(rowsEvent.getLogSequenceNumber()),
columnCount);
result.setActualTableName(rowsEvent.getTableName());
result.setCsn(rowsEvent.getCsn());
return result;
diff --git
a/kernel/data-pipeline/dialect/postgresql/src/test/java/org/apache/shardingsphere/data/pipeline/postgresql/ingest/incremental/wal/WALEventConverterTest.java
b/kernel/data-pipeline/dialect/postgresql/src/test/java/org/apache/shardingsphere/data/pipeline/postgresql/ingest/incremental/wal/WALEventConverterTest.java
index ed6c483fd4d..13e40c965c9 100644
---
a/kernel/data-pipeline/dialect/postgresql/src/test/java/org/apache/shardingsphere/data/pipeline/postgresql/ingest/incremental/wal/WALEventConverterTest.java
+++
b/kernel/data-pipeline/dialect/postgresql/src/test/java/org/apache/shardingsphere/data/pipeline/postgresql/ingest/incremental/wal/WALEventConverterTest.java
@@ -17,6 +17,7 @@
package
org.apache.shardingsphere.data.pipeline.postgresql.ingest.incremental.wal;
+import
org.apache.shardingsphere.data.pipeline.api.PipelineDataSourceConfiguration;
import
org.apache.shardingsphere.data.pipeline.api.type.StandardPipelineDataSourceConfiguration;
import
org.apache.shardingsphere.data.pipeline.core.constant.PipelineSQLOperationType;
import
org.apache.shardingsphere.data.pipeline.core.datasource.PipelineDataSource;
@@ -28,6 +29,7 @@ import
org.apache.shardingsphere.data.pipeline.core.ingest.dumper.mapper.TableAn
import org.apache.shardingsphere.data.pipeline.core.ingest.record.DataRecord;
import
org.apache.shardingsphere.data.pipeline.core.ingest.record.PlaceholderRecord;
import org.apache.shardingsphere.data.pipeline.core.ingest.record.Record;
+import
org.apache.shardingsphere.data.pipeline.core.metadata.loader.PipelineTableMetaDataLoader;
import
org.apache.shardingsphere.data.pipeline.core.metadata.loader.StandardPipelineTableMetaDataLoader;
import
org.apache.shardingsphere.data.pipeline.core.metadata.model.PipelineColumnMetaData;
import
org.apache.shardingsphere.data.pipeline.core.metadata.model.PipelineTableMetaData;
@@ -39,15 +41,18 @@ import
org.apache.shardingsphere.data.pipeline.postgresql.ingest.incremental.wal
import
org.apache.shardingsphere.data.pipeline.postgresql.ingest.incremental.wal.event.PlaceholderEvent;
import
org.apache.shardingsphere.data.pipeline.postgresql.ingest.incremental.wal.event.UpdateRowEvent;
import
org.apache.shardingsphere.data.pipeline.postgresql.ingest.incremental.wal.event.WriteRowEvent;
+import org.apache.shardingsphere.database.connector.core.type.DatabaseType;
import
org.apache.shardingsphere.infra.exception.generic.UnsupportedSQLOperationException;
import
org.apache.shardingsphere.infra.metadata.identifier.ShardingSphereIdentifier;
+import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
-import org.mockito.internal.configuration.plugins.Plugins;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
import org.postgresql.replication.LogSequenceNumber;
-import java.lang.reflect.Method;
import java.sql.Connection;
import java.sql.SQLException;
import java.sql.Statement;
@@ -60,14 +65,21 @@ import java.util.List;
import java.util.Map;
import java.util.function.Function;
import java.util.stream.Collectors;
+import java.util.stream.Stream;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.isA;
import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.when;
class WALEventConverterTest {
+ private final DatabaseType databaseType =
TypedSPILoader.getService(DatabaseType.class, "PostgreSQL");
+
private WALEventConverter walEventConverter;
private final LogSequenceNumber logSequenceNumber =
LogSequenceNumber.valueOf("0/14EFDB8");
@@ -127,25 +139,20 @@ class WALEventConverterTest {
}
@Test
- void assertWriteRowEvent() throws ReflectiveOperationException {
- DataRecord actual = getDataRecord(createWriteRowEvent());
+ void assertWriteRowEvent() {
+ DataRecord actual = (DataRecord)
walEventConverter.convert(createWriteRowEvent(""));
assertThat(actual.getType(), is(PipelineSQLOperationType.INSERT));
assertThat(actual.getColumnCount(), is(3));
}
- private WriteRowEvent createWriteRowEvent() {
+ private WriteRowEvent createWriteRowEvent(final String schemaName) {
WriteRowEvent result = new WriteRowEvent();
- result.setSchemaName("");
+ result.setSchemaName(schemaName);
result.setTableName("t_order");
result.setAfterRow(Arrays.asList(101, 1, "OK"));
return result;
}
- private DataRecord getDataRecord(final WriteRowEvent rowsEvent) throws
ReflectiveOperationException {
- Method method =
WALEventConverter.class.getDeclaredMethod("handleWriteRowEvent",
WriteRowEvent.class, PipelineTableMetaData.class);
- return (DataRecord) Plugins.getMemberAccessor().invoke(method,
walEventConverter, rowsEvent, pipelineTableMetaData);
- }
-
@Test
void assertConvertBeginTXEvent() {
BeginTXEvent beginTXEvent = new BeginTXEvent(100L, null);
@@ -196,6 +203,74 @@ class WALEventConverterTest {
assertThat(walEventConverter.convert(mockUnknownTableEvent()),
isA(PlaceholderRecord.class));
}
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("provideDifferentConcreteSchemaNames")
+ void assertConvertWithDifferentConcreteSchema(final String name, final
String expectedSchemaName, final String eventSchemaName) {
+ PipelineTableMetaDataLoader metaDataLoader =
mock(PipelineTableMetaDataLoader.class);
+ when(metaDataLoader.getTableMetaData(expectedSchemaName,
"t_order")).thenReturn(pipelineTableMetaData);
+ WALEventConverter converter =
createSchemaAwareWALEventConverter(expectedSchemaName, metaDataLoader);
+ Record actual =
converter.convert(createWriteRowEvent(eventSchemaName));
+ assertThat(actual, isA(PlaceholderRecord.class));
+ verifyNoInteractions(metaDataLoader);
+ }
+
+ private static Stream<Arguments> provideDifferentConcreteSchemaNames() {
+ return Stream.of(
+ Arguments.of("different schema", "test", "public"),
+ Arguments.of("quoted upper-case schema", "public",
"\"PUBLIC\""),
+ Arguments.of("quoted schema and unquoted token", "CaseSchema",
"CaseSchema"),
+ Arguments.of("quoted and unquoted case collision",
"CaseSchema", "caseschema"));
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("provideSameConcreteSchemaNames")
+ void assertConvertWithSameConcreteSchema(final String name, final String
expectedSchemaName, final String eventSchemaName) {
+ PipelineTableMetaDataLoader metaDataLoader =
mock(PipelineTableMetaDataLoader.class);
+ when(metaDataLoader.getTableMetaData(expectedSchemaName,
"t_order")).thenReturn(pipelineTableMetaData);
+ WALEventConverter converter =
createSchemaAwareWALEventConverter(expectedSchemaName, metaDataLoader);
+ Record actual =
converter.convert(createWriteRowEvent(eventSchemaName));
+ assertThat(actual, isA(DataRecord.class));
+ verify(metaDataLoader).getTableMetaData(expectedSchemaName, "t_order");
+ }
+
+ private static Stream<Arguments> provideSameConcreteSchemaNames() {
+ return Stream.of(
+ Arguments.of("lower-case schema", "public", "public"),
+ Arguments.of("unquoted mixed-case token", "caseschema",
"CaseSchema"),
+ Arguments.of("quoted upper-case schema", "PUBLIC",
"\"PUBLIC\""),
+ Arguments.of("quoted mixed-case schema", "CaseSchema",
"\"CaseSchema\""));
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("provideUnspecifiedSchema")
+ void assertConvertWithUnspecifiedSchema(final String name, final String
expectedSchemaName, final String eventSchemaName) {
+ PipelineTableMetaDataLoader metaDataLoader =
mock(PipelineTableMetaDataLoader.class);
+ when(metaDataLoader.getTableMetaData(expectedSchemaName,
"t_order")).thenReturn(pipelineTableMetaData);
+ WALEventConverter converter =
createSchemaAwareWALEventConverter(expectedSchemaName, metaDataLoader);
+ Record actual =
converter.convert(createWriteRowEvent(eventSchemaName));
+ assertThat(actual, isA(DataRecord.class));
+ verify(metaDataLoader).getTableMetaData(expectedSchemaName, "t_order");
+ }
+
+ private static Stream<Arguments> provideUnspecifiedSchema() {
+ return Stream.of(
+ Arguments.of("null expected schema", null, "public"),
+ Arguments.of("empty expected schema", "", "public"),
+ Arguments.of("wildcard expected schema", "*", "public"),
+ Arguments.of("null event schema", "public", null),
+ Arguments.of("empty event schema", "public", ""),
+ Arguments.of("wildcard event schema", "public", "*"));
+ }
+
+ private WALEventConverter createSchemaAwareWALEventConverter(final String
expectedSchemaName, final PipelineTableMetaDataLoader metaDataLoader) {
+ PipelineDataSourceConfiguration dataSourceConfig =
mock(PipelineDataSourceConfiguration.class);
+ when(dataSourceConfig.getDatabaseType()).thenReturn(databaseType);
+ DumperCommonContext commonContext = new DumperCommonContext(null,
dataSourceConfig,
+ new ActualAndLogicTableNameMapper(Collections.singletonMap(new
ShardingSphereIdentifier("t_order"), new ShardingSphereIdentifier("t_order"))),
+ new
TableAndSchemaNameMapper(Collections.singletonMap("t_order",
expectedSchemaName)));
+ return new WALEventConverter(new
IncrementalDumperContext(commonContext, null, false), metaDataLoader);
+ }
+
@Test
void assertConvertFailure() {
AbstractRowEvent event = new AbstractRowEvent() {
diff --git
a/kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/api/CDCJobAPI.java
b/kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/api/CDCJobAPI.java
index 0e59b6db65e..6d251ee5631 100644
---
a/kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/api/CDCJobAPI.java
+++
b/kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/api/CDCJobAPI.java
@@ -69,6 +69,7 @@ import
org.apache.shardingsphere.infra.exception.ShardingSpherePreconditions;
import org.apache.shardingsphere.infra.instance.metadata.InstanceType;
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.ShardingSphereIdentifier;
import org.apache.shardingsphere.infra.util.datetime.DateTimeFormatterFactory;
import org.apache.shardingsphere.infra.util.yaml.YamlEngine;
import org.apache.shardingsphere.infra.yaml.config.pojo.YamlRootConfiguration;
@@ -83,12 +84,14 @@ import java.time.LocalDateTime;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Optional;
import java.util.Properties;
+import java.util.Set;
import java.util.stream.Collectors;
/**
@@ -136,6 +139,7 @@ public final class CDCJobAPI implements TransmissionJobAPI {
log.warn("CDC job already exists in registry center, ignore, job
id is `{}`", jobConfig.getJobId());
} else {
checkDataSources(jobConfig);
+ checkSchemaTableNames(jobConfig.getSchemaTableNames());
governanceFacade.getJobFacade().getJob().create(jobConfig.getJobId(),
jobType.getOption().getJobClass());
JobConfigurationPOJO jobConfigPOJO =
jobConfigManager.convertToJobConfigurationPOJO(jobConfig);
jobConfigPOJO.setDisabled(true);
@@ -159,6 +163,15 @@ public final class CDCJobAPI implements TransmissionJobAPI
{
return jobConfig.getJobShardingDataNodes().stream().flatMap(each ->
each.getEntries().stream()).flatMap(dataNodeEntry ->
dataNodeEntry.getDataNodes().stream()).collect(Collectors.toList());
}
+ private void checkSchemaTableNames(final Collection<String>
schemaTableNames) {
+ Set<ShardingSphereIdentifier> tableNames = new
HashSet<>(schemaTableNames.size(), 1F);
+ for (String each : schemaTableNames) {
+ String tableName = each.substring(each.lastIndexOf('.') + 1);
+ ShardingSpherePreconditions.checkState(tableNames.add(new
ShardingSphereIdentifier(tableName)),
+ () -> new
PipelineInvalidParameterException(String.format("More than one schema table has
the same table name `%s`.", tableName)));
+ }
+ }
+
private YamlCDCJobConfiguration getYamlCDCJobConfiguration(final
StreamDataParameter param, final CDCSinkType sinkType, final Properties
sinkProps, final PipelineContextKey contextKey) {
YamlCDCJobConfiguration result = new YamlCDCJobConfiguration();
List<String> schemaTableNames = param.getSchemaTableNames();
diff --git
a/kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/handler/CDCBackendHandler.java
b/kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/handler/CDCBackendHandler.java
index 49bfd12beb3..7167f58dd79 100644
---
a/kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/handler/CDCBackendHandler.java
+++
b/kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/handler/CDCBackendHandler.java
@@ -38,7 +38,6 @@ import
org.apache.shardingsphere.data.pipeline.cdc.exception.StreamDatabaseNotFo
import org.apache.shardingsphere.data.pipeline.cdc.generator.CDCResponseUtils;
import
org.apache.shardingsphere.data.pipeline.cdc.protocol.request.AckStreamingRequestBody;
import
org.apache.shardingsphere.data.pipeline.cdc.protocol.request.StreamDataRequestBody;
-import
org.apache.shardingsphere.data.pipeline.cdc.protocol.request.StreamDataRequestBody.SchemaTable;
import
org.apache.shardingsphere.data.pipeline.cdc.protocol.response.CDCResponse;
import
org.apache.shardingsphere.data.pipeline.cdc.protocol.response.CDCResponse.ResponseCase;
import
org.apache.shardingsphere.data.pipeline.cdc.protocol.response.StreamDataResult;
@@ -51,7 +50,6 @@ import
org.apache.shardingsphere.data.pipeline.core.job.api.TransmissionJobAPI;
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.util.PipelineDataNodeUtils;
-import
org.apache.shardingsphere.database.connector.core.metadata.database.metadata.DialectDatabaseMetaData;
import
org.apache.shardingsphere.database.connector.core.type.DatabaseTypeRegistry;
import org.apache.shardingsphere.infra.datanode.DataNode;
import org.apache.shardingsphere.infra.exception.ShardingSpherePreconditions;
@@ -59,13 +57,11 @@ import
org.apache.shardingsphere.infra.metadata.database.ShardingSphereDatabase;
import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
import java.util.ArrayList;
-import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
-import java.util.stream.Collectors;
/**
* CDC backend handler.
@@ -100,22 +96,12 @@ public final class CDCBackendHandler {
ShardingSphereDatabase database =
PipelineContextManager.getProxyContext().getMetaDataContexts().getMetaData().getDatabase(requestBody.getDatabase());
ShardingSpherePreconditions.checkNotNull(database,
() -> new CDCExceptionWrapper(requestId, new
StreamDatabaseNotFoundException(String.format("%s database is not exists",
requestBody.getDatabase()))));
- Map<String, Set<String>> schemaTableNameMap;
- Collection<String> tableNames;
+ Map<String, Set<String>> schemaTableNameMap =
CDCSchemaTableUtils.parseTableExpressions(database,
requestBody.getSourceSchemaTableList());
Set<String> schemaTableNames = new HashSet<>();
- DialectDatabaseMetaData dialectDatabaseMetaData = new
DatabaseTypeRegistry(database.getProtocolType()).getDialectDatabaseMetaData();
- if (dialectDatabaseMetaData.getSchemaOption().isSchemaAvailable()) {
- schemaTableNameMap =
CDCSchemaTableUtils.parseTableExpressionWithSchema(database,
requestBody.getSourceSchemaTableList());
- // TODO if different schema have same table names, table name may
be overwritten, because the table name at sharding rule not contain schema.
- tableNames =
schemaTableNameMap.values().stream().flatMap(Collection::stream).collect(Collectors.toList());
- schemaTableNameMap.forEach((key, value) -> value.forEach(tableName
-> schemaTableNames.add(key.isEmpty() ? tableName : String.join(".", key,
tableName))));
- } else {
-
schemaTableNames.addAll(CDCSchemaTableUtils.parseTableExpressionWithoutSchema(database,
-
requestBody.getSourceSchemaTableList().stream().map(SchemaTable::getTable).collect(Collectors.toList())));
- tableNames = schemaTableNames;
- }
- ShardingSpherePreconditions.checkNotEmpty(tableNames, () -> new
CDCExceptionWrapper(requestId, new MissingRequiredStreamDataSourceException()));
- Map<String, List<DataNode>> tableAndDataNodesMap =
PipelineDataNodeUtils.buildTableAndDataNodesMap(database, tableNames);
+ schemaTableNameMap.forEach((schemaName, tableNames) ->
tableNames.forEach(
+ tableName -> schemaTableNames.add(schemaName.isEmpty() ?
tableName : String.join(".", schemaName, tableName))));
+ ShardingSpherePreconditions.checkNotEmpty(schemaTableNames, () -> new
CDCExceptionWrapper(requestId, new MissingRequiredStreamDataSourceException()));
+ Map<String, List<DataNode>> tableAndDataNodesMap =
PipelineDataNodeUtils.buildTableAndDataNodesMap(database, schemaTableNameMap);
// TODO Add globalCSNSupported to isolate it with
isDecodeWithTransaction flag, they're different. And also update CDCJobPreparer
needSorting flag.
boolean isDecodeWithTransaction = new
DatabaseTypeRegistry(database.getProtocolType()).getDialectDatabaseMetaData().getTransactionOption().isSupportGlobalCSN();
StreamDataParameter parameter = new
StreamDataParameter(requestBody.getDatabase(), new
ArrayList<>(schemaTableNames), requestBody.getFull(), tableAndDataNodesMap,
isDecodeWithTransaction);
diff --git
a/kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/util/CDCSchemaTableUtils.java
b/kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/util/CDCSchemaTableUtils.java
index 44aaf2d6ed3..ba986b053bb 100644
---
a/kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/util/CDCSchemaTableUtils.java
+++
b/kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/util/CDCSchemaTableUtils.java
@@ -30,12 +30,12 @@ import
org.apache.shardingsphere.infra.exception.kernel.metadata.TableNotFoundEx
import
org.apache.shardingsphere.infra.metadata.database.ShardingSphereDatabase;
import
org.apache.shardingsphere.infra.metadata.database.schema.model.ShardingSphereSchema;
import
org.apache.shardingsphere.infra.metadata.database.schema.model.ShardingSphereTable;
+import
org.apache.shardingsphere.sql.parser.statement.core.value.identifier.IdentifierValue;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
-import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;
@@ -47,19 +47,26 @@ import java.util.stream.Collectors;
public final class CDCSchemaTableUtils {
/**
- * Parse table expression with schema.
+ * Parse table expressions.
*
* @param database database
* @param schemaTables schema tables
- * @return map key is schema, value is table names
+ * @return map of resolved schema names to table names; the schema name is
empty only for database protocols without schema
*/
- public static Map<String, Set<String>>
parseTableExpressionWithSchema(final ShardingSphereDatabase database, final
Collection<SchemaTable> schemaTables) {
+ public static Map<String, Set<String>> parseTableExpressions(final
ShardingSphereDatabase database, final Collection<SchemaTable> schemaTables) {
+ DialectDatabaseMetaData dialectDatabaseMetaData = new
DatabaseTypeRegistry(database.getProtocolType()).getDialectDatabaseMetaData();
+ return dialectDatabaseMetaData.getSchemaOption().isSchemaAvailable()
+ ? parseTableExpressionsWithSchema(database, schemaTables,
dialectDatabaseMetaData)
+ : parseTableExpressionsWithoutSchema(database, schemaTables);
+ }
+
+ private static Map<String, Set<String>>
parseTableExpressionsWithSchema(final ShardingSphereDatabase database, final
Collection<SchemaTable> schemaTables,
+
final DialectDatabaseMetaData dialectDatabaseMetaData) {
Collection<String> systemSchemas =
DatabaseTypedSPILoader.getService(DialectSystemDatabase.class,
database.getProtocolType()).getSystemSchemas();
if (schemaTables.stream().anyMatch(each -> "*".equals(each.getTable())
&& ("*".equals(each.getSchema()) || each.getSchema().isEmpty()))) {
return parseTableExpressionWithAllTables(database, systemSchemas);
}
Map<String, Set<String>> result = new HashMap<>();
- DialectDatabaseMetaData dialectDatabaseMetaData = new
DatabaseTypeRegistry(database.getProtocolType()).getDialectDatabaseMetaData();
for (SchemaTable each : schemaTables) {
if ("*".equals(each.getSchema())) {
result.putAll(parseTableExpressionWithAllSchema(database,
systemSchemas, each));
@@ -67,11 +74,12 @@ public final class CDCSchemaTableUtils {
result.putAll(parseTableExpressionWithAllTable(database,
each));
} else {
String schemaName = each.getSchema();
- if
(dialectDatabaseMetaData.getSchemaOption().getDefaultSchema().isPresent() &&
schemaName.isEmpty()) {
- schemaName =
dialectDatabaseMetaData.getSchemaOption().getDefaultSchema().get();
+ if (schemaName.isEmpty()) {
+ schemaName =
dialectDatabaseMetaData.getSchemaOption().getDefaultSchema().orElse("");
}
-
ShardingSpherePreconditions.checkNotNull(database.getSchema(schemaName).getTable(each.getTable()),
() -> new TableNotFoundException(each.getTable()));
- result.computeIfAbsent(schemaName, ignored -> new
HashSet<>()).add(each.getTable());
+ ShardingSphereSchema schema = database.getSchema(new
IdentifierValue(schemaName));
+
ShardingSpherePreconditions.checkNotNull(schema.getTable(each.getTable()), ()
-> new TableNotFoundException(each.getTable()));
+ result.computeIfAbsent(schema.getName(), ignored -> new
HashSet<>()).add(each.getTable());
}
}
return result;
@@ -100,24 +108,27 @@ public final class CDCSchemaTableUtils {
private static Map<String, Set<String>>
parseTableExpressionWithAllTable(final ShardingSphereDatabase database, final
SchemaTable schemaTable) {
String schemaName = schemaTable.getSchema();
- ShardingSphereSchema schema = database.getSchema(schemaName);
+ ShardingSphereSchema schema = database.getSchema(new
IdentifierValue(schemaName));
ShardingSpherePreconditions.checkNotNull(schema, () -> new
SchemaNotFoundException(schemaTable.getSchema()));
Collection<ShardingSphereTable> tables = schema.getAllTables();
Map<String, Set<String>> result = new HashMap<>(tables.size(), 1F);
- tables.forEach(each -> result.computeIfAbsent(schemaName, ignored ->
new HashSet<>()).add(each.getName()));
+ tables.forEach(each -> result.computeIfAbsent(schema.getName(),
ignored -> new HashSet<>()).add(each.getName()));
return result;
}
- /**
- * Parse table expression without schema.
- *
- * @param database database
- * @param tableNames table names
- * @return parsed table names
- */
- public static Collection<String> parseTableExpressionWithoutSchema(final
ShardingSphereDatabase database, final List<String> tableNames) {
+ private static Map<String, Set<String>>
parseTableExpressionsWithoutSchema(final ShardingSphereDatabase database, final
Collection<SchemaTable> schemaTables) {
+ Set<String> tableNames =
schemaTables.stream().map(SchemaTable::getTable).collect(Collectors.toSet());
+ if (tableNames.isEmpty()) {
+ return Collections.emptyMap();
+ }
+ if (!tableNames.contains("*")) {
+ return Collections.singletonMap("", tableNames);
+ }
ShardingSphereSchema schema = database.getSchema(database.getName());
- Set<String> allTableNames = null == schema ? Collections.emptySet() :
new
HashSet<>(schema.getAllTables().stream().map(ShardingSphereTable::getName).collect(Collectors.toSet()));
- return tableNames.stream().anyMatch("*"::equals) ? allTableNames : new
HashSet<>(tableNames);
+ if (null == schema) {
+ return Collections.emptyMap();
+ }
+ Set<String> allTableNames =
schema.getAllTables().stream().map(ShardingSphereTable::getName).collect(Collectors.toSet());
+ return allTableNames.isEmpty() ? Collections.emptyMap() :
Collections.singletonMap("", allTableNames);
}
}
diff --git
a/kernel/data-pipeline/scenario/cdc/core/src/test/java/org/apache/shardingsphere/data/pipeline/cdc/api/CDCJobAPITest.java
b/kernel/data-pipeline/scenario/cdc/core/src/test/java/org/apache/shardingsphere/data/pipeline/cdc/api/CDCJobAPITest.java
index b4498696515..f23a98eb03d 100644
---
a/kernel/data-pipeline/scenario/cdc/core/src/test/java/org/apache/shardingsphere/data/pipeline/cdc/api/CDCJobAPITest.java
+++
b/kernel/data-pipeline/scenario/cdc/core/src/test/java/org/apache/shardingsphere/data/pipeline/cdc/api/CDCJobAPITest.java
@@ -197,14 +197,15 @@ class CDCJobAPITest {
@Test
void assertCreateSkipsExistingJob() {
putContext(Collections.singletonMap("foo_ds",
mock(StorageUnit.class)));
- CDCJobConfiguration jobConfig =
createJobConfiguration(Collections.singletonList("foo_readwrite_ds"), true);
+ List<String> schemaTableNames = Arrays.asList("foo_schema.foo_tbl",
"bar_schema.FOO_TBL");
+ CDCJobConfiguration jobConfig =
createJobConfiguration(Collections.singletonList("foo_readwrite_ds"), true,
schemaTableNames);
PipelineGovernanceFacade governanceFacade =
mock(PipelineGovernanceFacade.class, RETURNS_DEEP_STUBS);
when(governanceFacade.getJobFacade().getConfiguration().isExisted("foo_job")).thenReturn(true);
when(PipelineAPIFactory.getPipelineGovernanceFacade(any())).thenReturn(governanceFacade);
try (
MockedConstruction<YamlCDCJobConfigurationSwapper> ignored =
mockConstruction(YamlCDCJobConfigurationSwapper.class,
(mock, context) ->
when(mock.swapToObject(any(YamlCDCJobConfiguration.class))).thenReturn(jobConfig)))
{
- StreamDataParameter param = new StreamDataParameter("foo_db", new
LinkedList<>(Collections.singletonList("foo_schema.foo_tbl")), true,
+ StreamDataParameter param = new StreamDataParameter("foo_db", new
LinkedList<>(schemaTableNames), true,
Collections.singletonMap("foo_schema.foo_tbl",
Collections.singletonList(new DataNode("foo_readwrite_ds.foo_tbl"))), false);
assertThat(jobAPI.create(param, CDCSinkType.SOCKET, new
Properties()), is("foo_job"));
verify(governanceFacade.getJobFacade().getJob(),
never()).create(anyString(), any());
@@ -215,7 +216,8 @@ class CDCJobAPITest {
@Test
void assertCreateThrowsWhenDataSourceDoesNotExist() throws
ReflectiveOperationException {
putContext(Collections.singletonMap("foo_ds",
mock(StorageUnit.class)));
- CDCJobConfiguration jobConfig =
createJobConfiguration(Arrays.asList("bar_ds", "foo_readwrite_ds"), true);
+ List<String> schemaTableNames = Arrays.asList("foo_schema.foo_tbl",
"bar_schema.FOO_TBL");
+ CDCJobConfiguration jobConfig =
createJobConfiguration(Arrays.asList("bar_ds", "foo_readwrite_ds"), true,
schemaTableNames);
PipelineGovernanceFacade governanceFacade =
mock(PipelineGovernanceFacade.class, RETURNS_DEEP_STUBS);
when(PipelineAPIFactory.getPipelineGovernanceFacade(any())).thenReturn(governanceFacade);
PipelineJobConfigurationManager jobConfigManager =
mock(PipelineJobConfigurationManager.class);
@@ -224,7 +226,7 @@ class CDCJobAPITest {
try (
MockedConstruction<YamlCDCJobConfigurationSwapper> ignored =
mockConstruction(YamlCDCJobConfigurationSwapper.class,
(mock, context) ->
when(mock.swapToObject(any(YamlCDCJobConfiguration.class))).thenReturn(jobConfig)))
{
- StreamDataParameter param = new StreamDataParameter("foo_db", new
LinkedList<>(Collections.singletonList("foo_schema.foo_tbl")), true,
+ StreamDataParameter param = new StreamDataParameter("foo_db", new
LinkedList<>(schemaTableNames), true,
Collections.singletonMap("foo_schema.foo_tbl",
Arrays.asList(new DataNode("bar_ds.foo_tbl"), new
DataNode("foo_readwrite_ds.foo_tbl"))), false);
PipelineInvalidParameterException actual =
assertThrows(PipelineInvalidParameterException.class, () ->
jobAPI.create(param, CDCSinkType.SOCKET, new Properties()));
assertThat(actual.getMessage(),
containsString("foo_readwrite_ds"));
@@ -233,6 +235,25 @@ class CDCJobAPITest {
}
}
+ @Test
+ void assertCreateThrowsWhenSchemaTableNamesHaveDuplicateBareTable() {
+ putContext(Collections.singletonMap("foo_ds",
mock(StorageUnit.class)));
+ List<String> schemaTableNames = Arrays.asList("schema_a.t",
"schema_b.T");
+ CDCJobConfiguration jobConfig =
createJobConfiguration(Collections.singletonList("foo_ds"), true,
schemaTableNames);
+ PipelineGovernanceFacade governanceFacade =
mock(PipelineGovernanceFacade.class, RETURNS_DEEP_STUBS);
+
when(PipelineAPIFactory.getPipelineGovernanceFacade(any())).thenReturn(governanceFacade);
+ try (
+ MockedConstruction<YamlCDCJobConfigurationSwapper> ignored =
mockConstruction(YamlCDCJobConfigurationSwapper.class,
+ (mock, context) ->
when(mock.swapToObject(any(YamlCDCJobConfiguration.class))).thenReturn(jobConfig)))
{
+ StreamDataParameter param = new StreamDataParameter("foo_db", new
LinkedList<>(schemaTableNames), true,
+ Collections.singletonMap("t",
Collections.singletonList(new DataNode("foo_ds.t"))), false);
+ PipelineInvalidParameterException actual =
assertThrows(PipelineInvalidParameterException.class, () ->
jobAPI.create(param, CDCSinkType.SOCKET, new Properties()));
+ assertThat(actual.getMessage(), is("There is invalid parameter
value. More than one schema table has the same table name `T`."));
+ verify(governanceFacade.getJobFacade().getJob(),
never()).create(anyString(), any());
+ verify(governanceFacade.getJobFacade().getConfiguration(),
never()).persist(anyString(), any());
+ }
+ }
+
@Test
void assertCreateFullJobSkipsInitIncrementalPosition() throws
ReflectiveOperationException {
putContext(Collections.singletonMap("foo_ds",
mock(StorageUnit.class)));
@@ -441,6 +462,10 @@ class CDCJobAPITest {
}
private CDCJobConfiguration createJobConfiguration(final List<String>
dataSourceNames, final boolean full) {
+ return createJobConfiguration(dataSourceNames, full,
Collections.singletonList("foo_schema.foo_tbl"));
+ }
+
+ private CDCJobConfiguration createJobConfiguration(final List<String>
dataSourceNames, final boolean full, final List<String> schemaTableNames) {
Map<String, Map<String, Object>> dataSources = new LinkedHashMap<>(2,
1F);
dataSources.put("foo_ds", createStandardDataSourceProperties());
dataSources.put("bar_ds", createStandardDataSourceProperties());
@@ -448,7 +473,7 @@ class CDCJobAPITest {
List<JobDataNodeLine> jobDataNodeLines = dataSourceNames.stream().map(
each -> new JobDataNodeLine(Collections.singletonList(new
JobDataNodeEntry("foo_tbl", Collections.singletonList(new DataNode(each +
".foo_tbl"))))))
.collect(Collectors.toList());
- return new CDCJobConfiguration("foo_job", "foo_db",
Collections.singletonList("foo_schema.foo_tbl"), full, mock(),
+ return new CDCJobConfiguration("foo_job", "foo_db", schemaTableNames,
full, mock(),
dataSourceConfig, jobDataNodeLines.get(0), jobDataNodeLines,
false, new CDCJobConfiguration.SinkConfiguration(CDCSinkType.SOCKET, new
Properties()), 1, 0);
}
diff --git
a/kernel/data-pipeline/scenario/cdc/core/src/test/java/org/apache/shardingsphere/data/pipeline/cdc/handler/CDCBackendHandlerTest.java
b/kernel/data-pipeline/scenario/cdc/core/src/test/java/org/apache/shardingsphere/data/pipeline/cdc/handler/CDCBackendHandlerTest.java
index 77599c0cfd4..44126f1f566 100644
---
a/kernel/data-pipeline/scenario/cdc/core/src/test/java/org/apache/shardingsphere/data/pipeline/cdc/handler/CDCBackendHandlerTest.java
+++
b/kernel/data-pipeline/scenario/cdc/core/src/test/java/org/apache/shardingsphere/data/pipeline/cdc/handler/CDCBackendHandlerTest.java
@@ -67,8 +67,9 @@ import org.mockito.internal.configuration.plugins.Plugins;
import java.util.ArrayList;
import java.util.Collections;
-import java.util.LinkedHashSet;
+import java.util.Map;
import java.util.Properties;
+import java.util.Set;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.is;
@@ -76,7 +77,6 @@ import static org.hamcrest.Matchers.isA;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyCollection;
-import static org.mockito.ArgumentMatchers.anyList;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
import static org.mockito.Mockito.lenient;
@@ -117,10 +117,11 @@ class CDCBackendHandlerTest {
@Test
void assertStreamDataWhenSchemaAvailable() {
ShardingSphereDatabase database = mockDatabase();
- mockDialectMetaData(true, false);
- when(CDCSchemaTableUtils.parseTableExpressionWithSchema(eq(database),
anyCollection()))
- .thenReturn(Collections.singletonMap("foo_schema", new
LinkedHashSet<>(Collections.singleton("foo_tbl"))));
- when(PipelineDataNodeUtils.buildTableAndDataNodesMap(eq(database),
anyCollection())).thenReturn(Collections.singletonMap("foo_tbl",
Collections.singletonList(mock(DataNode.class))));
+ mockDialectMetaData(false);
+ Map<String, Set<String>> schemaTableNames =
Collections.singletonMap("foo_schema", Collections.singleton("foo_tbl"));
+ when(CDCSchemaTableUtils.parseTableExpressions(eq(database),
anyCollection())).thenReturn(schemaTableNames);
+ DataNode fooDataNode = new DataNode("foo_ds", "foo_schema", "foo_tbl");
+ when(PipelineDataNodeUtils.buildTableAndDataNodesMap(database,
schemaTableNames)).thenReturn(Collections.singletonMap("foo_tbl",
Collections.singletonList(fooDataNode)));
when(jobAPI.create(any(StreamDataParameter.class),
eq(CDCSinkType.SOCKET), any(Properties.class))).thenReturn("foo_job");
when(jobConfigManager.getJobConfiguration("foo_job")).thenReturn(createJobConfiguration());
StreamDataRequestBody requestBody =
StreamDataRequestBody.newBuilder().setDatabase("foo_db")
@@ -129,30 +130,40 @@ class CDCBackendHandlerTest {
assertThat(actualResponse.getStatus(), is(CDCResponse.Status.SUCCEED));
assertThat(actualResponse.getResponseCase(),
is(CDCResponse.ResponseCase.STREAM_DATA_RESULT));
assertThat(actualResponse.getStreamDataResult().getStreamingId(),
is("foo_job"));
+ ArgumentCaptor<StreamDataParameter> parameterCaptor =
ArgumentCaptor.forClass(StreamDataParameter.class);
+ verify(jobAPI).create(parameterCaptor.capture(),
eq(CDCSinkType.SOCKET), any(Properties.class));
+ assertThat(parameterCaptor.getValue().getSchemaTableNames(),
is(Collections.singletonList("foo_schema.foo_tbl")));
+
assertThat(parameterCaptor.getValue().getTableAndDataNodesMap().get("foo_tbl"),
is(Collections.singletonList(fooDataNode)));
}
@Test
void assertStreamDataWithoutSchema() {
ShardingSphereDatabase database = mockDatabase();
- mockDialectMetaData(false, true);
-
when(CDCSchemaTableUtils.parseTableExpressionWithoutSchema(eq(database),
anyList())).thenReturn(Collections.singleton("foo_tbl"));
- when(PipelineDataNodeUtils.buildTableAndDataNodesMap(eq(database),
anyCollection())).thenReturn(Collections.singletonMap("foo_tbl",
Collections.singletonList(mock(DataNode.class))));
+ mockDialectMetaData(true);
+ Map<String, Set<String>> schemaTableNames =
Collections.singletonMap("", Collections.singleton("foo_tbl"));
+ when(CDCSchemaTableUtils.parseTableExpressions(eq(database),
anyCollection())).thenReturn(schemaTableNames);
+ when(PipelineDataNodeUtils.buildTableAndDataNodesMap(database,
schemaTableNames)).thenReturn(Collections.singletonMap("foo_tbl",
Collections.singletonList(mock(DataNode.class))));
when(jobAPI.create(any(StreamDataParameter.class),
eq(CDCSinkType.SOCKET), any(Properties.class))).thenReturn("foo_job");
when(jobConfigManager.getJobConfiguration("foo_job")).thenReturn(createJobConfiguration());
StreamDataRequestBody requestBody =
StreamDataRequestBody.newBuilder().setDatabase("foo_db").addSourceSchemaTable(SchemaTable.newBuilder().setTable("foo_tbl")).build();
CDCResponse actualResponse = backendHandler.streamData("foo_req",
requestBody, createConnectionContext(), mock());
assertThat(actualResponse.getStatus(), is(CDCResponse.Status.SUCCEED));
assertThat(actualResponse.getStreamDataResult().getStreamingId(),
is("foo_job"));
+ ArgumentCaptor<StreamDataParameter> parameterCaptor =
ArgumentCaptor.forClass(StreamDataParameter.class);
+ verify(jobAPI).create(parameterCaptor.capture(),
eq(CDCSinkType.SOCKET), any(Properties.class));
+ assertThat(parameterCaptor.getValue().getSchemaTableNames(),
is(Collections.singletonList("foo_tbl")));
}
@Test
void assertStreamDataWhenTableNamesMissing() {
- ShardingSphereDatabase database = mockDatabase();
- mockDialectMetaData(false, false);
-
when(CDCSchemaTableUtils.parseTableExpressionWithoutSchema(eq(database),
anyList())).thenReturn(Collections.emptySet());
+ ShardingSphereDatabase database = mock(ShardingSphereDatabase.class);
+ mockProxyContext(database);
+ mockDialectMetaData(false);
+ when(CDCSchemaTableUtils.parseTableExpressions(eq(database),
anyCollection())).thenReturn(Collections.emptyMap());
StreamDataRequestBody requestBody =
StreamDataRequestBody.newBuilder().setDatabase("foo_db").addSourceSchemaTable(SchemaTable.newBuilder().setTable("foo_tbl")).build();
CDCExceptionWrapper actualException =
assertThrows(CDCExceptionWrapper.class, () ->
backendHandler.streamData("foo_req", requestBody, createConnectionContext(),
mock()));
assertThat(actualException.getCause(),
isA(MissingRequiredStreamDataSourceException.class));
+ verifyNoInteractions(jobAPI);
}
@Test
@@ -268,9 +279,8 @@ class CDCBackendHandlerTest {
when(PipelineContextManager.getProxyContext()).thenReturn(contextManager);
}
- private void mockDialectMetaData(final boolean schemaAvailable, final
boolean supportGlobalCSN) {
+ private void mockDialectMetaData(final boolean supportGlobalCSN) {
DialectDatabaseMetaData databaseMetaData =
mock(DialectDatabaseMetaData.class, RETURNS_DEEP_STUBS);
-
when(databaseMetaData.getSchemaOption().isSchemaAvailable()).thenReturn(schemaAvailable);
lenient().when(databaseMetaData.getTransactionOption().isSupportGlobalCSN()).thenReturn(supportGlobalCSN);
when(DatabaseTypedSPILoader.getService(any(),
any())).thenReturn(databaseMetaData);
}
diff --git
a/kernel/data-pipeline/scenario/cdc/core/src/test/java/org/apache/shardingsphere/data/pipeline/cdc/util/CDCSchemaTableUtilsTest.java
b/kernel/data-pipeline/scenario/cdc/core/src/test/java/org/apache/shardingsphere/data/pipeline/cdc/util/CDCSchemaTableUtilsTest.java
index 1796c0a8532..05500aeed79 100644
---
a/kernel/data-pipeline/scenario/cdc/core/src/test/java/org/apache/shardingsphere/data/pipeline/cdc/util/CDCSchemaTableUtilsTest.java
+++
b/kernel/data-pipeline/scenario/cdc/core/src/test/java/org/apache/shardingsphere/data/pipeline/cdc/util/CDCSchemaTableUtilsTest.java
@@ -27,6 +27,9 @@ import
org.apache.shardingsphere.infra.metadata.database.schema.model.ShardingSp
import
org.apache.shardingsphere.infra.metadata.database.schema.model.ShardingSphereTable;
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.ArrayList;
import java.util.Arrays;
@@ -38,6 +41,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.Properties;
+import java.util.stream.Stream;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.MatcherAssert.assertThat;
@@ -50,14 +54,14 @@ class CDCSchemaTableUtilsTest {
private final DatabaseType databaseType =
TypedSPILoader.getService(DatabaseType.class, "openGauss");
@Test
- void assertParseTableExpressionWithFullWildcard() {
+ void assertParseTableExpressionsWithFullWildcard() {
ShardingSphereSchema publicSchema = mockSchema("public", "t_order",
"t_order_item");
ShardingSphereSchema testSchema = mockSchema("test", "t_test");
ShardingSphereSchema systemSchema = mockSchema("pg_catalog", "t_pg");
ShardingSphereDatabase database =
new ShardingSphereDatabase("sharding_db", databaseType, null,
null, Arrays.asList(publicSchema, testSchema, systemSchema), new
ConfigurationProperties(new Properties()));
List<SchemaTable> schemaTables =
Collections.singletonList(SchemaTable.newBuilder().setSchema("*").setTable("*").build());
- Map<String, Set<String>> actualResult =
CDCSchemaTableUtils.parseTableExpressionWithSchema(database, schemaTables);
+ Map<String, Set<String>> actualResult =
CDCSchemaTableUtils.parseTableExpressions(database, schemaTables);
Map<String, Set<String>> expectedResult = new HashMap<>(2, 1F);
expectedResult.put("public", new HashSet<>(Arrays.asList("t_order",
"t_order_item")));
expectedResult.put("test", Collections.singleton("t_test"));
@@ -65,61 +69,139 @@ class CDCSchemaTableUtilsTest {
}
@Test
- void assertParseTableExpressionWithSchemaWildcard() {
+ void assertParseTableExpressionsWithUnqualifiedFullWildcard() {
+ ShardingSphereSchema publicSchema = mockSchema("public", "t_order");
+ ShardingSphereDatabase database =
+ new ShardingSphereDatabase("sharding_db", databaseType, null,
null, Collections.singleton(publicSchema), new ConfigurationProperties(new
Properties()));
+ List<SchemaTable> schemaTables =
Collections.singletonList(SchemaTable.newBuilder().setTable("*").build());
+ Map<String, Set<String>> actualResult =
CDCSchemaTableUtils.parseTableExpressions(database, schemaTables);
+ Map<String, Set<String>> expectedResult =
Collections.singletonMap("public", Collections.singleton("t_order"));
+ assertThat(actualResult, is(expectedResult));
+ }
+
+ @Test
+ void assertParseTableExpressionsWithSchemaWildcard() {
ShardingSphereSchema analyticsSchema = mockSchema("analytics",
"t_shared");
ShardingSphereSchema auditSchema = mockSchema("audit", "t_other");
ShardingSphereSchema systemSchema = mockSchema("pg_catalog",
"t_shared");
ShardingSphereDatabase database =
new ShardingSphereDatabase("sharding_db", databaseType, null,
null, Arrays.asList(analyticsSchema, auditSchema, systemSchema), new
ConfigurationProperties(new Properties()));
List<SchemaTable> schemaTables =
Collections.singletonList(SchemaTable.newBuilder().setSchema("*").setTable("t_shared").build());
- Map<String, Set<String>> actualResult =
CDCSchemaTableUtils.parseTableExpressionWithSchema(database, schemaTables);
+ Map<String, Set<String>> actualResult =
CDCSchemaTableUtils.parseTableExpressions(database, schemaTables);
Map<String, Set<String>> expectedResult =
Collections.singletonMap("analytics", Collections.singleton("t_shared"));
assertThat(actualResult, is(expectedResult));
}
@Test
- void assertParseTableExpressionWithAllTablesInSchema() {
- ShardingSphereSchema publicSchema = mockSchema("public", "t_order",
"t_order_item");
- ShardingSphereDatabase database = new
ShardingSphereDatabase("sharding_db", databaseType, null, null,
Collections.singleton(publicSchema), new ConfigurationProperties(new
Properties()));
- List<SchemaTable> schemaTables =
Collections.singletonList(SchemaTable.newBuilder().setSchema("public").setTable("*").build());
- Map<String, Set<String>> actualResult =
CDCSchemaTableUtils.parseTableExpressionWithSchema(database, schemaTables);
- Map<String, Set<String>> expectedResult =
Collections.singletonMap("public", new HashSet<>(Arrays.asList("t_order",
"t_order_item")));
+ void assertParseTableExpressionsWithAllTablesInSchema() {
+ ShardingSphereSchema quotedSchema = mockSchema("CaseSchema",
"t_order", "t_order_item");
+ ShardingSphereDatabase database = new
ShardingSphereDatabase("sharding_db", databaseType, null, null,
Collections.singleton(quotedSchema), new ConfigurationProperties(new
Properties()));
+ List<SchemaTable> schemaTables =
Collections.singletonList(SchemaTable.newBuilder().setSchema("\"CaseSchema\"").setTable("*").build());
+ Map<String, Set<String>> actualResult =
CDCSchemaTableUtils.parseTableExpressions(database, schemaTables);
+ Map<String, Set<String>> expectedResult =
Collections.singletonMap("CaseSchema", new HashSet<>(Arrays.asList("t_order",
"t_order_item")));
+ assertThat(actualResult, is(expectedResult));
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("provideSchemaNames")
+ void assertParseTableExpressionsResolvesCanonicalSchemaName(final String
name, final String inputSchemaName, final String expectedSchemaName) {
+ ShardingSphereSchema schema = mockSchema(expectedSchemaName,
"t_order");
+ ShardingSphereDatabase database =
+ new ShardingSphereDatabase("sharding_db", databaseType, null,
null, Collections.singleton(schema), new ConfigurationProperties(new
Properties()));
+ List<SchemaTable> schemaTables =
Collections.singletonList(SchemaTable.newBuilder().setSchema(inputSchemaName).setTable("t_order").build());
+ Map<String, Set<String>> actualResult =
CDCSchemaTableUtils.parseTableExpressions(database, schemaTables);
+ Map<String, Set<String>> expectedResult =
Collections.singletonMap(expectedSchemaName, Collections.singleton("t_order"));
+ assertThat(actualResult, is(expectedResult));
+ }
+
+ private static Stream<Arguments> provideSchemaNames() {
+ return Stream.of(
+ Arguments.of("lower-case schema", "public", "public"),
+ Arguments.of("unquoted upper-case schema", "PUBLIC", "public"),
+ Arguments.of("quoted upper-case schema", "\"PUBLIC\"",
"PUBLIC"),
+ Arguments.of("quoted mixed-case schema", "\"CaseSchema\"",
"CaseSchema"));
+ }
+
+ @Test
+ void
assertParseTableExpressionsDistinguishesQuotedAndUnquotedSchemaNames() {
+ ShardingSphereSchema unquotedSchema = mockSchema("caseschema",
"t_unquoted");
+ ShardingSphereSchema quotedSchema = mockSchema("CaseSchema",
"t_quoted");
+ ShardingSphereDatabase database = new ShardingSphereDatabase(
+ "sharding_db", databaseType, null, null,
Arrays.asList(unquotedSchema, quotedSchema), new ConfigurationProperties(new
Properties()));
+ List<SchemaTable> schemaTables = Arrays.asList(
+
SchemaTable.newBuilder().setSchema("CaseSchema").setTable("t_unquoted").build(),
+
SchemaTable.newBuilder().setSchema("\"CaseSchema\"").setTable("t_quoted").build());
+ Map<String, Set<String>> actualResult =
CDCSchemaTableUtils.parseTableExpressions(database, schemaTables);
+ Map<String, Set<String>> expectedResult = new HashMap<>(2, 1F);
+ expectedResult.put("caseschema", Collections.singleton("t_unquoted"));
+ expectedResult.put("CaseSchema", Collections.singleton("t_quoted"));
assertThat(actualResult, is(expectedResult));
}
@Test
- void assertParseTableExpressionFillDefaultSchema() {
+ void assertParseTableExpressionsFillDefaultSchema() {
ShardingSphereSchema publicSchema = mockSchema("public", "t_order");
ShardingSphereDatabase database = new
ShardingSphereDatabase("sharding_db", databaseType, null, null,
Collections.singleton(publicSchema), new ConfigurationProperties(new
Properties()));
List<SchemaTable> schemaTables =
Collections.singletonList(SchemaTable.newBuilder().setTable("t_order").build());
- Map<String, Set<String>> actualResult =
CDCSchemaTableUtils.parseTableExpressionWithSchema(database, schemaTables);
+ Map<String, Set<String>> actualResult =
CDCSchemaTableUtils.parseTableExpressions(database, schemaTables);
Map<String, Set<String>> expectedResult =
Collections.singletonMap("public", Collections.singleton("t_order"));
assertThat(actualResult, is(expectedResult));
}
@Test
- void assertParseTableExpressionWithMissingTable() {
+ void assertParseTableExpressionsWithMissingTable() {
ShardingSphereSchema publicSchema = mockSchema("public", "t_exist");
ShardingSphereDatabase database = new
ShardingSphereDatabase("sharding_db", databaseType, null, null,
Collections.singleton(publicSchema), new ConfigurationProperties(new
Properties()));
List<SchemaTable> schemaTables =
Collections.singletonList(SchemaTable.newBuilder().setSchema("public").setTable("t_missing").build());
- assertThrows(TableNotFoundException.class, () ->
CDCSchemaTableUtils.parseTableExpressionWithSchema(database, schemaTables));
+ assertThrows(TableNotFoundException.class, () ->
CDCSchemaTableUtils.parseTableExpressions(database, schemaTables));
}
@Test
- void assertParseTableExpressionWithoutSchema() {
+ void assertParseTableExpressionsWithoutSchemaWithWildcard() {
ShardingSphereSchema publicSchema = mockSchema("public", "t_order",
"t_order2");
ShardingSphereDatabase database = new ShardingSphereDatabase("public",
TypedSPILoader.getService(DatabaseType.class, "FIXTURE"), null, null,
Collections.singleton(publicSchema),
new ConfigurationProperties(new Properties()));
- List<String> tableNames = Collections.singletonList("*");
- Collection<String> actualWildcardTable =
CDCSchemaTableUtils.parseTableExpressionWithoutSchema(database, tableNames);
- Set<String> expectedWildcardTable = new
HashSet<>(Arrays.asList("t_order", "t_order2"));
- assertThat(actualWildcardTable, is(expectedWildcardTable));
+ List<SchemaTable> schemaTables =
Collections.singletonList(SchemaTable.newBuilder().setTable("*").build());
+ Map<String, Set<String>> actualResult =
CDCSchemaTableUtils.parseTableExpressions(database, schemaTables);
+ Map<String, Set<String>> expectedResult = Collections.singletonMap("",
new HashSet<>(Arrays.asList("t_order", "t_order2")));
+ assertThat(actualResult, is(expectedResult));
+ }
+
+ @Test
+ void assertParseTableExpressionsWithoutSchemaWithExplicitTable() {
ShardingSphereDatabase databaseWithoutSchema =
new ShardingSphereDatabase("missing",
TypedSPILoader.getService(DatabaseType.class, "FIXTURE"), null, null,
Collections.emptyList(), new ConfigurationProperties(new Properties()));
- List<String> singleTable = Collections.singletonList("t_order");
- Collection<String> actualTableNames =
CDCSchemaTableUtils.parseTableExpressionWithoutSchema(databaseWithoutSchema,
singleTable);
- Set<String> expectedTableNames = new HashSet<>(singleTable);
- assertThat(actualTableNames, is(expectedTableNames));
+ List<SchemaTable> schemaTables =
Collections.singletonList(SchemaTable.newBuilder().setSchema("ignored").setTable("t_order").build());
+ Map<String, Set<String>> actualResult =
CDCSchemaTableUtils.parseTableExpressions(databaseWithoutSchema, schemaTables);
+ Map<String, Set<String>> expectedResult = Collections.singletonMap("",
Collections.singleton("t_order"));
+ assertThat(actualResult, is(expectedResult));
+ }
+
+ @Test
+ void assertParseTableExpressionsWithoutSchemaWithUnmatchedWildcard() {
+ ShardingSphereDatabase databaseWithoutSchema =
+ new ShardingSphereDatabase("missing",
TypedSPILoader.getService(DatabaseType.class, "FIXTURE"), null, null,
Collections.emptyList(), new ConfigurationProperties(new Properties()));
+ List<SchemaTable> schemaTables =
Collections.singletonList(SchemaTable.newBuilder().setTable("*").build());
+ Map<String, Set<String>> actualResult =
CDCSchemaTableUtils.parseTableExpressions(databaseWithoutSchema, schemaTables);
+ assertThat(actualResult, is(Collections.emptyMap()));
+ }
+
+ @Test
+ void assertParseTableExpressionsWithoutSchemaWithEmptyMetadata() {
+ ShardingSphereSchema schema = mockSchema("foo_db");
+ ShardingSphereDatabase database = new ShardingSphereDatabase("foo_db",
TypedSPILoader.getService(DatabaseType.class, "FIXTURE"), null, null,
Collections.singleton(schema),
+ new ConfigurationProperties(new Properties()));
+ List<SchemaTable> schemaTables =
Collections.singletonList(SchemaTable.newBuilder().setTable("*").build());
+ Map<String, Set<String>> actualResult =
CDCSchemaTableUtils.parseTableExpressions(database, schemaTables);
+ assertThat(actualResult, is(Collections.emptyMap()));
+ }
+
+ @Test
+ void assertParseTableExpressionsWithEmptyExpressions() {
+ ShardingSphereDatabase database =
+ new ShardingSphereDatabase("foo_db",
TypedSPILoader.getService(DatabaseType.class, "FIXTURE"), null, null,
Collections.emptyList(), new ConfigurationProperties(new Properties()));
+ Map<String, Set<String>> actualResult =
CDCSchemaTableUtils.parseTableExpressions(database, Collections.emptyList());
+ assertThat(actualResult, is(Collections.emptyMap()));
}
private ShardingSphereSchema mockSchema(final String schemaName, final
String... tableNames) {
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 c28194c3a7c..32cc5055c00 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
@@ -57,6 +57,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.ShardingSphereIdentifier;
import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
import org.apache.shardingsphere.infra.util.json.JsonEngine;
import org.apache.shardingsphere.infra.yaml.config.pojo.YamlRootConfiguration;
@@ -82,6 +83,7 @@ import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Optional;
+import java.util.Set;
import java.util.stream.Collectors;
/**
@@ -112,11 +114,23 @@ public final class MigrationJobAPI implements
TransmissionJobAPI {
* @return job ID
*/
public String schedule(final PipelineContextKey contextKey, final
Collection<MigrationSourceTargetEntry> sourceTargetEntries, final String
targetDatabaseName) {
- MigrationJobConfiguration jobConfig = new
YamlMigrationJobConfigurationSwapper().swapToObject(buildYamlJobConfiguration(contextKey,
sourceTargetEntries, targetDatabaseName));
+ Collection<MigrationSourceTargetEntry> distinctSourceTargetEntries =
new HashSet<>(sourceTargetEntries);
+ checkSourceTableNames(distinctSourceTargetEntries);
+ MigrationJobConfiguration jobConfig = new
YamlMigrationJobConfigurationSwapper().swapToObject(buildYamlJobConfiguration(contextKey,
distinctSourceTargetEntries, targetDatabaseName));
jobManager.start(jobConfig);
return jobConfig.getJobId();
}
+ private void checkSourceTableNames(final
Collection<MigrationSourceTargetEntry> sourceTargetEntries) {
+ Map<String, Set<ShardingSphereIdentifier>> actualTableNames = new
HashMap<>(sourceTargetEntries.size(), 1F);
+ for (MigrationSourceTargetEntry each : sourceTargetEntries) {
+ String dataSourceName = each.getSource().getDataSourceName();
+ Set<ShardingSphereIdentifier> tableNames =
actualTableNames.computeIfAbsent(dataSourceName, key -> new HashSet<>());
+ ShardingSpherePreconditions.checkState(tableNames.add(new
ShardingSphereIdentifier(each.getSource().getTableName())),
+ () -> new PipelineInvalidParameterException("More than one
source table with the same table name for " + dataSourceName));
+ }
+ }
+
private YamlMigrationJobConfiguration buildYamlJobConfiguration(final
PipelineContextKey contextKey,
final
Collection<MigrationSourceTargetEntry> sourceTargetEntries, final String
targetDatabaseName) {
YamlMigrationJobConfiguration result = new
YamlMigrationJobConfiguration();
@@ -125,7 +139,7 @@ public final class MigrationJobAPI implements
TransmissionJobAPI {
Map<String, List<DataNode>> sourceDataNodes = new
LinkedHashMap<>(sourceTargetEntries.size(), 1F);
Map<String, YamlPipelineDataSourceConfiguration> configSources = new
LinkedHashMap<>(sourceTargetEntries.size(), 1F);
YamlDataSourceConfigurationSwapper dataSourceConfigSwapper = new
YamlDataSourceConfigurationSwapper();
- for (MigrationSourceTargetEntry each : new
HashSet<>(sourceTargetEntries).stream()
+ for (MigrationSourceTargetEntry each : sourceTargetEntries.stream()
.sorted(Comparator.comparing(MigrationSourceTargetEntry::getTargetTableName).thenComparing(each
-> each.getSource().format())).collect(Collectors.toList())) {
sourceDataNodes.computeIfAbsent(each.getTargetTableName(), key ->
new LinkedList<>()).add(each.getSource());
ShardingSpherePreconditions.checkState(1 ==
sourceDataNodes.get(each.getTargetTableName()).size(),
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
new file mode 100644
index 00000000000..380b123f7f0
--- /dev/null
+++
b/kernel/data-pipeline/scenario/migration/core/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/migration/api/MigrationJobAPITest.java
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shardingsphere.data.pipeline.scenario.migration.api;
+
+import org.apache.shardingsphere.data.pipeline.core.context.PipelineContextKey;
+import
org.apache.shardingsphere.data.pipeline.core.exception.param.PipelineInvalidParameterException;
+import
org.apache.shardingsphere.data.pipeline.core.job.service.PipelineJobManager;
+import
org.apache.shardingsphere.data.pipeline.core.metadata.PipelineDataSourcePersistService;
+import org.apache.shardingsphere.infra.datanode.DataNode;
+import org.apache.shardingsphere.infra.instance.metadata.InstanceType;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.internal.configuration.plugins.Plugins;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import java.util.Arrays;
+import java.util.Collection;
+
+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.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+class MigrationJobAPITest {
+
+ private final PipelineContextKey contextKey = new
PipelineContextKey(InstanceType.PROXY);
+
+ @Mock
+ private PipelineJobManager jobManager;
+
+ @Mock
+ private PipelineDataSourcePersistService dataSourcePersistService;
+
+ private MigrationJobAPI jobAPI;
+
+ @BeforeEach
+ void setUp() throws ReflectiveOperationException {
+ jobAPI = new MigrationJobAPI();
+
Plugins.getMemberAccessor().set(MigrationJobAPI.class.getDeclaredField("jobManager"),
jobAPI, jobManager);
+
Plugins.getMemberAccessor().set(MigrationJobAPI.class.getDeclaredField("dataSourcePersistService"),
jobAPI, dataSourcePersistService);
+ }
+
+ @Test
+ void assertScheduleThrowsForConflictingSourceTableIdentity() {
+ Collection<MigrationSourceTargetEntry> entries = Arrays.asList(
+ new MigrationSourceTargetEntry(new DataNode("foo_ds",
"foo_schema", "foo_tbl"), "foo_target"),
+ new MigrationSourceTargetEntry(new DataNode("foo_ds",
"bar_schema", "FOO_TBL"), "bar_target"));
+ PipelineInvalidParameterException actual =
assertThrows(PipelineInvalidParameterException.class, () ->
jobAPI.schedule(contextKey, entries, "foo_db"));
+ assertThat(actual.getMessage(), is("There is invalid parameter value.
More than one source table with the same table name for foo_ds"));
+ verify(dataSourcePersistService,
never()).load(any(PipelineContextKey.class), anyString());
+ verify(jobManager, never()).start(any());
+ }
+
+ @Test
+ void assertScheduleAcceptsExactDuplicateSourceTargetEntry() {
+ MigrationSourceTargetEntry entry = new MigrationSourceTargetEntry(new
DataNode("foo_ds", "foo_schema", "foo_tbl"), "foo_target");
+ IllegalStateException expected = new IllegalStateException("reached
configuration");
+ when(dataSourcePersistService.load(contextKey,
"MIGRATION")).thenThrow(expected);
+ IllegalStateException actual =
assertThrows(IllegalStateException.class,
+ () -> jobAPI.schedule(contextKey, Arrays.asList(entry, entry),
"foo_db"));
+ assertThat(actual, is(expected));
+ verify(dataSourcePersistService).load(contextKey, "MIGRATION");
+ verify(jobManager, never()).start(any());
+ }
+
+ @Test
+ void assertScheduleAcceptsSameTableNameFromDifferentDataSources() {
+ Collection<MigrationSourceTargetEntry> entries = Arrays.asList(
+ new MigrationSourceTargetEntry(new DataNode("foo_ds",
"foo_schema", "foo_tbl"), "foo_target"),
+ new MigrationSourceTargetEntry(new DataNode("bar_ds",
"bar_schema", "FOO_TBL"), "bar_target"));
+ IllegalStateException expected = new IllegalStateException("reached
configuration");
+ when(dataSourcePersistService.load(contextKey,
"MIGRATION")).thenThrow(expected);
+ IllegalStateException actual =
assertThrows(IllegalStateException.class, () -> jobAPI.schedule(contextKey,
entries, "foo_db"));
+ assertThat(actual, is(expected));
+ verify(dataSourcePersistService).load(contextKey, "MIGRATION");
+ verify(jobManager, never()).start(any());
+ }
+}
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 1800b3f528a..c78dd42c3a3 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
@@ -28,6 +28,8 @@ import
org.apache.shardingsphere.data.pipeline.scenario.migration.api.MigrationJ
import
org.apache.shardingsphere.data.pipeline.scenario.migration.api.MigrationSourceTargetEntry;
import
org.apache.shardingsphere.data.pipeline.scenario.migration.distsql.segment.MigrationSourceTargetSegment;
import
org.apache.shardingsphere.data.pipeline.scenario.migration.distsql.statement.updatable.MigrateTableStatement;
+import
org.apache.shardingsphere.database.connector.core.metadata.identifier.IdentifierNormalizeEngine;
+import
org.apache.shardingsphere.database.connector.core.metadata.identifier.IdentifierScope;
import
org.apache.shardingsphere.database.connector.core.type.DatabaseTypeRegistry;
import
org.apache.shardingsphere.distsql.handler.aware.DistSQLExecutorDatabaseAware;
import
org.apache.shardingsphere.distsql.handler.engine.update.DistSQLUpdateExecutor;
@@ -68,19 +70,26 @@ public final class MigrateTableExecutor implements
DistSQLUpdateExecutor<Migrate
private Collection<MigrationSourceTargetEntry>
getMigrationSourceTargetEntries(final PipelineContextKey contextKey, final
MigrateTableStatement sqlStatement) {
Collection<MigrationSourceTargetEntry> result = new LinkedList<>();
for (MigrationSourceTargetSegment each :
sqlStatement.getSourceTargetEntries()) {
- String schemaName = null == each.getSourceSchemaName() ?
getDefaultSchemaName(contextKey, each.getSourceDatabaseName()).orElse(null) :
each.getSourceSchemaName();
+ String schemaName = getSchemaName(contextKey,
each.getSourceDatabaseName(), each.getSourceSchemaName()).orElse(null);
result.add(new MigrationSourceTargetEntry(new
DataNode(each.getSourceDatabaseName(), schemaName, each.getSourceTableName()),
each.getTargetTableName()));
}
return result;
}
- private Optional<String> getDefaultSchemaName(final PipelineContextKey
contextKey, final String sourceDatabaseName) {
- if (new
DatabaseTypeRegistry(database.getProtocolType()).getDialectDatabaseMetaData().getSchemaOption().isSchemaAvailable())
{
- Map<String, DataSourcePoolProperties> metaDataDataSource = new
PipelineDataSourcePersistService().load(contextKey, "MIGRATION");
- Map<String, Object> sourceDataSourcePoolProps = new
YamlDataSourceConfigurationSwapper().swapToMap(metaDataDataSource.get(sourceDatabaseName));
- return Optional.of(PipelineSchemaUtils.getDefaultSchema(new
StandardPipelineDataSourceConfiguration(sourceDataSourcePoolProps)));
+ private Optional<String> getSchemaName(final PipelineContextKey
contextKey, final String sourceDatabaseName, final String sourceSchemaName) {
+ Map<String, DataSourcePoolProperties> metaDataDataSource = new
PipelineDataSourcePersistService().load(contextKey, "MIGRATION");
+ DataSourcePoolProperties sourceDataSourcePoolProps =
metaDataDataSource.get(sourceDatabaseName);
+ if (null == sourceDataSourcePoolProps) {
+ return Optional.ofNullable(sourceSchemaName);
}
- return Optional.empty();
+ Map<String, Object> sourceDataSourceProps = new
YamlDataSourceConfigurationSwapper().swapToMap(sourceDataSourcePoolProps);
+ StandardPipelineDataSourceConfiguration sourceDataSourceConfig = new
StandardPipelineDataSourceConfiguration(sourceDataSourceProps);
+ if (!new
DatabaseTypeRegistry(sourceDataSourceConfig.getDatabaseType()).getDialectDatabaseMetaData().getSchemaOption().isSchemaAvailable())
{
+ return Optional.ofNullable(sourceSchemaName);
+ }
+ return Optional.ofNullable(null == sourceSchemaName
+ ? PipelineSchemaUtils.getDefaultSchema(sourceDataSourceConfig)
+ :
IdentifierNormalizeEngine.normalize(IdentifierNormalizeEngine.resolvePolicy(sourceDataSourceConfig.getDatabaseType(),
null, IdentifierScope.SCHEMA), sourceSchemaName));
}
@Override
diff --git
a/kernel/data-pipeline/scenario/migration/distsql/handler/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/migration/distsql/handler/update/MigrateTableExecutorTest.java
b/kernel/data-pipeline/scenario/migration/distsql/handler/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/migration/distsql/handler/update/MigrateTableExecutorTest.java
new file mode 100644
index 00000000000..6f4184fea51
--- /dev/null
+++
b/kernel/data-pipeline/scenario/migration/distsql/handler/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/migration/distsql/handler/update/MigrateTableExecutorTest.java
@@ -0,0 +1,214 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package
org.apache.shardingsphere.data.pipeline.scenario.migration.distsql.handler.update;
+
+import
org.apache.shardingsphere.data.pipeline.api.type.StandardPipelineDataSourceConfiguration;
+import org.apache.shardingsphere.data.pipeline.core.context.PipelineContextKey;
+import org.apache.shardingsphere.data.pipeline.core.job.api.TransmissionJobAPI;
+import
org.apache.shardingsphere.data.pipeline.core.metadata.PipelineDataSourcePersistService;
+import
org.apache.shardingsphere.data.pipeline.core.metadata.loader.PipelineSchemaUtils;
+import
org.apache.shardingsphere.data.pipeline.scenario.migration.api.MigrationJobAPI;
+import
org.apache.shardingsphere.data.pipeline.scenario.migration.api.MigrationSourceTargetEntry;
+import
org.apache.shardingsphere.data.pipeline.scenario.migration.distsql.segment.MigrationSourceTargetSegment;
+import
org.apache.shardingsphere.data.pipeline.scenario.migration.distsql.statement.updatable.MigrateTableStatement;
+import
org.apache.shardingsphere.database.connector.core.metadata.database.metadata.DialectDatabaseMetaData;
+import
org.apache.shardingsphere.database.connector.core.metadata.identifier.IdentifierCasePolicy;
+import
org.apache.shardingsphere.database.connector.core.metadata.identifier.IdentifierCasePolicyFactory;
+import
org.apache.shardingsphere.database.connector.core.metadata.identifier.IdentifierNormalizeEngine;
+import
org.apache.shardingsphere.database.connector.core.metadata.identifier.IdentifierScope;
+import org.apache.shardingsphere.database.connector.core.type.DatabaseType;
+import
org.apache.shardingsphere.database.connector.core.type.DatabaseTypeRegistry;
+import
org.apache.shardingsphere.infra.datasource.pool.props.domain.DataSourcePoolProperties;
+import
org.apache.shardingsphere.infra.metadata.database.ShardingSphereDatabase;
+import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
+import
org.apache.shardingsphere.infra.yaml.config.swapper.resource.YamlDataSourceConfigurationSwapper;
+import org.apache.shardingsphere.mode.manager.ContextManager;
+import org.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.ArgumentCaptor;
+import org.mockito.Captor;
+import org.mockito.Mock;
+import org.mockito.MockedConstruction;
+import org.mockito.MockedStatic;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+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.assertNull;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+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;
+
+@ExtendWith(MockitoExtension.class)
+class MigrateTableExecutorTest {
+
+ private static final String SOURCE_DATABASE_NAME = "foo_source";
+
+ private static final String TARGET_DATABASE_NAME = "foo_target";
+
+ private final MigrateTableExecutor executor = new MigrateTableExecutor();
+
+ @Mock
+ private ShardingSphereDatabase database;
+
+ @Mock(answer = Answers.RETURNS_DEEP_STUBS)
+ private ContextManager contextManager;
+
+ @Mock
+ private MigrationJobAPI jobAPI;
+
+ @Mock
+ private DatabaseType schemaCapableDatabaseType;
+
+ @Mock
+ private DatabaseType schemaLessDatabaseType;
+
+ @Mock
+ private DataSourcePoolProperties sourceProps;
+
+ @Captor
+ private ArgumentCaptor<Collection<MigrationSourceTargetEntry>>
entriesCaptor;
+
+ @BeforeEach
+ void setUp() {
+ executor.setDatabase(database);
+ when(database.getName()).thenReturn(TARGET_DATABASE_NAME);
+
when(contextManager.getMetaDataContexts().getMetaData().containsDatabase(TARGET_DATABASE_NAME)).thenReturn(true);
+ }
+
+ @Test
+ void assertExecuteUpdateUsesSourceDatabaseTypeForDefaultSchema() {
+ mockSourceProps();
+ try (
+ MockedConstruction<PipelineDataSourcePersistService> ignored =
mockConstruction(PipelineDataSourcePersistService.class,
+ (mock, context) ->
when(mock.load(any(PipelineContextKey.class),
eq("MIGRATION"))).thenReturn(Collections.singletonMap(SOURCE_DATABASE_NAME,
sourceProps)));
+ MockedConstruction<StandardPipelineDataSourceConfiguration>
ignoredConfig = mockConstruction(StandardPipelineDataSourceConfiguration.class,
+ (mock, context) ->
when(mock.getDatabaseType()).thenReturn(schemaCapableDatabaseType));
+ MockedConstruction<DatabaseTypeRegistry> ignoredRegistry =
mockDatabaseTypeRegistries();
+ MockedStatic<TypedSPILoader> spiLoader =
mockStatic(TypedSPILoader.class, Answers.CALLS_REAL_METHODS);
+ MockedStatic<PipelineSchemaUtils> schemaUtils =
mockStatic(PipelineSchemaUtils.class)) {
+ spiLoader.when(() ->
TypedSPILoader.getService(TransmissionJobAPI.class,
"MIGRATION")).thenReturn(jobAPI);
+ schemaUtils.when(() ->
PipelineSchemaUtils.getDefaultSchema(any(StandardPipelineDataSourceConfiguration.class))).thenReturn("CaseSchema");
+ executor.executeUpdate(createStatement(null), contextManager);
+ verify(jobAPI).schedule(any(PipelineContextKey.class),
entriesCaptor.capture(), eq(TARGET_DATABASE_NAME));
+
assertThat(entriesCaptor.getValue().iterator().next().getSource().getSchemaName(),
is("CaseSchema"));
+ verify(database, never()).getProtocolType();
+ }
+ }
+
+ @Test
+ void assertExecuteUpdateDoesNotLoadDefaultSchemaForSchemaLessSource() {
+ mockSourceProps();
+ try (
+ MockedConstruction<PipelineDataSourcePersistService> ignored =
mockConstruction(PipelineDataSourcePersistService.class,
+ (mock, context) ->
when(mock.load(any(PipelineContextKey.class),
eq("MIGRATION"))).thenReturn(Collections.singletonMap(SOURCE_DATABASE_NAME,
sourceProps)));
+ MockedConstruction<StandardPipelineDataSourceConfiguration>
ignoredConfig = mockConstruction(StandardPipelineDataSourceConfiguration.class,
+ (mock, context) ->
when(mock.getDatabaseType()).thenReturn(schemaLessDatabaseType));
+ MockedConstruction<DatabaseTypeRegistry> ignoredRegistry =
mockDatabaseTypeRegistries();
+ MockedStatic<TypedSPILoader> spiLoader =
mockStatic(TypedSPILoader.class, Answers.CALLS_REAL_METHODS);
+ MockedStatic<PipelineSchemaUtils> schemaUtils =
mockStatic(PipelineSchemaUtils.class)) {
+ spiLoader.when(() ->
TypedSPILoader.getService(TransmissionJobAPI.class,
"MIGRATION")).thenReturn(jobAPI);
+ schemaUtils.when(() ->
PipelineSchemaUtils.getDefaultSchema(any(StandardPipelineDataSourceConfiguration.class))).thenReturn("public");
+ executor.executeUpdate(createStatement(null), contextManager);
+ verify(jobAPI).schedule(any(PipelineContextKey.class),
entriesCaptor.capture(), eq(TARGET_DATABASE_NAME));
+
assertNull(entriesCaptor.getValue().iterator().next().getSource().getSchemaName());
+ schemaUtils.verify(() ->
PipelineSchemaUtils.getDefaultSchema(any(StandardPipelineDataSourceConfiguration.class)),
never());
+ verify(database, never()).getProtocolType();
+ }
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("provideExplicitSchemaNames")
+ void assertExecuteUpdateCanonicalizesExplicitSchema(final String name,
final String sourceSchemaName, final String expectedSchemaName) {
+ IdentifierCasePolicy identifierCasePolicy =
IdentifierCasePolicyFactory.newLowerCasePolicySet().getPolicy(IdentifierScope.SCHEMA);
+ try (
+ MockedConstruction<PipelineDataSourcePersistService> ignored =
mockConstruction(PipelineDataSourcePersistService.class,
+ (mock, context) ->
when(mock.load(any(PipelineContextKey.class),
eq("MIGRATION"))).thenReturn(Collections.singletonMap(SOURCE_DATABASE_NAME,
sourceProps)));
+ MockedConstruction<YamlDataSourceConfigurationSwapper>
ignoredSwapper = mockConstruction(YamlDataSourceConfigurationSwapper.class,
+ (mock, context) ->
when(mock.swapToMap(sourceProps)).thenReturn(Collections.emptyMap()));
+ MockedConstruction<StandardPipelineDataSourceConfiguration>
ignoredConfig = mockConstruction(StandardPipelineDataSourceConfiguration.class,
+ (mock, context) ->
when(mock.getDatabaseType()).thenReturn(schemaCapableDatabaseType));
+ MockedConstruction<DatabaseTypeRegistry> ignoredRegistry =
mockDatabaseTypeRegistries();
+ MockedStatic<TypedSPILoader> spiLoader =
mockStatic(TypedSPILoader.class, Answers.CALLS_REAL_METHODS);
+ MockedStatic<PipelineSchemaUtils> schemaUtils =
mockStatic(PipelineSchemaUtils.class);
+ MockedStatic<IdentifierNormalizeEngine> normalizeEngine =
mockStatic(IdentifierNormalizeEngine.class, Answers.CALLS_REAL_METHODS)) {
+ spiLoader.when(() ->
TypedSPILoader.getService(TransmissionJobAPI.class,
"MIGRATION")).thenReturn(jobAPI);
+ normalizeEngine.when(() ->
IdentifierNormalizeEngine.resolvePolicy(schemaCapableDatabaseType, null,
IdentifierScope.SCHEMA)).thenReturn(identifierCasePolicy);
+ executor.executeUpdate(createStatement(sourceSchemaName),
contextManager);
+ verify(jobAPI).schedule(any(PipelineContextKey.class),
entriesCaptor.capture(), eq(TARGET_DATABASE_NAME));
+
assertThat(entriesCaptor.getValue().iterator().next().getSource().getSchemaName(),
is(expectedSchemaName));
+ schemaUtils.verify(() ->
PipelineSchemaUtils.getDefaultSchema(any(StandardPipelineDataSourceConfiguration.class)),
never());
+ }
+ }
+
+ private static Stream<Arguments> provideExplicitSchemaNames() {
+ return Stream.of(
+ Arguments.of("lower-case schema", "foo_schema", "foo_schema"),
+ Arguments.of("unquoted upper-case schema", "FOO_SCHEMA",
"foo_schema"),
+ Arguments.of("quoted upper-case schema", "\"FOO_SCHEMA\"",
"FOO_SCHEMA"),
+ Arguments.of("quoted mixed-case schema", "\"CaseSchema\"",
"CaseSchema"));
+ }
+
+ @Test
+ void assertExecuteUpdateDefersMissingSourceValidation() {
+ 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);
+ MockedStatic<PipelineSchemaUtils> schemaUtils =
mockStatic(PipelineSchemaUtils.class)) {
+ spiLoader.when(() ->
TypedSPILoader.getService(TransmissionJobAPI.class,
"MIGRATION")).thenReturn(jobAPI);
+ executor.executeUpdate(createStatement(null), contextManager);
+ verify(jobAPI).schedule(any(PipelineContextKey.class),
entriesCaptor.capture(), eq(TARGET_DATABASE_NAME));
+
assertNull(entriesCaptor.getValue().iterator().next().getSource().getSchemaName());
+ schemaUtils.verify(() ->
PipelineSchemaUtils.getDefaultSchema(any(StandardPipelineDataSourceConfiguration.class)),
never());
+ verify(database, never()).getProtocolType();
+ }
+ }
+
+ private MigrateTableStatement createStatement(final String
sourceSchemaName) {
+ MigrationSourceTargetSegment entry = new
MigrationSourceTargetSegment(SOURCE_DATABASE_NAME, sourceSchemaName, "foo_tbl",
"foo_tbl");
+ return new MigrateTableStatement(null, Collections.singleton(entry));
+ }
+
+ private void mockSourceProps() {
+
when(sourceProps.getAllStandardProperties()).thenReturn(Collections.emptyMap());
+
when(sourceProps.getPoolClassName()).thenReturn("com.zaxxer.hikari.HikariDataSource");
+ }
+
+ private MockedConstruction<DatabaseTypeRegistry>
mockDatabaseTypeRegistries() {
+ return mockConstruction(DatabaseTypeRegistry.class, (mock, context) ->
{
+ DialectDatabaseMetaData databaseMetaData =
mock(DialectDatabaseMetaData.class, Answers.RETURNS_DEEP_STUBS);
+
when(databaseMetaData.getSchemaOption().isSchemaAvailable()).thenReturn(schemaCapableDatabaseType
== context.arguments().get(0));
+
when(mock.getDialectDatabaseMetaData()).thenReturn(databaseMetaData);
+ });
+ }
+}
diff --git
a/test/e2e/operation/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/operation/pipeline/cases/migration/general/PostgreSQLMigrationGeneralE2EIT.java
b/test/e2e/operation/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/operation/pipeline/cases/migration/general/PostgreSQLMigrationGeneralE2EIT.java
index 819945b8d29..1d92f7eb410 100644
---
a/test/e2e/operation/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/operation/pipeline/cases/migration/general/PostgreSQLMigrationGeneralE2EIT.java
+++
b/test/e2e/operation/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/operation/pipeline/cases/migration/general/PostgreSQLMigrationGeneralE2EIT.java
@@ -47,6 +47,7 @@ import java.util.concurrent.TimeUnit;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
@PipelineE2ESettings(database = {
@@ -57,6 +58,8 @@ class PostgreSQLMigrationGeneralE2EIT extends
AbstractMigrationE2EIT {
private static final String SOURCE_TABLE_NAME = "t_order";
+ private static final long CROSS_SCHEMA_SENTINEL_ORDER_ID = 20000L;
+
@ParameterizedTest(name = "{0}")
@EnabledIf("isEnabled")
@ArgumentsSource(PipelineE2ETestCaseArgumentsProvider.class)
@@ -68,6 +71,8 @@ class PostgreSQLMigrationGeneralE2EIT extends
AbstractMigrationE2EIT {
IntPkLargeOrderDAO orderDAO = new
IntPkLargeOrderDAO(containerComposer.getSourceDataSource(),
containerComposer.getDatabaseType(),
containerComposer.createQualifiedTableWithSchema(SOURCE_TABLE_NAME).format());
orderDAO.createTable();
+ IntPkLargeOrderDAO publicOrderDAO = new
IntPkLargeOrderDAO(containerComposer.getSourceDataSource(),
containerComposer.getDatabaseType(), "public." + SOURCE_TABLE_NAME);
+ publicOrderDAO.createTable();
IntPkOrderItemDAO orderItemDAO = new
IntPkOrderItemDAO(containerComposer.getSourceDataSource(),
containerComposer.getDatabaseType(), PipelineContainerComposer.SCHEMA_NAME);
orderItemDAO.createTable();
containerComposer.createSourceTableIndexList(PipelineContainerComposer.SCHEMA_NAME,
SOURCE_TABLE_NAME);
@@ -85,6 +90,8 @@ class PostgreSQLMigrationGeneralE2EIT extends
AbstractMigrationE2EIT {
Awaitility.waitAtMost(10L, TimeUnit.SECONDS).pollInterval(1L,
TimeUnit.SECONDS).until(() -> !distSQLFacade.listJobIds().isEmpty());
String jobId = distSQLFacade.getJobIdByTableName("ds_0.test." +
SOURCE_TABLE_NAME);
distSQLFacade.waitJobPreparingStageFinished(jobId);
+ distSQLFacade.waitJobIncrementalStageStarted(jobId);
+ publicOrderDAO.insert(CROSS_SCHEMA_SENTINEL_ORDER_ID, 1, "OK");
String qualifiedTableName = String.join(".",
PipelineContainerComposer.SCHEMA_NAME, SOURCE_TABLE_NAME);
containerComposer.startIncrementTask(new
E2EIncrementalTask(containerComposer.getSourceDataSource(), qualifiedTableName,
new SnowflakeKeyGenerateAlgorithm(),
containerComposer.getDatabaseType(), 20));
@@ -93,6 +100,7 @@ class PostgreSQLMigrationGeneralE2EIT extends
AbstractMigrationE2EIT {
// TODO Insert new record in t_order_item
DataSource jdbcDataSource =
containerComposer.generateShardingSphereDataSourceFromProxy();
containerComposer.assertRecordExists(jdbcDataSource,
qualifiedTableName, 10000);
+ assertRecordAbsent(jdbcDataSource, qualifiedTableName,
CROSS_SCHEMA_SENTINEL_ORDER_ID);
checkOrderMigration(distSQLFacade, jobId);
startMigrationWithSchema(containerComposer, "t_order_item",
"t_order_item");
checkOrderItemMigration(distSQLFacade);
@@ -106,6 +114,15 @@ class PostgreSQLMigrationGeneralE2EIT extends
AbstractMigrationE2EIT {
}
}
+ private void assertRecordAbsent(final DataSource dataSource, final String
tableName, final long orderId) throws SQLException {
+ try (
+ Connection connection = dataSource.getConnection();
+ Statement statement = connection.createStatement();
+ ResultSet resultSet =
statement.executeQuery(String.format("SELECT 1 FROM %s WHERE order_id = %d",
tableName, orderId))) {
+ assertFalse(resultSet.next(), "Cross-schema sentinel must not be
migrated");
+ }
+ }
+
private void checkOrderMigration(final PipelineE2EDistSQLFacade
distSQLFacade, final String jobId) throws SQLException {
distSQLFacade.waitJobIncrementalStageFinished(jobId);
distSQLFacade.pauseJob(jobId);
