private void createDataEvents(
            RelationalSnapshotChangeEventSource.RelationalSnapshotContext snapshotContext,
            TableId tableId)
            throws Exception {
        EventDispatcher.SnapshotReceiver snapshotReceiver =
                dispatcher.getSnapshotChangeEventReceiver();
        LOG.debug("Snapshotting table {}", tableId);
        Table table = databaseSchema.tableFor(tableId);
        if (table != null) {
            createDataEventsForTable(snapshotContext, snapshotReceiver, table);
        } else {
            LOG.warn("Debezium doesn't capture {}, ignore this table", snapshotSplit.getTableId());
        }
        snapshotReceiver.completeSnapshot();
    }
for chunk that equals to the whole table we do not need to normalize
the snapshot data and the binlog data, just set high watermark to low watermark
@return highWatermark
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/debezium/task/MySqlSnapshotSplitReadTask.java