presto-thrift/src/main/java/io/prestosql/plugin/thrift/ThriftSplitManager.java/ThriftSplitSource/getNextBatch
class ThriftSplitSource:        @Override
        public CompletableFuture<ConnectorSplitBatch> getNextBatch(ConnectorPartitionHandle partitionHandle, int maxSize)
        {
            checkState(future.get() == null || future.get().isDone(), "previous batch not completed");
            checkState(hasMoreData.get(), "this method cannot be invoked when there's no more data");
            PrestoThriftId currentToken = nextToken.get();
            ListenableFuture<PrestoThriftSplitBatch> splitsFuture = client.getSplits(
                    schemaTableName,
                    new PrestoThriftNullableColumnSet(columnNames.orElse(null)),
                    constraint,
                    maxSize,
                    new PrestoThriftNullableToken(currentToken));
            ListenableFuture<ConnectorSplitBatch> resultFuture = Futures.transform(
                    splitsFuture,
                    batch -> {
                        requireNonNull(batch, "batch is null");
                        List<ConnectorSplit> splits = batch.getSplits().stream()
                                .map(ThriftSplitSource::toConnectorSplit)
                                .collect(toImmutableList());
                        checkState(nextToken.compareAndSet(currentToken, batch.getNextToken()));
                        checkState(hasMoreData.compareAndSet(true, nextToken.get() != null));
                        return new ConnectorSplitBatch(splits, isFinished());
                    }, directExecutor());
            resultFuture = catchingThriftException(resultFuture);
            future.set(resultFuture);
            return toCompletableFuture(resultFuture);
        }