[Symfony - Doctrine] Iterate large amount of data without fetching all at once

CollectionStreamer.php

<?php

namespace App\Service\Doctrine;

use Doctrine\ORM\EntityManagerInterface;
use Generator;

class CollectionStreamer
{
    public static function stream(callable $fetch, int $chunkSize, callable $clear): Generator
    {
        $offset = 0;

        while (true) {
            $collection = $fetch($chunkSize, $offset);
            $itemClass = null;

            foreach ($collection as $item) {
                if ($itemClass === null) {
                    $itemClass = $item::class;
                }

                yield $item;
            }

            $clear($collection);

            if (count($collection) < $chunkSize) {
                break;
            }

            $offset += $chunkSize;
        }
    }
}

MyController.php

<?php

// example usage in a service

namespace App\Service;

use App\Repository\ProductRepository;

class MyService
{
    public function __construct(private ProductRepository $productRepository)
    {
    }
  
    public function exampleMethod(Category $category)
    {
        $products = $this->productRepository->streamByCategory($category);
      
        foreach ($products as $product) {
            dump($product);
        }
    }
}

ProductRepository.php

<?php

// example repository

namespace App\Repository;

use App\Entity\Product;
use App\Entity\Category;
use App\Service\Doctrine\CollectionStreamer;
use Doctrine\Persistence\ManagerRegistry;

class ProductRepository extends ServiceEntityRepository
{
    public function __construct(
        private ManagerRegistry $registry,
    ) {
        parent::__construct($registry, Product::class);
    }
  
    public function streamByCategory(Category $category, int $chunkSize = 100): Generator
    {
        $fetch = function (int $limit, int $offset) use ($cateogry) {
            return $this->findByCategory($category, $limit, $offset);
        };

        $clear = function () {
            $this->getEntityManager()->clear(Product::class);
        };

        return CollectionStreamer::stream($fetch, $chunkSize, $clear);
    }
  
    private function findByCategory(Category $category, int $limit, int $offset)
    {
        return $this->createQueryBuilder('p')
            ->andWhere('p.category = :category')
            ->setParameter('category', $category)
            ->setMaxResults($limit)
            ->setFirstResult($offset)
            ->getQuery()
            ->getResult();
    }
}

Note: this "streamer" will iterate through all the rows in the table, we can add another argument to prevent that happen :)