@Override
    public Collection<SnapshotSplit> split(SplitContext splitContext) {
        long chunkSizeInBytes = splitContext.getChunkSizeMB() * 1024 * 1024;

        long sizeInBytes = splitContext.getSizeInBytes();
        long count = splitContext.getDocumentCount();

        // If collection's total uncompressed size less than chunk size,
        // treat the entire collection as single chunk.
        if (sizeInBytes < chunkSizeInBytes) {
            return SingleSplitStrategy.INSTANCE.split(splitContext);
        }

        int numChunks = (int) (sizeInBytes / chunkSizeInBytes) + 1;
        int numberOfSamples;
        if (count < DEFAULT_SAMPLING_THRESHOLD) {
            // full sampling if document count less than sampling size threshold.
            numberOfSamples = (int) count;
        } else {
            // sampled using sample rate.
            numberOfSamples = (int) Math.floor(count * DEFAULT_SAMPLING_RATE);
        }

        TableId collectionId = splitContext.getCollectionId();

        MongoCollection<BsonDocument> collection =
                collectionFor(splitContext.getMongoClient(), collectionId, BsonDocument.class);

        List<Bson> pipeline = new ArrayList<>();
        if (numberOfSamples != count) {
            pipeline.add(sample(numberOfSamples));
        }
        pipeline.add(bucketAuto("$" + ID_FIELD, numChunks));

        List<BsonDocument> chunks =
                collection.aggregate(pipeline).allowDiskUse(true).into(new ArrayList<>());

        RowType rowType = shardKeysToRowType(Collections.singleton(ID_FIELD));

        List<SnapshotSplit> snapshotSplits = new ArrayList<>(chunks.size() + 2);

        Map<TableId, TableChanges.TableChange> schema = new HashMap<>();
        schema.put(collectionId, collectionSchema(collectionId));

        SnapshotSplit firstSplit =
                new SnapshotSplit(
                        collectionId,
                        splitId(collectionId, 0),
                        rowType,
                        minLowerBoundOfId(),
                        boundOfId(lowerBoundOfBucket(chunks.get(0))),
                        null,
                        schema);
        snapshotSplits.add(firstSplit);

        for (int i = 0; i < chunks.size(); i++) {
            BsonDocument bucket = chunks.get(i);
            snapshotSplits.add(
                    new SnapshotSplit(
                            collectionId,
                            splitId(collectionId, i + 1),
                            rowType,
                            boundOfId(lowerBoundOfBucket(bucket)),
                            boundOfId(upperBoundOfBucket(bucket)),
                            null,
                            schema));
        }

        SnapshotSplit lastSplit =
                new SnapshotSplit(
                        collectionId,
                        splitId(collectionId, chunks.size() + 1),
                        rowType,
                        boundOfId(upperBoundOfBucket(chunks.get(chunks.size() - 1))),
                        maxUpperBoundOfId(),
                        null,
                        schema);
        snapshotSplits.add(lastSplit);

        return snapshotSplits;
    }
The Sample Splitter.

<p>Uses the document size and 5% sampling rate of the collection to generate chunks for the
collection.

<pre>
If all of the following conditions are true, $sample uses a pseudo-random cursor to select the N documents:

     $sample is the first stage of the pipeline.
     N is less than 5% of the total documents in the collection.
     The collection contains more than 100 documents.
     If any of the previous conditions are false, $sample:

Reads all documents that are output from a preceding aggregation stage or a collection scan.
     Performs a random sort to select N documents.
</pre>
Copy from com.ververica:flink-connector-mongodb-cdc:2.3.0.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/source/assigners/splitters/SampleBucketSplitStrategy.java