github-forks
/
flink-cdc-connectors
mirror of https://github.com/ververica/flink-cdc-connectors.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fix comment

Browse Source
pull/3865/head
MOBIN-F 2 weeks ago
parent 440d1d28f1
commit 866f0abe59
4 changed files with 83 additions and 63 deletions
Show Stats Download Patch File Download Diff File

38
flink-cdc-common/src/main/java/org/apache/flink/cdc/common/utils/SchemaMergingUtils.java
Unescape Escape View File

@ -71,16 +71,13 @@ import java.time.ZoneId;
import java.time.ZoneOffset; import java.time.ZoneOffset;
import java.time.ZonedDateTime; import java.time.ZonedDateTime;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections; import java.util.Collections;
import java.util.HashMap; import java.util.HashMap;
import java.util.LinkedHashSet;
import java.util.List; import java.util.List;
import java.util.Map; import java.util.Map;
import java.util.Objects; import java.util.Objects;
import java.util.stream.Collectors; import java.util.stream.Collectors;
import java.util.stream.IntStream;
import static org.apache.flink.cdc.common.utils.Preconditions.checkState;
/** /**
* Utils for merging {@link Schema}s and {@link DataType}s. Prefer using this over {@link * Utils for merging {@link Schema}s and {@link DataType}s. Prefer using this over {@link
@ -173,13 +170,17 @@ public class SchemaMergingUtils {
} }
/** Merge compatible schemas. */ /** Merge compatible schemas. */
public static Schema getCommonSchema(LinkedHashSet<Schema> schemas) { public static Schema getCommonSchema(Collection<Schema> schemas) {
if (schemas.size() == 1) { if (schemas.size() == 1) {
return schemas.iterator().next(); return schemas.iterator().next();
} else { } else {
Schema outputSchema = null; Schema outputSchema = null;
for (Schema schema : schemas) { for (Schema schema : schemas) {
try {
validateTransformColumn(outputSchema, schema); validateTransformColumn(outputSchema, schema);
} catch (SchemaUtils.SchemaValidationException e) {
throw new IllegalStateException("Schema validation failed.", e);
}
outputSchema = getLeastCommonSchema(outputSchema, schema); outputSchema = getLeastCommonSchema(outputSchema, schema);
} }
return outputSchema; return outputSchema;
@ -187,35 +188,34 @@ public class SchemaMergingUtils {
} }
public static void validateTransformColumn( public static void validateTransformColumn(
@Nullable Schema currentSchema, Schema upcomingSchema) { @Nullable Schema currentSchema, Schema upcomingSchema)
throws SchemaUtils.SchemaValidationException {
if (currentSchema != null) { if (currentSchema != null) {
checkState( if (currentSchema.getColumnCount() != upcomingSchema.getColumnCount()) {
currentSchema.getColumnCount() == upcomingSchema.getColumnCount(), throw new SchemaUtils.SchemaValidationException(
String.format( String.format(
"Unable to merge schema %s and %s with different column counts.", "Unable to merge schema %s and %s with different column counts.",
currentSchema, upcomingSchema)); currentSchema, upcomingSchema));
}
List<Column> currentColumns = currentSchema.getColumns(); List<Column> currentColumns = currentSchema.getColumns();
List<Column> upcomingColumns = upcomingSchema.getColumns(); List<Column> upcomingColumns = upcomingSchema.getColumns();
IntStream.range(0, currentColumns.size()) for (int i = 0; i < currentColumns.size(); i++) {
.forEach(
i -> {
Column currentColumn = currentColumns.get(i); Column currentColumn = currentColumns.get(i);
Column upcomingColumn = upcomingColumns.get(i); Column upcomingColumn = upcomingColumns.get(i);
checkState( if (!Objects.equals(currentColumn.getName(), upcomingColumn.getName())) {
Objects.equals( throw new SchemaUtils.SchemaValidationException(
currentColumn.getName(), upcomingColumn.getName()),
String.format( String.format(
"Unable to merge column %s and %s with different name.", "Unable to merge column %s and %s with different name.",
currentColumn, upcomingColumn)); currentColumn, upcomingColumn));
checkState( }
Objects.equals( if (!Objects.equals(currentColumn.getComment(), upcomingColumn.getComment())) {
currentColumn.getComment(), throw new SchemaUtils.SchemaValidationException(
upcomingColumn.getComment()),
String.format( String.format(
"Unable to merge column %s and %s with different comments.", "Unable to merge column %s and %s with different comments.",
currentColumn, upcomingColumn)); currentColumn, upcomingColumn));
}); }
}
} }
} }

43
flink-cdc-common/src/main/java/org/apache/flink/cdc/common/utils/SchemaUtils.java
Unescape Escape View File

@ -41,7 +41,7 @@ import javax.annotation.CheckReturnValue;
import javax.annotation.Nullable; import javax.annotation.Nullable;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.LinkedHashSet; import java.util.Collection;
import java.util.LinkedList; import java.util.LinkedList;
import java.util.List; import java.util.List;
import java.util.Map; import java.util.Map;
@ -50,8 +50,6 @@ import java.util.Optional;
import java.util.stream.Collectors; import java.util.stream.Collectors;
import java.util.stream.IntStream; import java.util.stream.IntStream;
import static org.apache.flink.cdc.common.utils.Preconditions.checkState;
/** Utils for {@link Schema} to perform the ability of evolution. */ /** Utils for {@link Schema} to perform the ability of evolution. */
@PublicEvolving @PublicEvolving
public class SchemaUtils { public class SchemaUtils {
@ -455,42 +453,54 @@ public class SchemaUtils {
return lSchema.copy(mergedColumns); return lSchema.copy(mergedColumns);
} }
public static void validateMetaSchemaCompatibility(LinkedHashSet<Schema> schemas) { public static void validateMetaSchemaCompatibility(Collection<Schema> schemas) {
if (schemas.size() > 1) { if (schemas.size() > 1) {
Schema outputSchema = null; Schema outputSchema = null;
for (Schema schema : schemas) { for (Schema schema : schemas) {
try {
validateMetaSchemaCompatible(outputSchema, schema); validateMetaSchemaCompatible(outputSchema, schema);
} catch (SchemaValidationException e) {
throw new IllegalStateException("Schema validation failed.", e);
}
outputSchema = schema; outputSchema = schema;
} }
} }
} }
public static void validateMetaSchemaCompatible( public static void validateMetaSchemaCompatible(
@Nullable Schema currentSchema, Schema upcomingSchema) { @Nullable Schema currentSchema, Schema upcomingSchema)
throws SchemaValidationException {
if (currentSchema == null) { if (currentSchema == null) {
return; return;
} }
checkState(
currentSchema.primaryKeys().equals(upcomingSchema.primaryKeys()), if (!currentSchema.primaryKeys().equals(upcomingSchema.primaryKeys())) {
throw new SchemaValidationException(
String.format( String.format(
"Unable to merge schema %s and %s with different primary keys.", "Unable to merge schema %s and %s with different primary keys.",
currentSchema, upcomingSchema)); currentSchema, upcomingSchema));
checkState( }
currentSchema.partitionKeys().equals(upcomingSchema.partitionKeys()),
if (!currentSchema.partitionKeys().equals(upcomingSchema.partitionKeys())) {
throw new SchemaValidationException(
String.format( String.format(
"Unable to merge schema %s and %s with different partition keys.", "Unable to merge schema %s and %s with different partition keys.",
currentSchema, upcomingSchema)); currentSchema, upcomingSchema));
checkState( }
currentSchema.options().equals(upcomingSchema.options()), if (!currentSchema.options().equals(upcomingSchema.options())) {
throw new SchemaValidationException(
String.format( String.format(
"Unable to merge schema %s and %s with different options.", "Unable to merge schema %s and %s with different options.",
currentSchema, upcomingSchema)); currentSchema, upcomingSchema));
checkState( }
Objects.equals(currentSchema.comment(), upcomingSchema.comment()),
if (!Objects.equals(currentSchema.comment(), upcomingSchema.comment())) {
throw new SchemaValidationException(
String.format( String.format(
"Unable to merge schema %s and %s with different comments.", "Unable to merge schema %s and %s with different comments.",
currentSchema, upcomingSchema)); currentSchema, upcomingSchema));
} }
}
/** /**
* Try to combine two columns with potential incompatible type. * Try to combine two columns with potential incompatible type.
@ -629,4 +639,11 @@ public class SchemaUtils {
throw new IllegalArgumentException( throw new IllegalArgumentException(
"Failed to get precision of non-exact decimal type " + dataType); "Failed to get precision of non-exact decimal type " + dataType);
} }
/** Thrown to indicate that schema validation has failed due to incompatible schema. */
public static class SchemaValidationException extends Exception {
public SchemaValidationException(String message) {
super(message);
}
}
} }

6
flink-cdc-common/src/test/java/org/apache/flink/cdc/common/utils/SchemaUtilsTest.java
Unescape Escape View File

@ -519,7 +519,8 @@ public class SchemaUtilsTest {
.physicalColumn("col2", DataTypes.STRING()) .physicalColumn("col2", DataTypes.STRING())
.primaryKey("col2") .primaryKey("col2")
.build()))) .build())))
.isExactlyInstanceOf(IllegalStateException.class) .rootCause()
.isExactlyInstanceOf(SchemaUtils.SchemaValidationException.class)
.as("test primary key conflict") .as("test primary key conflict")
.hasMessage( .hasMessage(
"Unable to merge schema columns={`col1` STRING,`col2` STRING}, primaryKeys=col1, options=() " "Unable to merge schema columns={`col1` STRING,`col2` STRING}, primaryKeys=col1, options=() "
@ -539,7 +540,8 @@ public class SchemaUtilsTest {
.physicalColumn("col2", DataTypes.STRING()) .physicalColumn("col2", DataTypes.STRING())
.partitionKey("col2") .partitionKey("col2")
.build()))) .build())))
.isExactlyInstanceOf(IllegalStateException.class) .rootCause()
.isExactlyInstanceOf(SchemaUtils.SchemaValidationException.class)
.as("test partition key conflict") .as("test partition key conflict")
.hasMessage( .hasMessage(
"Unable to merge schema columns={`col1` STRING,`col2` STRING}, primaryKeys=, partitionKeys=col1, options=() " "Unable to merge schema columns={`col1` STRING,`col2` STRING}, primaryKeys=, partitionKeys=col1, options=() "

9
flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineTransformITCase.java
Unescape Escape View File

@ -35,6 +35,7 @@ import org.apache.flink.cdc.common.schema.Column;
import org.apache.flink.cdc.common.schema.Schema; import org.apache.flink.cdc.common.schema.Schema;
import org.apache.flink.cdc.common.types.DataType; import org.apache.flink.cdc.common.types.DataType;
import org.apache.flink.cdc.common.types.DataTypes; import org.apache.flink.cdc.common.types.DataTypes;
import org.apache.flink.cdc.common.utils.SchemaUtils;
import org.apache.flink.cdc.composer.PipelineExecution; import org.apache.flink.cdc.composer.PipelineExecution;
import org.apache.flink.cdc.composer.definition.PipelineDef; import org.apache.flink.cdc.composer.definition.PipelineDef;
import org.apache.flink.cdc.composer.definition.SinkDef; import org.apache.flink.cdc.composer.definition.SinkDef;
@ -400,7 +401,7 @@ class FlinkPipelineTransformITCase {
null)), null)),
Collections.emptyList())) Collections.emptyList()))
.rootCause() .rootCause()
.isExactlyInstanceOf(IllegalStateException.class) .isExactlyInstanceOf(SchemaUtils.SchemaValidationException.class)
.hasMessage( .hasMessage(
"Unable to merge schema columns={`id` BIGINT NOT NULL,`name` VARCHAR(255),`age` TINYINT,`description` STRING}, primaryKeys=id, options=() " "Unable to merge schema columns={`id` BIGINT NOT NULL,`name` VARCHAR(255),`age` TINYINT,`description` STRING}, primaryKeys=id, options=() "
+ "and columns={`id` BIGINT NOT NULL,`name` STRING}, primaryKeys=id, options=() with different column counts."); + "and columns={`id` BIGINT NOT NULL,`name` STRING}, primaryKeys=id, options=() with different column counts.");
@ -408,7 +409,7 @@ class FlinkPipelineTransformITCase {
@ParameterizedTest @ParameterizedTest
@EnumSource @EnumSource
void testMultiTransformColumNameCompatibility(ValuesDataSink.SinkApi sinkApi) { void testMultiTransformColumnNameCompatibility(ValuesDataSink.SinkApi sinkApi) {
assertThatThrownBy( assertThatThrownBy(
() -> () ->
runGenericTransformTest( runGenericTransformTest(
@ -435,7 +436,7 @@ class FlinkPipelineTransformITCase {
null)), null)),
Collections.emptyList())) Collections.emptyList()))
.rootCause() .rootCause()
.isExactlyInstanceOf(IllegalStateException.class) .isExactlyInstanceOf(SchemaUtils.SchemaValidationException.class)
.hasMessage( .hasMessage(
"Unable to merge column `age` TINYINT and `name` STRING with different name."); "Unable to merge column `age` TINYINT and `name` STRING with different name.");
} }
@ -468,7 +469,7 @@ class FlinkPipelineTransformITCase {
null)), null)),
Collections.emptyList())) Collections.emptyList()))
.rootCause() .rootCause()
.isExactlyInstanceOf(IllegalStateException.class) .isExactlyInstanceOf(SchemaUtils.SchemaValidationException.class)
.hasMessage( .hasMessage(
"Unable to merge schema columns={`id` BIGINT,`name` VARCHAR(255),`age` TINYINT,`description` STRING}, primaryKeys=id, partitionKeys=age, options=() " "Unable to merge schema columns={`id` BIGINT,`name` VARCHAR(255),`age` TINYINT,`description` STRING}, primaryKeys=id, partitionKeys=age, options=() "
+ "and columns={`id` BIGINT,`name` VARCHAR(255),`age` TINYINT,`description` STRING}, primaryKeys=id, partitionKeys=id, options=() with different partition keys."); + "and columns={`id` BIGINT,`name` VARCHAR(255),`age` TINYINT,`description` STRING}, primaryKeys=id, partitionKeys=id, options=() with different partition keys.");

Write Preview
Loading…
Cancel
Save