private void parseSchemaByDdl(
            String ddl, TableId tableId, Map<TableId, TableChange> tableChangeMap) {
        final MySqlOffsetContext offsetContext = MySqlOffsetContext.initial(connectorConfig);
        List<SchemaChangeEvent> schemaChangeEvents =
                databaseSchema.parseSnapshotDdl(
                        ddl, tableId.catalog(), offsetContext, Instant.now());
        for (SchemaChangeEvent schemaChangeEvent : schemaChangeEvents) {
            for (TableChange tableChange : schemaChangeEvent.getTableChanges()) {
                tableChangeMap.put(tableId, tableChange);
            }
        }
    }
Gets table schema for the given table path. It will request to MySQL server by running `SHOW
CREATE TABLE` if cache missed.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/schema/MySqlSchema.java