github-actions[bot] commented on code in PR #68616:
URL: https://github.com/apache/doris/pull/68616#discussion_r4266330962
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/deserialize/MySqlDebeziumJsonDeserializer.java:
##########
@@ -125,45 +147,59 @@ private DeserializeResult handleSchemaChangeEvent(
parser.parse(ddl, tables);
changes = parser.getAndClearParsedChanges();
} catch (Exception e) {
- LOG.warn(
- "[SCHEMA-CHANGE-SKIPPED] MySQL schema change DDL parser
failed. No Doris DDL "
- + "emitted, FE baseline will advance to the schema
carried by the "
- + "history record. DDL: {}",
- ddl,
- e);
- return DeserializeResult.schemaChange(
- Collections.emptyList(), freshSchemas,
Collections.emptyList());
+ LOG.warn("Failed to parse MySQL schema change DDL: {}", ddl, e);
+ return unsupportedSchemaChange(record, ddl, "Cannot parse schema
change", freshSchemas);
}
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());
+ for (Map.Entry<TableId, TableChanges.TableChange> entry :
freshSchemas.entrySet()) {
+ if (entry.getValue().getType() !=
TableChanges.TableChangeType.ALTER) {
+ continue;
+ }
+ TableId tableId = entry.getKey();
+ Set<String> excluded =
+ excludeColumnsCache.getOrDefault(tableId.table(),
Collections.emptySet());
+ Set<String> before =
+ tableSchemas.get(tableId).getTable().columns().stream()
+ .map(Column::name)
+ .filter(name -> !excluded.contains(name))
+ .collect(Collectors.toSet());
+ Set<String> after =
+ entry.getValue().getTable().columns().stream()
+ .map(Column::name)
+ .filter(name -> !excluded.contains(name))
+ .collect(Collectors.toSet());
+ if (!before.equals(after)) {
+ return unsupportedSchemaChange(
+ record, ddl, "Unrecognized column change",
freshSchemas);
+ }
+ }
+ return DeserializeResult.schemaChange(Collections.emptyList(),
freshSchemas);
}
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);
+ MySqlSchemaChange.Type type = change.getType();
+ if (type == MySqlSchemaChange.Type.ADD || type ==
MySqlSchemaChange.Type.DROP) {
+ supportedChanges.add(change);
+ continue;
+ }
+
+ TableId tableId = change.getTableId();
+ String columnName = change.getColumnName();
+ String newColumnName = change.getNewColumnName();
+ Set<String> excluded =
+ excludeColumnsCache.getOrDefault(tableId.table(),
Collections.emptySet());
+ if (excluded.contains(columnName) &&
excluded.contains(newColumnName)) {
Review Comment:
[P2] Resolve MySQL column references before exclusion checks
With `table.t.exclude_columns=secret` and source `t(id INT PRIMARY KEY,
secret INT)`, the valid `ALTER TABLE t CHANGE COLUMN SECRET secret BIGINT`
keeps the same excluded column. MySQL resolves the old reference without case
sensitivity, while the listener preserves `SECRET`/`secret` from the DDL. Thus
`excluded.contains(columnName)` is false and this branch returns an unsupported
change; the pipeline pauses the job and disables automatic resume although no
target column is affected. Resolve the old reference against the baseline, or
compare it using MySQL's column-name rules, before checking exclusion
membership. Add this case to the deserializer tests. Source behavior: [MySQL
old-name
lookup](https://raw.githubusercontent.com/mysql/mysql-server/5.7/sql/sql_table.cc),
[Debezium name
parsing](https://raw.githubusercontent.com/debezium/debezium/v1.9.8.Final/debezium-connector-mysql/src/main/java/io/debezium/connector/mysql/antlr/MySqlAntlrDdlParser.java).
--
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]
