class Factory {

        private final List<OrderedDocCollector> orderedDocCollectors;
        private final Executor executor;
        private final IntSupplier availableThreads;
        private final PagingIterator<ShardId, Row> pagingIterator;
        private final Map<ShardId, OrderedDocCollector> collectorsByShardId;

        Factory(List<OrderedDocCollector> orderedDocCollectors,
                Comparator<Row> rowComparator,
                RowAccounting<Row> rowAccounting,
                Executor executor,
                IntSupplier availableThreads,
                boolean requiresScroll) {
            this.orderedDocCollectors = orderedDocCollectors;
            this.executor = executor;
            this.availableThreads = availableThreads;
            if (orderedDocCollectors.size() == 1) {
                pagingIterator = requiresScroll ?
                    new RamAccountingPageIterator<>(PassThroughPagingIterator.repeatable(), rowAccounting)
                    : PassThroughPagingIterator.oneShot();
                collectorsByShardId = null;
            } else {
                collectorsByShardId = toMapByShardId(orderedDocCollectors);
                pagingIterator = new RamAccountingPageIterator<>(
                    PagingIterator.createSorted(rowComparator, requiresScroll),
                    rowAccounting
                );
            }
        }

        BatchIterator<Row> create() {
            return new BatchPagingIterator<>(
                pagingIterator,
                this::tryFetchMore,
                this::allExhausted,
                throwable -> close()
            );
        }

        private KillableCompletionStage<List<KeyIterable<ShardId, Row>>> tryFetchMore(ShardId shardId) {
            if (allExhausted()) {
                return KillableCompletionStage.whenKilled(
                    CompletableFuture.failedFuture(new IllegalStateException("Cannot fetch more if source is exhausted")),
                    t -> { });
            }
            CompletionStage<List<KeyIterable<ShardId, Row>>> stage;
            if (shardId == null) {
                // when running inside threads, the threads must be cancelled/interrupted to stop further processing
                stage = ThreadPools.runWithAvailableThreads(
                        executor,
                        availableThreads,
                        Lists.map(orderedDocCollectors, Function.identity()));
            } else {
                stage = loadFrom(collectorsByShardId.get(shardId));
            }
            return KillableCompletionStage.whenKilled(stage, this::kill);
        }

        private static CompletionStage<List<KeyIterable<ShardId, Row>>> loadFrom(OrderedDocCollector collector) {
            try {
                return CompletableFuture.completedFuture(singletonList(collector.get()));
            } catch (Exception e) {
                return CompletableFuture.failedFuture(e);
            }
        }

        private void close() {
            for (OrderedDocCollector collector : orderedDocCollectors) {
                collector.close();
            }
        }

        private boolean allExhausted() {
            for (OrderedDocCollector collector : orderedDocCollectors) {
                if (!collector.exhausted) {
                    return false;
                }
            }
            return true;
        }

        private void kill(Throwable t) {
            for (OrderedDocCollector collector : orderedDocCollectors) {
                collector.kill(t);
            }
        }
    }


    private static Map<ShardId, OrderedDocCollector> toMapByShardId(List<OrderedDocCollector> collectors) {<FILL_FUNCTION_BODY>

        Map<ShardId, OrderedDocCollector> collectorsByShardId = new HashMap<>(collectors.size());
        for (OrderedDocCollector collector : collectors) {
            collectorsByShardId.put(collector.shardId(), collector);
        }
        return collectorsByShardId;
    
<no_super_class>