alexey.rassokhin/tet123
1
0
Fork 0
Code Issues Pull Requests 1 Packages Projects Releases Wiki Activity

DBZ-5511 Keep consistence with method serialize of Connect/Json Table Change Serializer

Browse Source
This commit is contained in:
harveyyue 2022-08-13 22:32:54 +08:00 committed by Jiri Pechanec
parent 727e8819e9
commit a08420b21c
2 changed files with 12 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
debezium-core/src/main/java/io/debezium/relational/history/ConnectTableChangeSerializer.java
View File

@ -45,6 +45,8 @@ public class ConnectTableChangeSerializer implements TableChanges.TableChangesSe
public static final String AUTO_INCREMENTED_KEY = "autoIncremented";
public static final String GENERATED_KEY = "generated";
public static final String COMMENT_KEY = "comment";
public static final String DEFAULT_VALUE_EXPRESSION = "defaultValueExpression";
public static final String ENUM_VALUES = "enumValues";
private final Schema columnSchema;
private final Schema tableSchema;
@ -66,6 +68,8 @@ public ConnectTableChangeSerializer(SchemaNameAdjuster schemaNameAdjuster) {
.field(AUTO_INCREMENTED_KEY, Schema.OPTIONAL_BOOLEAN_SCHEMA)
.field(GENERATED_KEY, Schema.OPTIONAL_BOOLEAN_SCHEMA)
.field(COMMENT_KEY, Schema.OPTIONAL_STRING_SCHEMA)
.field(DEFAULT_VALUE_EXPRESSION, Schema.OPTIONAL_STRING_SCHEMA)
.field(ENUM_VALUES, SchemaBuilder.array(Schema.STRING_SCHEMA).optional().build())
.build();
tableSchema = SchemaBuilder.struct()
@ -145,6 +149,11 @@ private Struct toStruct(Column column) {
struct.put(GENERATED_KEY, column.isGenerated());
struct.put(COMMENT_KEY, column.comment());
column.defaultValueExpression().ifPresent(d -> struct.put(DEFAULT_VALUE_EXPRESSION, d));
if (column.enumValues() != null && !column.enumValues().isEmpty()) {
struct.put(ENUM_VALUES, column.enumValues());
}
return struct;
}

3
debezium-core/src/test/java/io/debezium/relational/history/HistoryRecordTest.java
View File

@ -127,5 +127,8 @@ public void canSerializeAndDeserializeHistoryRecord() throws Exception {
assertThat(tableStruct.get(ConnectTableChangeSerializer.COMMENT_KEY)).isEqualTo("table comment");
List<Struct> columnStructs = (List<Struct>) tableStruct.get(ConnectTableChangeSerializer.COLUMNS_KEY);
assertThat(columnStructs.get(0).get(ConnectTableChangeSerializer.COMMENT_KEY)).isEqualTo("first comment");
assertThat(columnStructs.get(0).get(ConnectTableChangeSerializer.ENUM_VALUES)).isNull();
assertThat(columnStructs.get(1).get(ConnectTableChangeSerializer.DEFAULT_VALUE_EXPRESSION)).isEqualTo("1");
assertThat(columnStructs.get(1).get(ConnectTableChangeSerializer.ENUM_VALUES)).isEqualTo(Collect.arrayListOf("1", "2"));
}
}