presto-cassandra/src/main/java/io/prestosql/plugin/cassandra/CassandraTokenSplitManager.java/CassandraTokenSplitManager/getSplits
class CassandraTokenSplitManager:    public List<TokenSplit> getSplits(String keyspace, String table, Optional<Long> sessionSplitsPerNode)
    {
        Set<TokenRange> tokenRanges = session.getTokenRanges();

        if (tokenRanges.isEmpty()) {
            throw new PrestoException(CASSANDRA_METADATA_ERROR, "The cluster metadata is not available. " +
                    "Please make sure that the Cassandra cluster is up and running, " +
                    "and that the contact points are specified correctly.");
        }

        if (tokenRanges.stream().anyMatch(TokenRange::isWrappedAround)) {
            tokenRanges = unwrap(tokenRanges);
        }

        Optional<TokenRing> tokenRing = createForPartitioner(session.getPartitioner());
        long totalPartitionsCount = getTotalPartitionsCount(keyspace, table, sessionSplitsPerNode);

        List<TokenSplit> splits = new ArrayList<>();
        for (TokenRange tokenRange : tokenRanges) {
            if (tokenRange.isEmpty()) {
                continue;
            }
            checkState(!tokenRange.isWrappedAround(), "all token ranges must be unwrapped at this step");

            List<String> endpoints = getEndpoints(keyspace, tokenRange);
            checkState(!endpoints.isEmpty(), "endpoints is empty for token range: %s", tokenRange);

            if (!tokenRing.isPresent()) {
                checkState(!tokenRange.isWrappedAround(), "all token ranges must be unwrapped at this step");
                splits.add(createSplit(tokenRange, endpoints));
                continue;
            }

            double tokenRangeRingFraction = tokenRing.get().getRingFraction(tokenRange.getStart().toString(), tokenRange.getEnd().toString());
            long partitionsCountEstimate = round(totalPartitionsCount * tokenRangeRingFraction);
            checkState(partitionsCountEstimate >= 0, "unexpected partitions count estimate: %s", partitionsCountEstimate);
            int subSplitCount = max(toIntExact(partitionsCountEstimate / splitSize), 1);
            List<TokenRange> subRanges = tokenRange.splitEvenly(subSplitCount);

            for (TokenRange subRange : subRanges) {
                if (subRange.isEmpty()) {
                    continue;
                }
                checkState(!subRange.isWrappedAround(), "all token ranges must be unwrapped at this step");
                splits.add(createSplit(subRange, endpoints));
            }
        }
        shuffle(splits, ThreadLocalRandom.current());
        return unmodifiableList(splits);
    }