class CollectingBatchIterator<T> implements BatchIterator<T> {

    private final Supplier<CompletableFuture<? extends Iterable<? extends T>>> loadItems;
    private final Runnable onClose;
    private final Consumer<? super Throwable> onKill;
    private final boolean involvesIO;

    private CompletableFuture<? extends Iterable<? extends T>> resultFuture;
    private Iterator<? extends T> it = Collections.emptyIterator();
    private T current;
    private volatile Throwable killed;

    private CollectingBatchIterator(Supplier<CompletableFuture<? extends Iterable<? extends T>>> loadItems,
                                    Runnable onClose,
                                    Consumer<? super Throwable> onKill,
                                    boolean involvesIO) {
        this.loadItems = loadItems;
        this.onClose = onClose;
        this.onKill = onKill;
        this.involvesIO = involvesIO;
    }

    /**
     * Create a BatchIterator which will consume the source, summing up the first column (must be of type long).
     *
     * <pre>
     *     source BatchIterator:
     *     [ 1, 2, 2, 1 ]
     *
     *     output:
     *     [ 6 ]
     * </pre>
     */
    public static BatchIterator<Row> summingLong(BatchIterator<Row> source) {
        return newInstance(
            source,
            Collectors.collectingAndThen(
                Collectors.summingLong(r -> (long) r.get(0)),
                sum -> Collections.singletonList(new Row1(sum))));
    }

    public static <T, A> BatchIterator<T> newInstance(BatchIterator<T> source,
                                                      Collector<T, A, ? extends Iterable<? extends T>> collector) {<FILL_FUNCTION_BODY>}

    public static <I, O> BatchIterator<O> newInstance(BatchIterator<I> source,
                                                      Function<BatchIterator<I>, CompletableFuture<? extends Iterable<? extends O>>> processSource,
                                                      boolean involvesIO) {
        return newInstance(
            source::close,
            source::kill,
            () -> processSource.apply(source).whenComplete((r, f) -> source.close()),
            involvesIO);
    }

    public static <T> BatchIterator<T> newInstance(Runnable onClose,
                                                   Consumer<? super Throwable> onKill,
                                                   Supplier<CompletableFuture<? extends Iterable<? extends T>>> loadItems,
                                                   boolean involvesIO) {
        return new CollectingBatchIterator<>(loadItems, onClose, onKill, involvesIO);
    }

    @Override
    public T currentElement() {
        return current;
    }

    @Override
    public void moveToStart() {
        raiseIfKilled();
        if (resultFuture != null) {
            if (resultFuture.isDone() == false) {
                throw new IllegalStateException("BatchIterator is loading");
            }
            it = resultFuture.join().iterator();
        }
        current = null;
    }

    @Override
    public boolean moveNext() {
        raiseIfKilled();
        if (it.hasNext()) {
            current = it.next();
            return true;
        }
        current = null;
        return false;
    }

    @Override
    public void close() {
        onClose.run();
        killed = BatchIterator.CLOSED;
    }

    @Override
    public CompletionStage<?> loadNextBatch() throws Exception {
        if (resultFuture == null) {
            resultFuture = loadItems.get()
                .whenComplete((r, t) -> {
                    if (t == null) {
                        it = r.iterator();
                    } else if (t instanceof RuntimeException) {
                        throw (RuntimeException) t;
                    } else {
                        throw new RuntimeException(t);
                    }
                });
            return resultFuture;
        }
        throw new IllegalStateException("BatchIterator already loaded");
    }

    @Override
    public boolean allLoaded() {
        return resultFuture != null;
    }

    @Override
    public void kill(@NotNull Throwable throwable) {
        onKill.accept(throwable);
        killed = throwable;
    }

    @Override
    public boolean hasLazyResultSet() {
        return involvesIO;
    }

    private void raiseIfKilled() {
        if (killed != null) {
            Exceptions.rethrowUnchecked(killed);
        }
    }
}

        return newInstance(
            source::close,
            source::kill,
            () -> source.collect(collector),
            source.hasLazyResultSet()
        );
    
<no_super_class>