yuxiqian commented on code in PR #4064:
URL: https://github.com/apache/flink-cdc/pull/4064#discussion_r2508976334
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/utils/RecordUtils.java:
##########
@@ -590,8 +591,77 @@ public static boolean
isOnLineSchemaChangeEvent(SourceRecord record) {
}
}
+ public static Optional<String> parseOnLineSchemaRenameEvent(SourceRecord
record) {
+ if (!isSchemaChangeEvent(record)) {
+ return Optional.empty();
+ }
+ Struct value = (Struct) record.value();
+ ObjectMapper mapper = new ObjectMapper();
+
+ try {
+ String ddl =
+ mapper.readTree(value.getString(HISTORY_RECORD_FIELD))
+ .get(HistoryRecord.Fields.DDL_STATEMENTS)
+ .asText()
+ .toLowerCase();
+ if (ddl.startsWith("rename table") || ddl.startsWith("rename /*
gh-ost */ table")) {
+ LOG.info("Checking if DDL might be an OSC renaming event...
{}", ddl);
+ List<String> tableNames =
+ Arrays.asList(
+ value.getStruct(Envelope.FieldName.SOURCE)
+ .getString(TABLE_NAME_KEY)
+ .split(","));
+ if (tableNames.size() != 2) {
Review Comment:
After some quick investigation I noticed that if we use wildcard matching
(`.*`) to capture tables or `scan.binlog.newly-added-tables.enabled`, shadow
tables will not be included in the `tableNames` field and will be regarded as
two independent renaming events. Maybe we need to handle this case gracefully,
too?
--
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]
