class UnboundedSortingLimitAndOffsetCollector implements Collector<Row, PriorityQueue<Object[]>, Bucket> {

    private final Collection<? extends Input<?>> inputs;
    private final Iterable<? extends CollectExpression<Row, ?>> expressions;
    private final int numOutputs;
    private final Comparator<Object[]> comparator;
    private final int initialCapacity;
    private final int offset;
    private final int maxNumberOfRowsInQueue;
    private final RowAccounting<Object[]> rowAccounting;

    /**
     * @param rowAccounting   sorting is a pipeline breaker so account for the used memory
     * @param inputs          contains output {@link Input}s and orderBy {@link Input}s
     * @param expressions     expressions linked to the inputs
     * @param numOutputs      number of output columns
     * @param comparator      used to sort the rows
     * @param initialCapacity the initial capacity of the backing queue
     * @param limit           the max number of rows the result should contain
     * @param offset          the number of rows to skip (after sort)
     */
    public UnboundedSortingLimitAndOffsetCollector(RowAccounting<Object[]> rowAccounting,
                                                   Collection<? extends Input<?>> inputs,
                                                   Iterable<? extends CollectExpression<Row, ?>> expressions,
                                                   int numOutputs,
                                                   Comparator<Object[]> comparator,
                                                   int initialCapacity,
                                                   int limit,
                                                   int offset) {
        if (initialCapacity <= 0) {
            throw new IllegalArgumentException("Invalid initial capacity: value must be > 0; got: " + initialCapacity);
        }
        if (limit <= 0) {
            throw new IllegalArgumentException("Invalid LIMIT: value must be > 0; got: " + limit);
        }
        if (offset < 0) {
            throw new IllegalArgumentException("Invalid OFFSET: value must be >= 0; got: " + offset);
        }
        this.rowAccounting = rowAccounting;
        this.inputs = inputs;
        this.expressions = expressions;
        this.numOutputs = numOutputs;
        this.comparator = comparator;
        this.initialCapacity = initialCapacity;
        this.offset = offset;
        this.maxNumberOfRowsInQueue = limit + offset;

        if (maxNumberOfRowsInQueue >= ArrayUtil.MAX_ARRAY_LENGTH || maxNumberOfRowsInQueue < 0) {
            // Throw exception to prevent confusing OOME in PriorityQueue
            // 1) if offset + limit exceeds maximum array length
            // 2) if offset + limit exceeds Integer.MAX_VALUE (then maxSize is negative!)
            throw new IllegalArgumentException(
                "Invalid LIMIT + OFFSET: value must be <= " + (ArrayUtil.MAX_ARRAY_LENGTH - 1) + "; got: " + maxNumberOfRowsInQueue);
        }
    }

    @Override
    public Supplier<PriorityQueue<Object[]>> supplier() {
        return () -> new PriorityQueue<>(initialCapacity, comparator.reversed());
    }

    @Override
    public BiConsumer<PriorityQueue<Object[]>, Row> accumulator() {
        return this::onNextRow;
    }

    @Override
    public BinaryOperator<PriorityQueue<Object[]>> combiner() {
        return (pq1, pq2) -> {
            throw new UnsupportedOperationException("combine not supported");
        };
    }

    @Override
    public Function<PriorityQueue<Object[]>, Bucket> finisher() {
        return this::pqToIterable;
    }

    @Override
    public Set<Characteristics> characteristics() {
        return Collections.emptySet();
    }

    private void onNextRow(PriorityQueue<Object[]> pq, Row row) {
        for (CollectExpression<Row, ?> expression : expressions) {
            expression.setNextRow(row);
        }
        Object[] rowCells = new Object[inputs.size()];
        int i = 0;
        for (Input<?> input : inputs) {
            rowCells[i] = input.value();
            i++;
        }
        rowAccounting.accountForAndMaybeBreak(rowCells);
        if (pq.size() == maxNumberOfRowsInQueue) {
            Object[] highestElementInOrder = pq.peek();
            if (highestElementInOrder == null || comparator.compare(rowCells, highestElementInOrder) < 0) {
                pq.poll();
                pq.add(rowCells);
            }
        } else {
            pq.add(rowCells);
        }
    }

    private Bucket pqToIterable(PriorityQueue<Object[]> pq) {<FILL_FUNCTION_BODY>}
}

        if (offset > pq.size()) {
            return new ArrayBucket(new Object[0][0], numOutputs);
        }
        int resultSize = Math.max(Math.min(maxNumberOfRowsInQueue - offset, pq.size() - offset), 0);

        Object[][] rows = new Object[resultSize][];
        for (int i = resultSize - 1; i >= 0; i--) {
            rows[i] = pq.poll();
        }
        return new ArrayBucket(rows, numOutputs);
    
<no_super_class>