private MySqlBinlogSplit discoverTableSchemasForBinlogSplit(MySqlBinlogSplit split) {
        final String splitId = split.splitId();
        if (split.getTableSchemas().isEmpty() || sourceConfig.isScanNewlyAddedTableEnabled()) {
            try (MySqlConnection jdbc =
                    DebeziumUtils.createMySqlConnection(sourceConfig.getDbzConfiguration())) {
                Map<TableId, TableChanges.TableChange> tableSchemas;
                if (split.getTableSchemas().isEmpty()) {
                    tableSchemas =
                            TableDiscoveryUtils.discoverSchemaForCapturedTableSchemas(sourceConfig, jdbc);
                    LOG.info("The table schema discovery for binlog split {} success", splitId);
                } else {
                    List<TableId> existedTables = new ArrayList<>(split.getTableSchemas().keySet());
                    tableSchemas =
                            TableDiscoveryUtils.discoverSchemaForNewAddedTables(existedTables, sourceConfig, jdbc);
                    LOG.info(
                            "The table schema discovery for new added tables of binlog split {} success",
                            split.splitId());
                }
                return MySqlBinlogSplit.fillTableSchemas(split, tableSchemas);
            } catch (SQLException e) {
                LOG.error("Failed to obtains table schemas due to {}", e.getMessage());
                throw new FlinkRuntimeException(e);
            }
        } else {
            LOG.warn(
                    "The binlog split {} has table schemas yet, skip the table schema discovery",
                    split);
            return split;
        }
    }
The source reader for MySQL source splits.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/reader/MySqlSourceReader.java