bulk insert

Dockerfile

@@ -0,0 +1,23 @@
+FROM php:8.3
+
+ARG EXTENSIONS="pcntl zip intl bcmath"
+
+RUN apt-get update && apt-get install -y \
+    git \
+    zip \
+    unzip \
+    curl \
+ && rm -rf /var/lib/apt/lists/*
+
+COPY --from=composer /usr/bin/composer /usr/bin/composer
+
+ADD https://github.com/mlocati/docker-php-extension-installer/releases/latest/download/install-php-extensions /usr/local/bin/
+
+RUN chmod +x /usr/local/bin/install-php-extensions && \
+    install-php-extensions $EXTENSIONS
+
+RUN mkdir -p /tmp/blackfire \
+    && architecture=$(uname -m) \
+    && curl -A "Docker" -L https://blackfire.io/api/v1/releases/cli/linux/$architecture | tar zxp -C /tmp/blackfire \
+    && mv /tmp/blackfire/blackfire /usr/bin/blackfire \
+    && rm -Rf /tmp/blackfire

src/Store/DoctrineDbalStore.php

@@ -8,6 +8,7 @@
 use Doctrine\DBAL\Connection;
 use Doctrine\DBAL\Query\QueryBuilder;
 use Doctrine\DBAL\Schema\Schema;
+use Doctrine\DBAL\Types\Type;
 use Doctrine\DBAL\Types\Types;
 use Patchlevel\EventSourcing\Aggregate\AggregateRoot;
 use Patchlevel\EventSourcing\EventBus\HeaderNotFound;
@@ -16,6 +17,9 @@
 use Patchlevel\EventSourcing\Schema\SchemaConfigurator;
 use Patchlevel\EventSourcing\Serializer\EventSerializer;
 
+use function array_fill;
+use function count;
+use function implode;
 use function is_int;
 use function is_string;
 use function sprintf;
@@ -27,6 +31,7 @@ public function __construct(
         private readonly EventSerializer $serializer,
         private readonly AggregateRootRegistry $aggregateRootRegistry,
         private readonly string $storeTableName = 'eventstore',
+        private readonly int $batch = 1000,
     ) {
     }
 
@@ -114,38 +119,78 @@ public function save(Message ...$messages): void
     {
         $this->connection->transactional(
             function (Connection $connection) use ($messages): void {
+                $jsonType = Type::getType(Types::JSON);
+                $booleanType = Type::getType(Types::BOOLEAN);
+                $dateTimeType = Type::getType(Types::DATETIMETZ_IMMUTABLE);
+
+                $placeholders = [];
+
+                $columns = [
+                    'aggregate',
+                    'aggregate_id',
+                    'playhead',
+                    'event',
+                    'payload',
+                    'recorded_on',
+                    'new_stream_start',
+                    'archived',
+                    'custom_headers',
+                ];
+
+                $placeholder = implode(', ', array_fill(0, count($columns), '?'));
+
                 foreach ($messages as $message) {
                     $data = $this->serializer->serialize($message->event());
 
                     try {
-                        $connection->insert(
-                            $this->storeTableName,
-                            [
-                                'aggregate' => $this->aggregateRootRegistry->aggregateName($message->aggregateClass()),
-                                'aggregate_id' => $message->aggregateId(),
-                                'playhead' => $message->playhead(),
-                                'event' => $data->name,
-                                'payload' => $data->payload,
-                                'recorded_on' => $message->recordedOn(),
-                                'new_stream_start' => $message->newStreamStart(),
-                                'archived' => $message->archived(),
-                                'custom_headers' => $message->customHeaders(),
-                            ],
-                            [
-                                'recorded_on' => Types::DATETIMETZ_IMMUTABLE,
-                                'custom_headers' => Types::JSON,
-                                'new_stream_start' => Types::BOOLEAN,
-                                'archived' => Types::BOOLEAN,
-                            ],
-                        );
+                        $parameters[] = $this->aggregateRootRegistry->aggregateName($message->aggregateClass());
+                        $parameters[] = $message->aggregateId();
+                        $parameters[] = $message->playhead();
+                        $parameters[] = $data->name;
+                        $parameters[] = $data->payload;
+                        $parameters[] = $dateTimeType->convertToDatabaseValue($message->recordedOn(), $connection->getDatabasePlatform());
+                        $parameters[] = $booleanType->convertToDatabaseValue($message->newStreamStart(), $connection->getDatabasePlatform());
+                        $parameters[] = $booleanType->convertToDatabaseValue($message->archived(), $connection->getDatabasePlatform());
+                        $parameters[] = $jsonType->convertToDatabaseValue($message->customHeaders(), $connection->getDatabasePlatform());
                     } catch (HeaderNotFound $e) {
                         throw new MissingDataForStorage($e->name, $e);
                     }
+
+                    $placeholders[] = $placeholder;
                 }
+
+                if ($parameters === []) {
+                    return;
+                }
+
+                $query = sprintf(
+                    "INSERT INTO %s (%s) VALUES\n(%s)",
+                    $this->storeTableName,
+                    implode(', ', $columns),
+                    implode("),\n(", $placeholders),
+                );
+
+                $connection->executeStatement($query, $parameters);
             },
         );
     }
 
+    /**
+     * @param string[] $fields
+     *
+     * @return string[]
+     */
+    private function placeholder(array $fields): array
+    {
+        $placeholders = [];
+
+        foreach ($fields as $field) {
+            $placeholders[] = ':' . $field;
+        }
+
+        return $placeholders;
+    }
+
     /**
      * @param Closure():ClosureReturn $function
      *

tests/Benchmark/blackfire.php

@@ -0,0 +1,54 @@
+<?php
+
+declare(strict_types=1);
+
+use Doctrine\DBAL\Driver\PDO\SQLite\Driver;
+use Doctrine\DBAL\DriverManager;
+use Patchlevel\EventSourcing\EventBus\DefaultEventBus;
+use Patchlevel\EventSourcing\Metadata\AggregateRoot\AttributeAggregateRootRegistryFactory;
+use Patchlevel\EventSourcing\Repository\DefaultRepository;
+use Patchlevel\EventSourcing\Schema\DoctrineSchemaDirector;
+use Patchlevel\EventSourcing\Serializer\DefaultEventSerializer;
+use Patchlevel\EventSourcing\Store\DoctrineDbalStore;
+use Patchlevel\EventSourcing\Tests\Benchmark\BasicImplementation\Aggregate\Profile;
+use Patchlevel\EventSourcing\Tests\Benchmark\BasicImplementation\ProfileId;
+
+require_once __DIR__ . '/../../vendor/autoload.php';
+
+const DB_PATH = __DIR__ . '/BasicImplementation/data/db.sqlite3';
+
+if (file_exists(DB_PATH)) {
+    unlink(DB_PATH);
+}
+
+$connection = DriverManager::getConnection([
+    'driverClass' => Driver::class,
+    'path' => DB_PATH,
+]);
+
+$bus = new DefaultEventBus();
+
+$store = new DoctrineDbalStore(
+    $connection,
+    DefaultEventSerializer::createFromPaths([__DIR__ . '/BasicImplementation/Events']),
+    (new AttributeAggregateRootRegistryFactory())->create([__DIR__ . '/BasicImplementation/Aggregate']),
+    'eventstore',
+);
+
+$repository = new DefaultRepository($store, $bus, Profile::metadata());
+
+$schemaDirector = new DoctrineSchemaDirector(
+    $connection,
+    $store,
+);
+
+$schemaDirector->create();
+
+$id = ProfileId::generate();
+$profile = Profile::create($id, 'Peter');
+
+for ($i = 0; $i < 10_000; $i++) {
+    $profile->changeName('Peter ' . $i);
+}
+
+$repository->save($profile);