Make PrimaryReadReplicaConnection enforcement explicit (#9239)

* Move primary replica connection logic into ORM explicitly.

* Housekeeping: Use full named variables

* Housekeeping: phpcs

Author: beberlei

lib/Doctrine/ORM/Id/SequenceGenerator.php

@@ -4,6 +4,7 @@
 
 namespace Doctrine\ORM\Id;
 
+use Doctrine\DBAL\Connections\PrimaryReadReplicaConnection;
 use Doctrine\ORM\EntityManager;
 use Serializable;
 
@@ -54,11 +55,14 @@ public function generate(EntityManager $em, $entity)
     {
         if ($this->_maxValue === null || $this->_nextValue === $this->_maxValue) {
             // Allocate new values
-            $conn = $em->getConnection();
-            $sql  = $conn->getDatabasePlatform()->getSequenceNextValSQL($this->_sequenceName);
+            $connection = $em->getConnection();
+            $sql        = $connection->getDatabasePlatform()->getSequenceNextValSQL($this->_sequenceName);
 
-            // Using `query` to force usage of the master server in MasterSlaveConnection
-            $this->_nextValue = (int) $conn->executeQuery($sql)->fetchOne();
+            if ($connection instanceof PrimaryReadReplicaConnection) {
+                $connection->ensureConnectedToPrimary();
+            }
+
+            $this->_nextValue = (int) $connection->executeQuery($sql)->fetchOne();
             $this->_maxValue  = $this->_nextValue + $this->_allocationSize;
         }
 

lib/Doctrine/ORM/Id/UuidGenerator.php

@@ -4,6 +4,7 @@
 
 namespace Doctrine\ORM\Id;
 
+use Doctrine\DBAL\Connections\PrimaryReadReplicaConnection;
 use Doctrine\DBAL\Platforms\AbstractPlatform;
 use Doctrine\Deprecations\Deprecation;
 use Doctrine\ORM\EntityManager;
@@ -39,9 +40,13 @@ public function __construct()
      */
     public function generate(EntityManager $em, $entity)
     {
-        $conn = $em->getConnection();
-        $sql  = 'SELECT ' . $conn->getDatabasePlatform()->getGuidExpression();
+        $connection = $em->getConnection();
+        $sql        = 'SELECT ' . $connection->getDatabasePlatform()->getGuidExpression();
 
-        return $conn->executeQuery($sql)->fetchOne();
+        if ($connection instanceof PrimaryReadReplicaConnection) {
+            $connection->ensureConnectedToPrimary();
+        }
+
+        return $connection->executeQuery($sql)->fetchOne();
     }
 }

lib/Doctrine/ORM/Query/Exec/MultiTableDeleteExecutor.php

@@ -5,6 +5,7 @@
 namespace Doctrine\ORM\Query\Exec;
 
 use Doctrine\DBAL\Connection;
+use Doctrine\DBAL\Connections\PrimaryReadReplicaConnection;
 use Doctrine\DBAL\Types\Type;
 use Doctrine\ORM\Query\AST;
 use Doctrine\ORM\Query\AST\DeleteStatement;
@@ -49,6 +50,10 @@ public function __construct(AST\Node $AST, $sqlWalker)
         $platform      = $conn->getDatabasePlatform();
         $quoteStrategy = $em->getConfiguration()->getQuoteStrategy();
 
+        if ($conn instanceof PrimaryReadReplicaConnection) {
+            $conn->ensureConnectedToPrimary();
+        }
+
         $primaryClass    = $em->getClassMetadata($AST->deleteClause->abstractSchemaName);
         $primaryDqlAlias = $AST->deleteClause->aliasIdentificationVariable;
         $rootClass       = $em->getClassMetadata($primaryClass->rootEntityName);

lib/Doctrine/ORM/Query/Exec/MultiTableUpdateExecutor.php

@@ -5,6 +5,7 @@
 namespace Doctrine\ORM\Query\Exec;
 
 use Doctrine\DBAL\Connection;
+use Doctrine\DBAL\Connections\PrimaryReadReplicaConnection;
 use Doctrine\DBAL\Types\Type;
 use Doctrine\ORM\Query\AST;
 use Doctrine\ORM\Query\AST\UpdateStatement;
@@ -55,6 +56,10 @@ public function __construct(AST\Node $AST, $sqlWalker)
         $platform      = $conn->getDatabasePlatform();
         $quoteStrategy = $em->getConfiguration()->getQuoteStrategy();
 
+        if ($conn instanceof PrimaryReadReplicaConnection) {
+            $conn->ensureConnectedToPrimary();
+        }
+
         $updateClause = $AST->updateClause;
         $primaryClass = $sqlWalker->getEntityManager()->getClassMetadata($updateClause->abstractSchemaName);
         $rootClass    = $em->getClassMetadata($primaryClass->rootEntityName);

lib/Doctrine/ORM/Query/Exec/SingleTableDeleteUpdateExecutor.php

@@ -5,6 +5,7 @@
 namespace Doctrine\ORM\Query\Exec;
 
 use Doctrine\DBAL\Connection;
+use Doctrine\DBAL\Connections\PrimaryReadReplicaConnection;
 use Doctrine\ORM\Query\AST;
 use Doctrine\ORM\Query\SqlWalker;
 
@@ -37,6 +38,10 @@ public function __construct(AST\Node $AST, $sqlWalker)
      */
     public function execute(Connection $conn, array $params, array $types)
     {
+        if ($conn instanceof PrimaryReadReplicaConnection) {
+            $conn->ensureConnectedToPrimary();
+        }
+
         return $conn->executeStatement($this->_sqlStatements, $params, $types);
     }
 }

lib/Doctrine/ORM/UnitOfWork.php

@@ -9,6 +9,7 @@
 use Doctrine\Common\Collections\Collection;
 use Doctrine\Common\EventManager;
 use Doctrine\Common\Proxy\Proxy;
+use Doctrine\DBAL\Connections\PrimaryReadReplicaConnection;
 use Doctrine\DBAL\LockMode;
 use Doctrine\Deprecations\Deprecation;
 use Doctrine\ORM\Cache\Persister\CachedPersister;
@@ -338,6 +339,12 @@ public function __construct(EntityManagerInterface $em)
      */
     public function commit($entity = null)
     {
+        $connection = $this->em->getConnection();
+
+        if ($connection instanceof PrimaryReadReplicaConnection) {
+            $connection->ensureConnectedToPrimary();
+        }
+
         // Raise preFlush
         if ($this->evm->hasListeners(Events::preFlush)) {
             $this->evm->dispatchEvent(Events::preFlush, new PreFlushEventArgs($this->em));