github-actions[bot] commented on code in PR #65325:
URL: https://github.com/apache/doris/pull/65325#discussion_r3535556205
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/deserialize/MySqlDebeziumJsonDeserializer.java:
##########
@@ -59,8 +75,154 @@ 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()) {
+ LOG.warn(
+ "[SCHEMA-CHANGE-SKIPPED] MySQL schema change event
produced no supported "
+ + "column change. No Doris DDL emitted, FE
baseline will advance to "
+ + "the schema carried by the history record. DDL:
{}",
+ ddl);
+ return DeserializeResult.schemaChange(
+ Collections.emptyList(), freshSchemas,
Collections.emptyList());
+ }
+
+ List<MySqlSchemaChange> supportedChanges = new ArrayList<>();
+ for (MySqlSchemaChange change : changes) {
+ if (change.getType() == MySqlSchemaChange.Type.UNSUPPORTED) {
+ LOG.warn(
+ "[SCHEMA-CHANGE-SKIPPED] MySQL unsupported schema
change {} for table {}: {}."
+ + " Doris DDL will not be emitted for this
change, FE baseline will"
+ + " advance to the schema carried by the
history record. DDL: {}",
+ change.getSourceOperation(),
+ change.getTableId() == null
+ ? "<unknown>"
+ : change.getTableId().identifier(),
+ change.getReason(),
+ ddl);
+ continue;
+ }
+ supportedChanges.add(change);
+ }
+ if (supportedChanges.isEmpty()) {
+ return DeserializeResult.schemaChange(
+ Collections.emptyList(), freshSchemas,
Collections.emptyList());
+ }
+ for (MySqlSchemaChange change : supportedChanges) {
+ TableId tableId = change.getTableId();
+ if (tableId == null || !tableSchemas.containsKey(tableId)) {
+ LOG.warn(
+ "[SCHEMA-CHANGE-SKIPPED] MySQL schema change table {}
has no baseline."
+ + " No Doris DDL emitted, FE baseline will
advance to the schema"
+ + " carried by the history record. DDL: {}",
+ tableId == null ? "<unknown>" : tableId.identifier(),
+ ddl);
+ return DeserializeResult.schemaChange(
+ Collections.emptyList(), freshSchemas,
Collections.emptyList());
+ }
+ }
+
+ List<SchemaChangeOperation> schemaChanges = new ArrayList<>();
+ for (MySqlSchemaChange change : supportedChanges) {
+ TableId tableId = change.getTableId();
+ Set<String> excludedCols =
+ excludeColumnsCache.getOrDefault(tableId.table(),
Collections.emptySet());
+ String targetTable = resolveTargetTable(tableId.table());
+ String db = context.get(Constants.DORIS_TARGET_DB);
+ if (change.getType() == MySqlSchemaChange.Type.ADD) {
+ String columnName = change.getColumn().name();
+ if (excludedCols.contains(columnName)) {
+ LOG.info(
+ "[SCHEMA-CHANGE] MySQL table {}: added column '{}'
is excluded, skipping ADD",
+ tableId.identifier(),
+ columnName);
+ continue;
+ }
+ String colType =
SchemaChangeHelper.mysqlColumnToDorisType(change.getColumn());
+ schemaChanges.add(
+ SchemaChangeOperation.addColumn(
+ targetTable,
+ columnName,
+ SchemaChangeHelper.buildAddColumnSql(
+ db, targetTable, columnName, colType,
null, null)));
+ } else {
+ String columnName = change.getColumnName();
+ if (excludedCols.contains(columnName)) {
+ LOG.info(
+ "[SCHEMA-CHANGE] MySQL table {}: dropped column
'{}' is excluded, skipping DROP",
+ tableId.identifier(),
+ columnName);
+ continue;
+ }
+ schemaChanges.add(
+ SchemaChangeOperation.dropColumn(
+ targetTable,
+ columnName,
Review Comment:
`mvn -f fs_brokers/cdc_client/pom.xml -DskipTests
-Dspotless.check.skip=false validate` currently fails in `spotless:check`
before the module compiles. Spotless reports five changed Java files need
formatting, including this file around the `buildDropColumnSql` wrapping here
and the return below, plus `SchemaChangeHelper.java`,
`CustomMySqlAntlrDdlParser.java`, `CustomAlterTableParserListener.java`, and
`CustomMySqlAntlrDdlParserListener.java`. Please run the configured formatter
for this module and commit the result; otherwise the CDC client CI/style gate
will fail before tests run.
--
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]
