public Collection<MySqlSnapshotSplit> generateSplits(TableId tableId) {
        try (JdbcConnection jdbc = openJdbcConnection(sourceConfig)) {

            LOG.info("Start splitting table {} into chunks...", tableId);
            long start = System.currentTimeMillis();

            Table table = mySqlSchema.getTableSchema(jdbc, tableId).getTable();

            List<ChunkRange> chunks = getChunks(tableId, jdbc, table);

            // convert chunks into splits
            List<MySqlSnapshotSplit> splits = new ArrayList<>();

            RowType splitType = ChunkUtils.getSplitType(getSplitColumn(table));
            for (int i = 0; i < chunks.size(); i++) {
                ChunkRange chunk = chunks.get(i);
                MySqlSnapshotSplit split =
                        createSnapshotSplit(
                                jdbc,
                                tableId,
                                i,
                                splitType,
                                chunk.getChunkStart(),
                                chunk.getChunkEnd());
                splits.add(split);
            }

            long end = System.currentTimeMillis();
            LOG.info(
                    "Split table {} into {} chunks, time cost: {}ms.",
                    tableId,
                    splits.size(),
                    end - start);
            return splits;
        } catch (Exception e) {
            throw new FlinkRuntimeException(
                    String.format("Generate Splits for table %s error", tableId), e);
        }
    }
Generates all snapshot splits (chunks) for the give table path.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/assigners/ChunkSplitter.java