### Question:
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.isEmpty()) { 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); } @Inject  CassandraTokenSplitManager(CassandraSession session, CassandraClientConfig config);  CassandraTokenSplitManager(CassandraSession session, int splitSize, Optional<Long> configSplitsPerNode); List<TokenSplit> getSplits(String keyspace, String table, Optional<Long> sessionSplitsPerNode); long getTotalPartitionsCount(String keyspace, String table, Optional<Long> sessionSplitsPerNode);  }### Answer:
@Test public void testEmptyTable() throws Exception { String tableName = "empty_table"; session.execute(format("CREATE TABLE %s.%s (key text PRIMARY KEY)", KEYSPACE, tableName)); server.refreshSizeEstimates(KEYSPACE, tableName); List<TokenSplit> splits = splitManager.getSplits(KEYSPACE, tableName, Optional.empty()); assertThat(splits).hasSize(1); session.execute(format("DROP TABLE %s.%s", KEYSPACE, tableName)); }
@Test public void testNonEmptyTable() throws Exception { String tableName = "non_empty_table"; session.execute(format("CREATE TABLE %s.%s (key text PRIMARY KEY)", KEYSPACE, tableName)); for (int i = 0; i < PARTITION_COUNT; i++) { session.execute(format("INSERT INTO %s.%s (key) VALUES ('%s')", KEYSPACE, tableName, "value" + i)); } server.refreshSizeEstimates(KEYSPACE, tableName); List<TokenSplit> splits = splitManager.getSplits(KEYSPACE, tableName, Optional.empty()); int expectedTokenSplitSize = PARTITION_COUNT / SPLIT_SIZE; assertThat(splits).hasSizeBetween(expectedTokenSplitSize, expectedTokenSplitSize + 1); session.execute(format("DROP TABLE %s.%s", KEYSPACE, tableName)); }