class SortingLimitAndOffsetCollectorBenchmark {

    private static final Comparator<Object[]> COMPARATOR = (o1, o2) -> Integer.compare((int) o2[0], (int) o1[0]);
    private static final RowCollectExpression INPUT = new RowCollectExpression(0);
    private static final List<Input<?>> INPUTS = List.of(INPUT);
    private static final Iterable<CollectExpression<Row, ?>> COLLECT_EXPRESSIONS = List.of(INPUT);

    private List<RowN> rows;
    private Collector<Row, ?, Bucket> boundedSortingCollector;
    private Collector<Row, ?, Bucket> unboundedSortingCollector;

    @Setup
    public void setUp() {
        rows = IntStream.range(0, 10_000_000)
            .mapToObj(RowN::new)
            .toList();
        boundedSortingCollector = new BoundedSortingLimitAndOffsetCollector(
            new RowAccounting<Object[]>() {

                @Override
                public long accountForAndMaybeBreak(Object[] row) {
                    return 42;
                }

                @Override
                public void release() {
                }
            },
            INPUTS,
            COLLECT_EXPRESSIONS,
            1,
            COMPARATOR,
            10_000,
            0);

        unboundedSortingCollector = new UnboundedSortingLimitAndOffsetCollector(
            new RowAccounting<Object[]>() {

                @Override
                public long accountForAndMaybeBreak(Object[] row) {
                    return 42;
                }

                @Override
                public void release() {
                }
            },
            INPUTS,
            COLLECT_EXPRESSIONS,
            1,
            COMPARATOR,
            10_000,
            10_000,
            0);
    }

    @Benchmark
    public Object measureBoundedSortingCollector() throws Exception {<FILL_FUNCTION_BODY>}

    @Benchmark
    public Object measureUnboundedSortingCollector() throws Exception {
        BatchIterator<Row> it = new InMemoryBatchIterator<>(rows, SENTINEL, false);
        BatchIterator<Row> sortingBatchIterator = CollectingBatchIterator.newInstance(it, unboundedSortingCollector);
        return sortingBatchIterator.loadNextBatch().toCompletableFuture().join();
    }
}

        BatchIterator<Row> it = new InMemoryBatchIterator<>(rows, SENTINEL, false);
        BatchIterator<Row> sortingBatchIterator = CollectingBatchIterator.newInstance(it, boundedSortingCollector);
        return sortingBatchIterator.loadNextBatch().toCompletableFuture().join();
    
<no_super_class>