Copilot commented on code in PR #65325:
URL: https://github.com/apache/doris/pull/65325#discussion_r3541639758
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/deserialize/MySqlDebeziumJsonDeserializer.java:
##########
@@ -59,8 +75,158 @@ public DeserializeResult deserialize(Map<String, String>
context, SourceRecord r
}
private DeserializeResult handleSchemaChangeEvent(
- SourceRecord record, Map<String, String> context) {
- // todo: record has mysql ddl, need to convert doris ddl
- return DeserializeResult.empty();
+ SourceRecord record, Map<String, String> context) throws
IOException {
+ HistoryRecord history = RecordUtils.getHistoryRecord(record);
+ String database =
history.document().getString(HistoryRecord.Fields.DATABASE_NAME);
+ String ddl =
history.document().getString(HistoryRecord.Fields.DDL_STATEMENTS);
+ Map<TableId, TableChanges.TableChange> freshSchemas =
deserializeTableChanges(history);
+ if (freshSchemas.isEmpty()) {
+ LOG.warn(
+ "[SCHEMA-CHANGE-SKIPPED] MySQL schema change event has no
table schema "
+ + "change. DDL: {}",
+ ddl);
+ return DeserializeResult.empty();
+ }
+
+ if (tableSchemas == null || tableSchemas.isEmpty()) {
+ LOG.info(
+ "[SCHEMA-CHANGE] MySQL schema baseline is empty, adopting
fresh schemas as "
+ + "baseline without emitting Doris DDL. DDL: {}",
+ ddl);
+ return DeserializeResult.schemaChange(
+ Collections.emptyList(), freshSchemas,
Collections.emptyList());
+ }
+ for (TableId tableId : freshSchemas.keySet()) {
+ if (!tableSchemas.containsKey(tableId)) {
+ LOG.info(
+ "[SCHEMA-CHANGE] MySQL table {} has no baseline,
adopting fresh schemas as "
+ + "baseline without emitting Doris DDL. DDL:
{}",
+ tableId.identifier(),
+ ddl);
+ return DeserializeResult.schemaChange(
+ Collections.emptyList(), freshSchemas,
Collections.emptyList());
+ }
+ }
+
+ Tables tables = toTables(tableSchemas);
+ CustomMySqlAntlrDdlParser parser = new CustomMySqlAntlrDdlParser();
+ parser.setCurrentDatabase(database);
+ parser.parse(ddl, tables);
+ List<MySqlSchemaChange> changes = parser.getAndClearParsedChanges();
+ if (changes.isEmpty()) {
Review Comment:
`handleSchemaChangeEvent` calls the ANTLR DDL parser directly. If the parser
throws (e.g. for a DDL variant it can’t parse or a null/empty DDL), the whole
streaming job will fail. This should be treated like an unsupported schema
change: emit no Doris DDL but still advance the FE schema baseline using the
history record’s schemas.
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/utils/SchemaChangeHelper.java:
##########
@@ -136,6 +106,97 @@ public static String columnToDorisType(Column column) {
return pgTypeNameToDorisType(column.typeName(), column.length(),
column.scale().orElse(-1));
}
+ /** Convert a Debezium MySQL Column to a Doris column type string. */
+ public static String mysqlColumnToDorisType(Column column) {
+ Preconditions.checkNotNull(column.typeName());
+ String mysqlTypeName = column.typeName().toUpperCase();
+ String[] typeFields = mysqlTypeName.split(" ");
+ String mysqlType = typeFields[0];
+ boolean unsigned = typeFields.length > 1 &&
"UNSIGNED".equals(typeFields[1]);
+ int length = column.length();
Review Comment:
`mysqlColumnToDorisType` only treats the type as UNSIGNED when the *second*
token equals `UNSIGNED`. Debezium can emit additional modifiers (e.g.
`ZEROFILL`) or different token ordering, which would cause unsigned
integer/decimal types to be mapped as signed in Doris.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
