Symfony/Doctrine Bulk insertion/upsert Manager

BulkManager.php

<?php

use Doctrine\DBAL\Connection;

class BulkManager
{
    public function __construct(private readonly Connection $connection) {}

    public function bulkInsert(string $tableName, array $records): int
    {
        if (empty($records)) {
            return 0;
        }

        $sql = $this->prepareInsertQuery($tableName, $records);
        $parameters = $this->flattenParameters($records);

        return $this->connection->executeStatement($sql, $parameters);
    }

    public function bulkUpsert(string $tableName, array $records, array $update = []): int
    {
        if (empty($records)) {
            return 0;
        }

        if (empty($update)) {
            return $this->bulkInsert($tableName, $records);
        }

        $insertSql = $this->prepareInsertQuery($tableName, $records);
        $updateClause = $this->prepareUpsertQuery($update);
        $parameters = array_merge(
            $this->flattenParameters($records),
            $this->getUpdateParameters($update)
        );

        return $this->connection->executeStatement("{$insertSql} {$updateClause}", $parameters);
    }

    private function prepareInsertQuery(string $tableName, array $records): string
    {
        if (! is_array(reset($records))) {
            $records = [$records];
        }

        $columns = array_map(function ($column) {
            return sprintf('`%s`', $column);
        }, array_keys(reset($records)));

        $placeholders = array_map(function ($record) {
            return sprintf('(%s)', implode(', ', array_fill(0, count($record), '?')));
        }, $records);

        return sprintf(
            'INSERT INTO `%s` (%s) VALUES %s',
            $tableName,
            implode(', ', $columns),
            implode(', ', $placeholders)
        );
    }

    private function prepareUpsertQuery(array $update): string
    {
        $columns = [];

        foreach ($update as $key => $column) {
            if (is_numeric($key)) {
                $columns[] = sprintf('`%s` = VALUES(`%s`)', $column, $column);
            } else {
                $columns[] = sprintf('`%s` = ?', $column);
            }
        }

        return sprintf('ON DUPLICATE KEY UPDATE %s', implode(', ', $columns));
    }

    private function flattenParameters(array $records): array
    {
        $parameters = [];

        foreach ($records as $record) {
            if (!is_array($record)) {
                $parameters[] = $record;
            } else {
                foreach (array_values($record) as $value) {
                    $parameters[] = $value;
                }
            }
        }

        return $parameters;
    }

    private function getUpdateParameters(array $update): array
    {
        $parameters = [];

        foreach ($update as $key => $value) {
            if (! is_numeric($key)) {
                $parameters[] = $value;
            }
        }

        return $parameters;
    }
}