Skip to content

Commit

Permalink
Fix test cases
Browse files Browse the repository at this point in the history
  • Loading branch information
@yuxiqian
yuxiqian committed Jul 17, 2024
1 parent 13746a0 commit afa70ae
Showing 1 changed file with 23 additions and 3 deletions.
26 changes: 23 additions & 3 deletions .../java/org/apache/flink/cdc/runtime/operators/schema/coordinator/SchemaDerivationTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -147,8 +147,9 @@ void testOneToOneMapping() {

@Test
void testMergingTablesWithExactSameSchema() {
SchemaManager schemaManager = new SchemaManager();
SchemaDerivation schemaDerivation =
new SchemaDerivation(new SchemaManager(), ROUTES, new HashMap<>());
new SchemaDerivation(schemaManager, ROUTES, new HashMap<>());

// Create table 1
List<SchemaChangeEvent> derivedChangesAfterCreateTable =
Expand All @@ -158,6 +159,8 @@ void testMergingTablesWithExactSameSchema() {
.asCreateTableEvent()
.hasTableId(MERGED_TABLE)
.hasSchema(SCHEMA);
derivedChangesAfterCreateTable.forEach(schemaManager::applyEvolvedSchemaChange);

// Create table 2
assertThat(schemaDerivation.applySchemaChange(new CreateTableEvent(TABLE_2, SCHEMA)))
.isEmpty();
Expand All @@ -177,6 +180,8 @@ void testMergingTablesWithExactSameSchema() {
.asAddColumnEvent()
.hasTableId(MERGED_TABLE)
.containsAddedColumns(newCol1, newCol2);
derivedChangesAfterAddColumn.forEach(schemaManager::applyEvolvedSchemaChange);

// Add column for table 2
assertThat(schemaDerivation.applySchemaChange(new AddColumnEvent(TABLE_2, newColumns)))
.isEmpty();
Expand All @@ -190,6 +195,8 @@ void testMergingTablesWithExactSameSchema() {
.asAlterColumnTypeEvent()
.hasTableId(MERGED_TABLE)
.containsTypeMapping(typeMapping);
derivedChangesAfterAlterColumnType.forEach(schemaManager::applyEvolvedSchemaChange);

// Alter column type for table 2
assertThat(
schemaDerivation.applySchemaChange(
Expand All @@ -215,6 +222,8 @@ void testMergingTablesWithExactSameSchema() {
.containsAddedColumns(
new AddColumnEvent.ColumnWithPosition(
new PhysicalColumn("last_name", DataTypes.STRING(), null)));
derivedChangesAfterRenameColumn.forEach(schemaManager::applyEvolvedSchemaChange);

// Rename column for table 2
assertThat(
schemaDerivation.applySchemaChange(
Expand All @@ -235,6 +244,8 @@ void testMergingTableWithDifferentSchemas() {
.asCreateTableEvent()
.hasTableId(MERGED_TABLE)
.hasSchema(SCHEMA);
derivedChangesAfterCreateTable.forEach(schemaManager::applyEvolvedSchemaChange);

// Create table 2
List<SchemaChangeEvent> derivedChangesAfterCreateTable2 =
schemaDerivation.applySchemaChange(
Expand All @@ -250,6 +261,7 @@ void testMergingTableWithDifferentSchemas() {
"gender", DataTypes.STRING(), null)))),
new AlterColumnTypeEvent(
MERGED_TABLE, ImmutableMap.of("age", DataTypes.BIGINT())));
derivedChangesAfterCreateTable2.forEach(schemaManager::applyEvolvedSchemaChange);

// Add column for table 1
AddColumnEvent.ColumnWithPosition newCol1 =
Expand All @@ -266,6 +278,8 @@ void testMergingTableWithDifferentSchemas() {
.asAddColumnEvent()
.hasTableId(MERGED_TABLE)
.containsAddedColumns(newCol1, newCol2);
derivedChangesAfterAddColumn.forEach(schemaManager::applyEvolvedSchemaChange);

// Add column for table 2
List<SchemaChangeEvent> derivedChangesAfterAddColumnForTable2 =
schemaDerivation.applySchemaChange(
Expand All @@ -284,6 +298,7 @@ void testMergingTableWithDifferentSchemas() {
.containsTypeMapping(
ImmutableMap.of(
"new_col1", DataTypes.STRING(), "new_col2", DataTypes.STRING()));
derivedChangesAfterAddColumnForTable2.forEach(schemaManager::applyEvolvedSchemaChange);

// Alter column type for table 1
ImmutableMap<String, DataType> typeMapping = ImmutableMap.of("age", DataTypes.BIGINT());
Expand Down Expand Up @@ -316,6 +331,8 @@ void testMergingTableWithDifferentSchemas() {
.containsAddedColumns(
new AddColumnEvent.ColumnWithPosition(
new PhysicalColumn("last_name", DataTypes.STRING(), null)));
derivedChangesAfterRenameColumn.forEach(schemaManager::applyEvolvedSchemaChange);

// Rename column for table 2
List<SchemaChangeEvent> derivedChangesAfterRenameColumnForTable2 =
schemaDerivation.applySchemaChange(
Expand All @@ -327,8 +344,9 @@ void testMergingTableWithDifferentSchemas() {
.containsAddedColumns(
new AddColumnEvent.ColumnWithPosition(
new PhysicalColumn("first_name", DataTypes.STRING(), null)));
derivedChangesAfterRenameColumnForTable2.forEach(schemaManager::applyEvolvedSchemaChange);

assertThat(schemaManager.getLatestUpstreamSchema(MERGED_TABLE))
assertThat(schemaManager.getLatestEvolvedSchema(MERGED_TABLE))
.contains(
Schema.newBuilder()
.column(Column.physicalColumn("id", DataTypes.BIGINT()))
Expand All @@ -344,8 +362,9 @@ void testMergingTableWithDifferentSchemas() {

@Test
void testIncompatibleTypes() {
SchemaManager schemaManager = new SchemaManager();
SchemaDerivation schemaDerivation =
new SchemaDerivation(new SchemaManager(), ROUTES, new HashMap<>());
new SchemaDerivation(schemaManager, ROUTES, new HashMap<>());
// Create table 1
List<SchemaChangeEvent> derivedChangesAfterCreateTable =
schemaDerivation.applySchemaChange(new CreateTableEvent(TABLE_1, SCHEMA));
Expand All @@ -354,6 +373,7 @@ void testIncompatibleTypes() {
.asCreateTableEvent()
.hasTableId(MERGED_TABLE)
.hasSchema(SCHEMA);
derivedChangesAfterCreateTable.forEach(schemaManager::applyEvolvedSchemaChange);

// Create table 2
assertThatThrownBy(
Expand Down

0 comments on commit afa70ae

Please sign in to comment.