[dbal] Optimize query so an index is used, not filesort.

Author: makasim

pkg/dbal/DbalConsumerHelperTrait.php

@@ -32,7 +32,7 @@ protected function fetchMessage(array $queues, int $redeliveryDelay): ?array
             ->andWhere('delivery_id IS NULL')
             ->andWhere('delayed_until IS NULL OR delayed_until <= :delayedUntil')
             ->andWhere('queue IN (:queues)')
-            ->addOrderBy('priority', 'desc')
+            ->addOrderBy('priority', 'asc')
             ->addOrderBy('published_at', 'asc')
             ->setParameter('delayedUntil', $now, ParameterType::INTEGER)
             ->setParameter('queues', array_values($queues), Connection::PARAM_STR_ARRAY)

pkg/dbal/DbalContext.php

@@ -154,7 +154,7 @@ public function convertMessage(array $dbalMessage): DbalMessage
             $dbalMessageObj->setRedelivered((bool) $dbalMessage['redelivered']);
         }
         if (isset($dbalMessage['priority'])) {
-            $dbalMessageObj->setPriority((int) $dbalMessage['priority']);
+            $dbalMessageObj->setPriority((int) (-1 * $dbalMessage['priority']));
         }
         if (isset($dbalMessage['published_at'])) {
             $dbalMessageObj->setPublishedAt((int) $dbalMessage['published_at']);
@@ -233,14 +233,18 @@ public function createDataBaseTable(): void
         $table->addColumn('redeliver_after', Type::BIGINT, ['notnull' => false]);
 
         $table->setPrimaryKey(['id']);
-        $table->addIndex(['published_at']);
-        $table->addIndex(['queue']);
-        $table->addIndex(['priority']);
-        $table->addIndex(['delayed_until']);
-        $table->addIndex(['priority', 'published_at']);
-        $table->addIndex(['redeliver_after']);
         $table->addUniqueIndex(['delivery_id']);
 
+        // try to select a message index
+        $table->addIndex(['delivery_id, delayed_until, queue']);
+        $table->addIndex(['priority', 'published_at']);
+
+        // redeliver failed messages
+        $table->addIndex(['delivery_id', 'redeliver_after']);
+
+        // remove expired messages
+        $table->addIndex(['time_to_live', 'delivery_id']);
+
         $sm->createTable($table);
     }
 }

pkg/dbal/DbalProducer.php

@@ -85,7 +85,7 @@ public function send(Destination $destination, Message $message): void
             'body' => $body,
             'headers' => JSON::encode($message->getHeaders()),
             'properties' => JSON::encode($message->getProperties()),
-            'priority' => $message->getPriority(),
+            'priority' => -1 * $message->getPriority(),
             'queue' => $destination->getQueueName(),
             'redelivered' => false,
             'delivery_id' => null,