dbal-bundle — DBAL-менеджер для highload-проектов на Symfony

Обзор

Dbal Bundle — Symfony-бандл для high-load систем, где стандартные возможности Doctrine ORM становятся узким местом. Бандл даёт абстракции и интерфейсы для прямой, эффективной и масштабируемой работы с базой на уровне Doctrine DBAL: bulk insert/update/upsert, cursor-стриминг, маппинг в DTO, advisory locks и верифицированную cross-vendor производительность на MySQL 8, MariaDB 10.5+ и PostgreSQL 12+.

Бандл не подменяет Doctrine ORM — он работает рядом: ORM остаётся на своей территории (Entity, миграции, CRUD), бандл закрывает инфраструктурные задачи (bulk, стриминг, read-modify-write с явным concurrency-контролем).

Архитектурное обоснование этого слоя — какие задачи ORM не тянет и где заканчивается его зона ответственности — см. Почему high-load Symfony нуждается в отдельном DBAL-слое рядом с ORM.

Быстрый старт

Минимальный сценарий: установить бандл, зарегистрировать и выполнить bulk insert через фабрику.

composer require elriseio/dbal-bundle

// config/bundles.php
return [
    Elrise\Bundle\DbalBundle\ElriseDbalBundle::class => ['all' => true],
];

class MyService
{
    public function __construct(private DbalManagerFactory $factory) {}

    public function insertUsers(array $emails): void
    {
        $inserter = $this->factory->createBulkInserter();
        $now = date('Y-m-d H:i:s');

        $inserter->insert('users', array_map(
            static fn (string $email) => ['email' => $email, 'created_at' => $now],
            $emails,
        ));
    }
}

Полная конфигурация, все bulk-операции, concurrency-хелперы и верифицированные таблицы производительности — ниже.

Что это и чем не является

Бандл делает:

• предоставляет BulkInserter, BulkUpdater, BulkUpserter, BulkDeleter с явным типом на каждое поле и нативным для СУБД upsert (ON DUPLICATE KEY UPDATE в MySQL/MariaDB, ON CONFLICT ... DO UPDATE в PostgreSQL);
• предоставляет DbalFinder и DbalMutator для типизированных чтений/записей с опциональным DTO-маппингом (без ORM hydration);
• предоставляет CursorIteratorInterface и OffsetIteratorInterface для потокового чтения больших результатов с ограниченной памятью;
• поддерживает несколько Doctrine Connection рядом (multi-database / шардированные потребители);
• экспонирует TransactionService с PostgreSQL advisory locks и row-level FOR UPDATE / FOR NO KEY UPDATE / FOR SHARE / FOR KEY SHARE (с MySQL 8+ семантикой);
• регистрирует семь Doctrine Types в build(), чтобы потребители могли объявлять колонки через #[Column(type: '...')] без ручной регистрации Type;
• эмитит PSR-3 события из каждой bulk-операции (start, end, constraint_violation, connection_error) при инжекте логгера;
• эмитит JSON caller-trace комментарий, добавляемый к каждому executeQuery / executeStatement / prepare, когда sql_comment_enabled включён, чтобы MySQL slow log и pg_stat_statements несли происхождение запроса;
• поставляет itests/ integration-слой с верифицированными цифрами на MySQL 8.4, MariaDB 11 и PostgreSQL 16 (см. §Верифицированная нагрузка).

Бандл не делает:

• не подменяет Doctrine ORM — работает рядом. Single-row чтения и транзакционные записи остаются на ORM, пока нет измеренной причины их переносить;
• не поставляет query DSL или query builder за пределами per-vendor SQL-хелперов в Sql\Builder\SqlBuilderInterface;
• не управляет схемой и миграциями — используйте Doctrine Migrations или любой уже знакомый инструмент;
• не поставляет ORM-style unit-of-work или identity map. Бандл работает построчно на уровне DBAL;
• не поддерживает связку Doctrine ORM 3.x + DBAL 3.x как runtime. Бандл рассчитан на DBAL 4.2+; использование DBAL 3.x упадёт на старте.

Совместимость

CI-гейт: scripts/check_interface_namespace_imports.sh отклоняет legacy Enum\*Interface импорты.

Архитектура

Бандл построен на интерфейсах и абстракциях, которые легко расширять и адаптировать под любые нужды.

В основе select-операций лежат генераторы (yield), что позволяет:

• обрабатывать большие объёмы данных с минимальным потреблением памяти;
• начинать обработку данных до завершения всего запроса (ленивая загрузка);
• реализовывать потоковую обработку и передачу данных — полезно при интеграции с очередями, API, логикой синхронизации и экспортами.

Основные интерфейсы

Finder / Mutator

• DbalFinderInterface: чтение данных, опционально с маппингом результата в DTO.
• DbalMutatorInterface: обновление, удаление, вставка, плюс чистый метод execute.

Bulk-операции

• BulkInserterInterface: вставка одной или нескольких строк.
• BulkUpdaterInterface: обновление одной или многих строк.
• BulkUpserterInterface: комбинированная операция обновления или вставки (upsert).
• BulkDeleterInterface: удаление, включая soft delete.

Итераторы

• CursorIteratorInterface: cursor-based чтение, подходит для стриминга.
• OffsetIteratorInterface: стандартная постраничная итерация.

Хелперы

• DtoFieldExtractor: извлекает и нормализует поля из DTO.
• DbalTypeGuesser: маппинг PHP-типов в SQL-типы.
• MysqlSqlBuilder: генератор SQL под MySQL.

Установка

composer require elriseio/dbal-bundle

Зарегистрируйте бандл в config/bundles.php:

return [
    Elrise\Bundle\DbalBundle\ElriseDbalBundle::class => ['all' => true],
];

Работа с DbalManagerFactory

Класс DbalManagerFactory позволяет удобно создавать компоненты DBAL-инфраструктуры с возможностью переопределения подключения (Connection) и конфигурации (DbalBundleConfig) на уровне каждого сервиса.

Быстрое создание DbalManager

Если хотите использовать все DBAL-компоненты сразу — достаточно createManager():

$dbalManager = $factory->createManager();

С кастомными Connection и DbalBundleConfig:

$dbalManager = $factory->createManager($customConnection, $customConfig);

Создание отдельных компонентов

Если нужен один из компонентов отдельно — используйте соответствующий метод:

$finder         = $factory->createFinder(...);
$mutator        = $factory->createMutator(...);
$cursorIterator = $factory->createCursorIterator(...);
$offsetIterator = $factory->createOffsetIterator(...);
$bulkInserter   = $factory->createBulkInserter(...);
$bulkUpdater    = $factory->createBulkUpdater(...);
$bulkUpserter   = $factory->createBulkUpserter(...);
$bulkDeleter    = $factory->createBulkDeleter(...);

Для каждого метода можно указать собственный Connection и (опционально) DbalBundleConfig. Особенно полезно при работе с несколькими БД или разными стратегиями конфигурации.

Пример в сервисе

class MyService
{
    public function __construct(private DbalManagerFactory $factory) {}

    public function updateBulkData(array $rows): void
    {
        $bulkUpdater = $this->factory->createBulkUpdater();
        $bulkUpdater->update('my_table', $rows);
    }
}

Bulk Insert

Модуль поддерживает массовую вставку с возможностью указать:

• имя таблицы;
• массив строк;
• автоматическую или ручную генерацию ID;
• явный тип на каждое поле.

Пример

/** @var BulkInserterInterface $inserter */
$inserter->insert('user_table', [
    [
        'id' => IdStrategy::AUTO_INCREMENT, // ID сгенерируется в БД
        'email' => ['user1@example.com', ParameterType::STRING],
        'created_at' => (new \DateTime())->format('Y-m-d H:i:s'),
    ],
    [
        'id' => IdStrategy::UUID, // ID сгенерируется в коде
        'email' => ['user2@example.com', ParameterType::STRING],
        'created_at' => (new \DateTime())->format('Y-m-d H:i:s'),
    ],
    [
        // ID сгенерируется в коде (UUIDv7 по умолчанию)
        'email' => ['user3@example.com', ParameterType::STRING],
        'created_at' => (new \DateTime())->format('Y-m-d H:i:s'),
    ],
]);

Массив ['value', ParameterType::TYPE] задаёт тип значения, совместимый с Doctrine\DBAL\ParameterType. Если тип не указан — он определяется автоматически.

Стратегии генерации ID (IdStrategy)

DbalBulkUpdater

DbalBulkUpdater обновляет от одной до многих строк в базе.

Пример

$bulkUpdater
    ->updateMany('api_history', [
        ['id' => 1, 'status' => 'success'],
        ['id' => 2, 'status' => 'success'],
    ]);

По умолчанию условие — поле id. Обновление идёт через CASE WHEN ... THEN ... одним запросом, без множественных round-trip’ов. Возвращается количество затронутых строк.

DbalBulkUpserter

DbalBulkUpserter вставляет или обновляет записи по ключевым полям. Если запись с id уже есть — обновляется; иначе — вставляется новая.

Пример

$bulkUpserter
    ->upsertMany('api_history', [
        [
            'id' => 123,
            'status' => 'success',
            'updated_at' => date('Y-m-d H:i:s'),
            'created_at' => date('Y-m-d H:i:s'),
        ],
        [
            'id' => IdStrategy::AUTO_INCREMENT,
            'status' => 'success',
            'updated_at' => date('Y-m-d H:i:s'),
            'created_at' => date('Y-m-d H:i:s'),
        ],
    ], ['status', 'updated_at']);

Обновляемые поля передаются третьим аргументом (replaceFields). id можно сгенерировать автоматически через IdStrategy::AUTO_INCREMENT.

PostgreSQL: RETURNING id за один round-trip

В PostgreSQL upsertManyReturningIds дописывает RETURNING <column> к upsert SQL и возвращает сгенерированные/обновлённые ID в порядке входных строк. MySQL / MariaDB RETURNING не поддерживают и бросят LogicException на этом вызове.

$ids = $bulkUpserter->upsertManyReturningIds(
    'api_history',
    [
        ['id' => IdStrategy::AUTO_INCREMENT, 'status' => 'success', 'updated_at' => date('Y-m-d H:i:s')],
        ['id' => IdStrategy::AUTO_INCREMENT, 'status' => 'pending', 'updated_at' => date('Y-m-d H:i:s')],
    ],
    ['status', 'updated_at'],
);
// $ids — это [123, 124] в порядке входных строк.

DbalFinder

DbalFinder даёт методы для типизированного чтения данных из базы.

Примеры

// Одна строка по SQL (LIMIT 1 добавляется автоматически).
$result = $finder->fetchOneBySql(
    'SELECT * FROM api_history WHERE id = :id',
    ['id' => $id],
    ApiDto::class
);

// Несколько строк с маппингом в DTO.
$results = $finder->fetchAllBySql(
    'SELECT * FROM api_history ORDER BY id LIMIT 10',
    [],
    ApiDto::class
);

// Запись по ID.
$result = $finder->findById($id, 'api_history', ApiDto::class);

// Записи по списку ID.
$result = $finder->findByIdList($idList, 'api_history', ApiDto::class);

Если DTO-класс не указан — вернётся массив.

DbalMutator

DbalMutator предназначен для безопасной вставки и изменения данных в таблицах.

Примеры

// Вставка одной строки.
$mutator->insert('api_history', [
    'type' => ['callback', ParameterType::STRING],
    'merchant_id' => '12345',
    'provider' => 'example-provider',
    'trace_id' => 'trace-001',
    'our_id' => 'our-001',
    'ext_id' => 'ext-001',
    'data' => json_encode(['source' => 'test']),
    'status' => 'success',
    'created_at' => date('Y-m-d H:i:s'),
    'updated_at' => date('Y-m-d H:i:s'),
]);

Поддерживаются поля с типами (например, ['value', ParameterType::STRING]). Если тип не указан — определяется автоматически.

⚠️ Важно

Перед использованием insert(), updateMany(), upsertMany() необходимо обязательно указать актуальные служебные поля через setFieldNames() или общий конфиг в fieldNames:

->setFieldNames([
    BundleConfigurationInterface::ID_NAME => 'id',
    BundleConfigurationInterface::CREATED_AT_NAME => 'created_at',
    BundleConfigurationInterface::UPDATED_AT_NAME => 'updated_at',
])

SQL caller-trace comment (opt-in)

Бандл экспонирует существующий DbalConnection::setAdditionalSqlCommentEnable toggle через конфигурацию doctrine_dbal бандла. Default: off (backward-compatible).

Когда включён, каждый executeQuery, executeStatement и prepare добавляет JSON caller-trace комментарий в начало SQL. Комментарий несёт applicationCaller (первый не-фреймворковый класс в backtrace) и entryPointController (Symfony-контроллер или имя консольной команды, если есть). Оператор видит происхождение каждого запроса в MySQL slow log, PostgreSQL pg_stat_statements или general log.

Включение per environment в config/packages/doctrine_dbal.yaml:

doctrine_dbal:
    sql_comment_enabled: true

Тот же флаг экспонирован на DbalBundleConfig::$sqlCommentEnabled для программного управления в тестах или кастомной wiring.

PSR-3 логирование bulk-операций (opt-in)

AbstractDbalWriteExecutor и его наследники (BulkInserter, BulkUpdater, BulkUpserter, BulkDeleter) принимают опциональный Psr\Log\LoggerInterface. Если логгер не инжектирован, executor использует NullLogger и ведёт себя как раньше. Когда логгер инжектирован (например, Symfony\Bridge\Monolog\Logger), каждая bulk-операция эмитит четыре event-типа:

{op} — один из insert, update, upsert, delete, soft_delete.

Wiring через services.yaml:

services:
    Elrise\Bundle\DbalBundle\Manager\Bulk\BulkInserter:
        parent: Elrise\Bundle\DbalBundle\Manager\Bulk\AbstractDbalWriteExecutor
        arguments:
            $logger: '@monolog.logger.dbal'

    monolog.logger.dbal:
        class: Symfony\Bridge\Monolog\Logger
        arguments: ['dbal']

composer.json уже заявляет psr/log: ^3.0; дополнительных зависимостей не требуется.

Concurrency-хелперы (advisory locks и row-level locks)

TransactionService экспонирует concurrency-хелперы для высокой пропускной способности.

PostgreSQL advisory locks

PostgreSQL advisory locks — application-level мьютексы, не привязанные к таблице. Полезны для leader election или single-writer секций.

// PG advisory lock — leader election или single-writer секция
$txService->transactional(function () use ($txService) {
    $txService->acquireAdvisoryLock(42); // блокирует до получения
    try {
        // ... критическая секция ...
    } finally {
        $txService->releaseAdvisoryLock(42);
    }
});

tryAdvisoryLock($key) — неблокирующий вариант; возвращает false, если другой держатель владеет локом. Все три advisory-метода — PostgreSQL-only и бросают LogicException на MySQL / MariaDB.

Row-level locks

SELECT ... WHERE ... FOR UPDATE лочит конкретные строки перед read-modify-write так, что другие транзакции не могут их менять до commit/rollback текущей.

// Row-level lock перед read-modify-write
$txService->transactional(function () use ($txService, $id) {
    $txService->lockRows('orders', ['id' => $id]); // FOR UPDATE
    $order = $finder->findById($id, 'orders');
    // ... modify and save ...
});

Enum LockMode (FOR_UPDATE, FOR_NO_KEY_UPDATE, FOR_SHARE, FOR_KEY_SHARE) покрывает оба row-lock варианта PostgreSQL и MySQL 8+ семантику. На MySQL legacy FOR_SHARE маппится в LOCK IN SHARE MODE. FOR_NO_KEY_UPDATE и FOR_KEY_SHARE — PostgreSQL-only и бросают LogicException на MySQL / MariaDB.

Зарегистрированные Doctrine Types

Бандл регистрирует семь Doctrine Types в build(), чтобы потребители могли объявлять колонки через #[Column(type: '...')] без ручной регистрации Type:

• float_array — FLOAT[] array literal (PostgreSQL).
• float4_array — FLOAT4[] array literal (PostgreSQL).
• int_array — INT[] array literal (PostgreSQL).
• text_array — TEXT[] array literal (PostgreSQL).
• numeric_array — NUMERIC[] array literal (PostgreSQL).
• jsonb — встроенный JsonbType Doctrine; эмитит JSONB на PostgreSQL 12+ и фоллбэчит на JSON в MySQL / MariaDB.
• jsonb_object — встроенный JsonbObjectType Doctrine; та же форма, что у jsonb, но нормализует object-формы при чтении.

Верифицированная нагрузка

Измерено itests/ integration load-testing пайплайном (ADR-0004) на реальных MySQL 8.4, PostgreSQL 16 и MariaDB 11 в Docker. Полный отчёт с per-vendor raw envelopes: itests/reports/WAVE_LT_CROSS_VENDOR_REPORT.md.

Single-vendor метрики (один прогон --rows 1000 --chunk 100 на сценарий):

Bounded-concurrency parallel runner (--workers 2 --concurrency 2, cursor_stream, --rows 500 на worker):

PG и MariaDB масштабируются линейно по воркерам. total_rows=400 в MySQL (вместо 1000) — известная коллизия unique-email между параллельными worker’ами; не correctness-regression.

Envelope contract (ADR-0004 Decision §5) соблюдён end-to-end на всех трёх вендорах; 306/306 PHPUnit проходит; php-cs-fixer показывает 0 violations. bulk_insert упирается в 4-7k rows/sec — зафиксированный gap до нативных COPY / LOAD DATA (ADR-0005, AR-036); не correctness-regression.

Тестирование

Бандл поставляет BulkTest console-command suite в BulkTestCommands/, который прогоняет каждую bulk-операцию против реальной базы. Фикстуры лежат в tests/_db/.

Wiring тестовых команд

Добавьте в services.yaml:

services:
    Elrise\Bundle\DbalBundle\Manager\Bulk\BulkUpserter:
        arguments:
            $connection: '@Doctrine\DBAL\Connection'
            $config: '@Elrise\Bundle\DbalBundle\Config\DbalBundleConfig'
            $sqlBuilder: '@Elrise\Bundle\DbalBundle\Sql\Builder\SqlBuilderInterface'

    Elrise\Bundle\DbalBundle\BulkTestCommands\BulkInsertManyCommand:
        arguments:
            $connection: '@Doctrine\DBAL\Connection'
            $bulkInserter: '@Elrise\Bundle\DbalBundle\Manager\Contract\BulkInserterInterface'
        tags: [ 'console.command' ]

    Elrise\Bundle\DbalBundle\BulkTestCommands\BulkUpdateManyCommand:
        arguments:
            $connection: '@Doctrine\DBAL\Connection'
            $bulkInserter: '@Elrise\Bundle\DbalBundle\Manager\Contract\BulkInserterInterface'
            $bulkUpdater: '@Elrise\Bundle\DbalBundle\Manager\Contract\BulkUpdaterInterface'
        tags: [ 'console.command' ]

    Elrise\Bundle\DbalBundle\BulkTestCommands\BulkUpsertManyCommand:
        arguments:
            $connection: '@Doctrine\DBAL\Connection'
            $bulkInserter: '@Elrise\Bundle\DbalBundle\Manager\Contract\BulkInserterInterface'
            $bulkUpserter: '@Elrise\Bundle\DbalBundle\Manager\Contract\BulkUpserterInterface'
        tags: [ 'console.command' ]

    Elrise\Bundle\DbalBundle\BulkTestCommands\BulkDeleteManyCommand:
        arguments:
            $connection: '@Doctrine\DBAL\Connection'
            $bulkDeleter: '@Elrise\Bundle\DbalBundle\Manager\Contract\BulkDeleterInterface'
            $bulkInserter: '@Elrise\Bundle\DbalBundle\Manager\Contract\BulkInserterInterface'
        tags: [ 'console.command' ]

    Elrise\Bundle\DbalBundle\BulkTestCommands\BulkSoftDeleteManyCommand:
        arguments:
            $connection: '@Doctrine\DBAL\Connection'
            $bulkDeleter: '@Elrise\Bundle\DbalBundle\Manager\Contract\BulkDeleterInterface'
            $bulkInserter: '@Elrise\Bundle\DbalBundle\Manager\Contract\BulkInserterInterface'
        tags: [ 'console.command' ]

Тестовая таблица

Прогоните SQL-фикстуры против вашей тестовой базы перед выполнением команд:

// для MySQL
tests/_db/init.sql

// для PostgreSQL
tests/_db/init_postgres.sql

Использование команд

bin/console dbal:test:run-all
bin/console dbal:test:bulk-insert-many
bin/console dbal:test:bulk-update-many
bin/console dbal:test:bulk-upsert-many
bin/console dbal:test:bulk-delete-many
bin/console dbal:test:bulk-soft-delete-many
bin/console dbal:test:cursor-iterator
bin/console dbal:test:offset-iterator
bin/console dbal:test:finder
bin/console dbal:test:mutator
bin/console dbal:test:transaction-service
bin/console dbal:test:insert

Каждая команда поддерживает:

• --chunk=<int> — размер чанка для пакетной обработки.
• --count=<int> — количество записей (по умолчанию 1000).
• --cycle=<int> — число повторов insert / update / delete (для бенчмарка).
• --track — включает логирование результатов.

Пример:

bin/console app:test:bulk-upsert-many --chunk=200 --count=5000 --cycle=5 --track

Логирование результатов

Когда передан флаг --track, команда сохраняет логи производительности в CSV-файл:

var/log/<тип_теста>_<timestamp>.csv

Каждая строка в логе содержит:

• номер итерации;
• время выполнения;
• использование памяти;
• изменение памяти;
• накопленное время.

Потрогать руками

dummy-market-agent — reference Symfony-сервис с Application Layer, CQRS, API Platform и DBAL-based persistence. Репозиторий запускается через Docker; в нём доступны API-операции, command/query handlers, persistence, container wiring, границы слоёв и тесты.

Известные пробелы в совместимости

• PostgreSQL 17 / MariaDB 12 / MySQL 9 пока не выделены в отдельные test-фикстуры. SQL-фрагменты бандла, как ожидается, останутся совместимыми, потому что таргетят стабильные core-фичи (ON DUPLICATE KEY UPDATE, ON CONFLICT ... DO UPDATE), но откройте issue, если встретите регрессию на новом сервере.

Источники

• github.com/elriseio/dbal-bundle — MIT.
• Packagist: elriseio/dbal-bundle
• Документация Doctrine DBAL
• Документация Symfony
• Архитектурное обоснование слоя — Почему high-load Symfony нуждается в отдельном DBAL-слое рядом с ORM