public static void collectGhostDdl(SourceRecord element, MySqlSplitState splitState, HistoryRecord historyRecord,
            String ghostTableRegex) {
        String ddl = historyRecord.document().getString(Fields.DDL_STATEMENTS);
        if (StringUtils.isBlank(ddl)) {
            return;
        }
        String tableName = org.apache.inlong.sort.cdc.mysql.source.utils.RecordUtils.getTableName(element);
        Pattern compile = Pattern.compile(ghostTableRegex);
        Matcher matcher = compile.matcher(tableName);
        if (matcher.find()) {
            tableName = matcher.group(1);
            TableId tableId = getTableId(element, tableName);
            MySqlBinlogSplitState mySqlBinlogSplitState = splitState.asBinlogSplitState();
            if (ddl.toUpperCase().startsWith(DDL_OP_ALTER)
                    && mySqlBinlogSplitState.getTableSchemas().containsKey(tableId)) {
                String matchTableInSqlRegex = ghostTableRegex;
                if (matchTableInSqlRegex.startsWith(CARET) && matchTableInSqlRegex.endsWith(DOLLAR)) {
                    matchTableInSqlRegex = matchTableInSqlRegex.substring(1, matchTableInSqlRegex.length() - 1);
                }
                mySqlBinlogSplitState.recordTableDdl(
                        tableId,
                        ddl.replace(GHOST_TAG, "")
                                .replaceAll("\\s+", " ")
                                .replaceAll(matchTableInSqlRegex, tableName));
            }
        }
    }
after setting `gh-ost.ddl.change=true`, the alter ddl statements generated by gh-ost
will be captured and stored in the state.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/utils/GhostUtils.java