Add indexed queries to MySqlEventStore

src/MySqlEventStore.php

@@ -721,6 +721,7 @@ private function createWhereClause(?MetadataMatcher $metadataMatcher): array
         }
 
         foreach ($metadataMatcher->data() as $key => $match) {
+            $this->convertToColumn($match);
             /** @var FieldType $fieldType */
             $fieldType = $match['fieldType'];
             $field = $match['field'];
@@ -866,4 +867,20 @@ private function createSchemaFor(string $tableName): void
             }
         }
     }
+
+    /**
+     * Convert metadata fields into indexed columns
+     *
+     * @example `_aggregate__id` => `aggregate_id`
+     */
+    private function convertToColumn(array &$match): void
+    {
+        if ($this->persistenceStrategy instanceof MySqlIndexedPersistenceStrategy) {
+            $indexedColumns = $this->persistenceStrategy->indexedMetadataFields();
+            if (\in_array($match['field'], \array_keys($indexedColumns), true)) {
+                $match['field'] = $indexedColumns[$match['field']];
+                $match['fieldType'] = FieldType::MESSAGE_PROPERTY();
+            }
+        }
+    }
 }

src/MySqlIndexedPersistenceStrategy.php

@@ -0,0 +1,29 @@
+<?php
+
+/**
+ * This file is part of prooph/pdo-event-store.
+ * (c) 2016-2019 prooph software GmbH <[email protected]>
+ * (c) 2016-2019 Sascha-Oliver Prolic <[email protected]>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+declare(strict_types=1);
+
+namespace Prooph\EventStore\Pdo;
+
+interface MySqlIndexedPersistenceStrategy
+{
+    /**
+     * Return an array of indexed columns to enable the use of indexes in MariaDB
+     *
+     * @example
+     *      [
+     *          '_aggregate_id' => 'aggregate_id',
+     *          '_aggregate_type' => 'aggregate_type',
+     *          '_aggregate_version' => 'aggregate_version',
+     *      ]
+     */
+    public function indexedMetadataFields(): array;
+}

src/PersistenceStrategy/MySqlSingleStreamStrategy.php

@@ -17,10 +17,11 @@
 use Prooph\Common\Messaging\MessageConverter;
 use Prooph\EventStore\Pdo\DefaultMessageConverter;
 use Prooph\EventStore\Pdo\HasQueryHint;
+use Prooph\EventStore\Pdo\MySqlIndexedPersistenceStrategy;
 use Prooph\EventStore\Pdo\Util\Json;
 use Prooph\EventStore\StreamName;
 
-final class MySqlSingleStreamStrategy implements MySqlPersistenceStrategy, HasQueryHint
+final class MySqlSingleStreamStrategy implements MySqlPersistenceStrategy, HasQueryHint, MySqlIndexedPersistenceStrategy
 {
     /**
      * @var MessageConverter
@@ -70,6 +71,15 @@ public function columnNames(): array
         ];
     }
 
+    public function indexedMetadataFields(): array
+    {
+        return [
+            '_aggregate_id' => 'aggregate_id',
+            '_aggregate_type' => 'aggregate_type',
+            '_aggregate_version' => 'aggregate_version',
+        ];
+    }
+
     public function prepareData(Iterator $streamEvents): array
     {
         $data = [];

tests/Assets/PersistenceStrategy/CustomMySqlSingleStreamStrategy.php

@@ -15,11 +15,12 @@
 
 use Iterator;
 use Prooph\EventStore\Pdo\HasQueryHint;
+use Prooph\EventStore\Pdo\MySqlIndexedPersistenceStrategy;
 use Prooph\EventStore\Pdo\PersistenceStrategy\MySqlPersistenceStrategy;
 use Prooph\EventStore\Pdo\Util\Json;
 use Prooph\EventStore\StreamName;
 
-final class CustomMySqlSingleStreamStrategy implements MySqlPersistenceStrategy, HasQueryHint
+final class CustomMySqlSingleStreamStrategy implements MySqlPersistenceStrategy, HasQueryHint, MySqlIndexedPersistenceStrategy
 {
     /**
      * @param string $tableName
@@ -59,6 +60,15 @@ public function columnNames(): array
         ];
     }
 
+    public function indexedMetadataFields(): array
+    {
+        return [
+            '_aggregate_id' => 'aggregate_id',
+            '_aggregate_type' => 'aggregate_type',
+            '_aggregate_version' => 'aggregate_version',
+        ];
+    }
+
     public function prepareData(Iterator $streamEvents): array
     {
         $data = [];