gong commented on code in PR #7579:
URL: https://github.com/apache/inlong/pull/7579#discussion_r1152702512
##########
inlong-sort/sort-connectors/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/debezium/table/RowDataDebeziumDeserializeSchema.java:
##########
@@ -582,31 +588,49 @@ private DeserializationRuntimeConverter
getAllMigrationConverter() {
@Override
public Object convert(Object dbzObj, Schema schema) {
- ConnectSchema connectSchema = (ConnectSchema) schema;
- List<Field> fields = connectSchema.fields();
+ if (dbzObj instanceof Struct) {
+ ConnectSchema connectSchema = (ConnectSchema) schema;
+ List<Field> fields = connectSchema.fields();
- Map<String, Object> data = new HashMap<>();
- Struct struct = (Struct) dbzObj;
-
- for (Field field : fields) {
- String fieldName = field.name();
- Object fieldValue = struct.getWithoutDefault(fieldName);
- Schema fieldSchema = schema.field(fieldName).schema();
- String schemaName = fieldSchema.name();
- if (schemaName != null) {
- fieldValue = getValueWithSchema(fieldValue,
schemaName);
- }
- if (fieldValue instanceof ByteBuffer) {
- // binary data (blob or varbinary in mysql) are stored
in bytebuffer
- // use utf-8 to decode as a string by default
- fieldValue = new String(((ByteBuffer)
fieldValue).array());
+ Map<String, Object> data = new HashMap<>();
+ Struct struct = (Struct) dbzObj;
+
+ for (Field field : fields) {
+ String fieldName = field.name();
+ Object fieldValue =
struct.getWithoutDefault(fieldName);
+ Schema fieldSchema = schema.field(fieldName).schema();
+ String schemaName = fieldSchema.name();
+ if (schemaName != null) {
+ fieldValue = getValueWithSchema(fieldValue,
schemaName);
+ }
+ if (fieldValue instanceof ByteBuffer) {
+ // binary data (blob or varbinary in mysql) are
stored in bytebuffer
+ // use utf-8 to decode as a string by default
+ fieldValue = new String(((ByteBuffer)
fieldValue).array());
+ }
+ data.put(fieldName, fieldValue);
}
- data.put(fieldName, fieldValue);
+
+ GenericRowData row = new GenericRowData(1);
+ row.setField(0, data);
+ return row;
}
+ return constructDdlRow(dbzObj);
+
+ }
+
+ private GenericRowData constructDdlRow(Object ddl) {
+ Map<String, Object> data = new HashMap<>();
GenericRowData row = new GenericRowData(1);
row.setField(0, data);
-
+ try {
+ data.put(DDL_FIELD_NAME, ddl);
+ } catch (Exception e) {
+ LOG.info("Failed to convert DDL to json string", e);
+ throw new RuntimeException(e);
+ }
Review Comment:
`try ` can't get `Failed to convert DDL to json string`, just put data to
map not to convert data
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
