class FetchProjector {

    private static final long MIN_BYTES_PER_BUCKETS = ByteSizeUnit.KB.toBytes(64);

    // We only know the memory of the incoming rows up-front, so the percentage is low
    // to leave space for the cells being fetched.
    private static final double BREAKER_LIMIT_PERCENTAGE = 0.20d;

    public static long computeReaderBucketsByteThreshold(CircuitBreaker circuitBreaker) {
        return Math.max(
            (long) (circuitBreaker.getFree() * BREAKER_LIMIT_PERCENTAGE),
            MIN_BYTES_PER_BUCKETS
        );
    }

    public static Projector create(FetchProjection projection,
                                   RamAccounting ramAccounting,
                                   LongSupplier getBucketsBytesThreshold,
                                   TransactionContext txnCtx,
                                   NodeContext nodeCtx,
                                   FetchOperation fetchOperation) {<FILL_FUNCTION_BODY>}
}

        final FetchRows fetchRows = FetchRows.create(
            txnCtx,
            nodeCtx,
            projection.fetchSources(),
            projection.outputSymbols()
        );
        CellsSizeEstimator estimateRowSize = CellsSizeEstimator.forColumns(projection.inputTypes());
        return (BatchIterator<Row> source) -> {
            final long maxBucketsSizeInBytes = getBucketsBytesThreshold.getAsLong();
            BatchIterator<ReaderBuckets> buckets = BatchIterators.chunks(
                source,
                projection.getFetchSize(),
                () -> new ReaderBuckets(fetchRows, projection::getFetchSourceByReader, estimateRowSize, ramAccounting),
                ReaderBuckets::add,
                readerBuckets -> readerBuckets.ramBytesUsed() > maxBucketsSizeInBytes
            );
            return new AsyncFlatMapBatchIterator<>(
                buckets,
                new FetchMapper(fetchOperation, projection.nodeReaders())
            );
        };
    
<no_super_class>