diff --git a/core/lib/Drupal/Core/Database/Connection.php b/core/lib/Drupal/Core/Database/Connection.php
index a49fcab838..03ed814452 100644
--- a/core/lib/Drupal/Core/Database/Connection.php
+++ b/core/lib/Drupal/Core/Database/Connection.php
@@ -74,7 +74,7 @@ abstract class Connection {
    *
    * @var string
    */
-  protected $statementClass = 'Drupal\Core\Database\Statement';
+  protected $statementClass = 'Drupal\Core\Database\StatementWrapper';
 
   /**
    * Whether this database connection supports transactional DDL.
@@ -247,6 +247,16 @@ public function __construct(\PDO $connection, array $connection_options) {
     $this->connectionOptions = $connection_options;
   }
 
+  /**
+   * Returns the client-level database connection object.
+   *
+   * @return mixed
+   *   The client-level database connection, for example \PDO.
+   */
+  public function getClientConnection() {
+    return $this->connection;
+  }
+
   /**
    * Opens a PDO connection.
    *
diff --git a/core/lib/Drupal/Core/Database/Driver/mysql/Connection.php b/core/lib/Drupal/Core/Database/Driver/mysql/Connection.php
index 2894517fd4..4323d41839 100644
--- a/core/lib/Drupal/Core/Database/Driver/mysql/Connection.php
+++ b/core/lib/Drupal/Core/Database/Driver/mysql/Connection.php
@@ -10,6 +10,7 @@
 use Drupal\Core\Database\DatabaseNotFoundException;
 use Drupal\Core\Database\DatabaseException;
 use Drupal\Core\Database\Connection as DatabaseConnection;
+use Drupal\Core\Database\StatementInterface;
 use Drupal\Component\Utility\Unicode;
 
 /**
@@ -69,6 +70,68 @@ class Connection extends DatabaseConnection {
    */
   protected $identifierQuotes = ['"', '"'];
 
+  /**
+   * Constructs a MySql Connection object.
+   *
+   * @param \PDO $connection
+   *   An object of the PDO class representing a database connection.
+   * @param array $connection_options
+   *   An array of options for the connection. May include the following:
+   *   - prefix
+   *   - namespace
+   *   - Other driver-specific options.
+   */
+  public function __construct(\PDO $connection, array $connection_options) {
+    // The 'transactions' option is deprecated.
+    if (isset($connection_options['transactions'])) {
+      @trigger_error('Passing a \'transactions\' connection option to Drupal\Core\Database\Connection::__construct is deprecated in drupal:9.1.0 and is removed in drupal:10.0.0. All database drivers must support transactions. See https://www.drupal.org/node/2278745', E_USER_DEPRECATED);
+      unset($connection_options['transactions']);
+    }
+
+    // Work out the database driver namespace if none is provided. This normally
+    // written to setting.php by installer or set by
+    // \Drupal\Core\Database\Database::parseConnectionInfo().
+    if (empty($connection_options['namespace'])) {
+      $connection_options['namespace'] = __NAMESPACE__;
+    }
+
+    // Initialize and prepare the connection prefix.
+    $this->setPrefix(isset($connection_options['prefix']) ? $connection_options['prefix'] : '');
+
+    $this->connection = $connection;
+    $this->connectionOptions = $connection_options;
+  }
+
+  /**
+   * Destructs a MySql Connection object.
+   */
+  public function __destruct() {
+    if ($this->needsCleanup) {
+      $this->nextIdDelete();
+    }
+    $this->schema = NULL;
+    $this->connection = NULL;
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function prepareStatement(string $query, array $options): StatementInterface {
+    $query = $this->prefixTables($query);
+    if (!($options['allow_square_brackets'] ?? FALSE)) {
+      $query = $this->quoteIdentifiers($query);
+    }
+    return new $this->statementClass($this, $query, $options['pdo'] ?? []);
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function prepare($statement, array $driver_options = []) {
+    @trigger_error('Connection::prepare() is deprecated in drupal:9.1.0 and is removed from drupal:10.0.0. Database drivers should instantiate \PDOStatement objects by calling \PDO::prepare in their Collection::prepareStatement method instead. \PDO::prepare should not be called outside of driver code. See https://www.drupal.org/node/3137786', E_USER_DEPRECATED);
+    throw new DatabaseExceptionWrapper('Connection::prepare() is deprecated');
+  }
+
   /**
    * {@inheritdoc}
    */
@@ -218,16 +281,6 @@ public function serialize() {
     return parent::serialize();
   }
 
-  /**
-   * {@inheritdoc}
-   */
-  public function __destruct() {
-    if ($this->needsCleanup) {
-      $this->nextIdDelete();
-    }
-    parent::__destruct();
-  }
-
   public function queryRange($query, $from, $count, array $args = [], array $options = []) {
     return $this->query($query . ' LIMIT ' . (int) $from . ', ' . (int) $count, $args, $options);
   }
diff --git a/core/lib/Drupal/Core/Database/Driver/pgsql/Connection.php b/core/lib/Drupal/Core/Database/Driver/pgsql/Connection.php
index d0af842404..41c760ffb9 100644
--- a/core/lib/Drupal/Core/Database/Driver/pgsql/Connection.php
+++ b/core/lib/Drupal/Core/Database/Driver/pgsql/Connection.php
@@ -63,10 +63,35 @@ class Connection extends DatabaseConnection {
   protected $identifierQuotes = ['"', '"'];
 
   /**
-   * Constructs a connection object.
+   * Constructs a PostgreSQL Connection object.
+   *
+   * @param \PDO $connection
+   *   An object of the PDO class representing a database connection.
+   * @param array $connection_options
+   *   An array of options for the connection. May include the following:
+   *   - prefix
+   *   - namespace
+   *   - Other driver-specific options.
    */
   public function __construct(\PDO $connection, array $connection_options) {
-    parent::__construct($connection, $connection_options);
+    // The 'transactions' option is deprecated.
+    if (isset($connection_options['transactions'])) {
+      @trigger_error('Passing a \'transactions\' connection option to Drupal\Core\Database\Connection::__construct is deprecated in drupal:9.1.0 and is removed in drupal:10.0.0. All database drivers must support transactions. See https://www.drupal.org/node/2278745', E_USER_DEPRECATED);
+      unset($connection_options['transactions']);
+    }
+
+    // Work out the database driver namespace if none is provided. This normally
+    // written to setting.php by installer or set by
+    // \Drupal\Core\Database\Database::parseConnectionInfo().
+    if (empty($connection_options['namespace'])) {
+      $connection_options['namespace'] = __NAMESPACE__;
+    }
+
+    // Initialize and prepare the connection prefix.
+    $this->setPrefix(isset($connection_options['prefix']) ? $connection_options['prefix'] : '');
+
+    $this->connection = $connection;
+    $this->connectionOptions = $connection_options;
 
     // Force PostgreSQL to use the UTF-8 character set by default.
     $this->connection->exec("SET NAMES 'UTF8'");
@@ -77,6 +102,22 @@ public function __construct(\PDO $connection, array $connection_options) {
     }
   }
 
+  /**
+   * Destructs a PostgreSQL Connection object.
+   */
+  public function __destruct() {
+    $this->schema = NULL;
+    $this->connection = NULL;
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function prepare($statement, array $driver_options = []) {
+    @trigger_error('Connection::prepare() is deprecated in drupal:9.1.0 and is removed from drupal:10.0.0. Database drivers should instantiate \PDOStatement objects by calling \PDO::prepare in their Collection::prepareStatement method instead. \PDO::prepare should not be called outside of driver code. See https://www.drupal.org/node/3137786', E_USER_DEPRECATED);
+    throw new DatabaseExceptionWrapper('Connection::prepare() is deprecated');
+  }
+
   /**
    * {@inheritdoc}
    */
@@ -196,7 +237,11 @@ public function prepareStatement(string $query, array $options): StatementInterf
     // automatically cast the fields to the right type for these operators,
     // so we need to alter the query and add the type-cast.
     $query = preg_replace('/ ([^ ]+) +(I*LIKE|NOT +I*LIKE|~\*|!~\*) /i', ' ${1}::text ${2} ', $query);
-    return parent::prepareStatement($query, $options);
+    $query = $this->prefixTables($query);
+    if (!($options['allow_square_brackets'] ?? FALSE)) {
+      $query = $this->quoteIdentifiers($query);
+    }
+    return new $this->statementClass($this, $query, $options['pdo'] ?? []);
   }
 
   public function queryRange($query, $from, $count, array $args = [], array $options = []) {
diff --git a/core/lib/Drupal/Core/Database/Statement.php b/core/lib/Drupal/Core/Database/Statement.php
index c6a120d611..429b33b603 100644
--- a/core/lib/Drupal/Core/Database/Statement.php
+++ b/core/lib/Drupal/Core/Database/Statement.php
@@ -2,6 +2,8 @@
 
 namespace Drupal\Core\Database;
 
+@trigger_error(__CLASS__ . '() is deprecated in drupal:9.1.0 and is removed from drupal:10.0.0. Database drivers should use or extend StatementWrapper instead, and encapsulate client-level statement objects. See https://www.drupal.org/node/TODO', E_USER_DEPRECATED);
+
 /**
  * Default implementation of StatementInterface.
  *
@@ -12,6 +14,12 @@
  * constructor.
  *
  * @see http://php.net/pdostatement
+ *
+ * @deprecated in drupal:9.1.0 and is removed from drupal:10.0.0. Database
+ *   drivers should use or extend StatementWrapper instead, and encapsulate
+ *   client-level statement objects.
+ *
+ * @see https://www.drupal.org/node/TODO
  */
 class Statement extends \PDOStatement implements StatementInterface {
 
diff --git a/core/lib/Drupal/Core/Database/StatementWrapper.php b/core/lib/Drupal/Core/Database/StatementWrapper.php
new file mode 100644
index 0000000000..28fa2d576f
--- /dev/null
+++ b/core/lib/Drupal/Core/Database/StatementWrapper.php
@@ -0,0 +1,274 @@
+<?php
+
+namespace Drupal\Core\Database;
+
+/**
+ * Implementation of StatementInterface encapsulating PDOStatement.
+ */
+class StatementWrapper implements \IteratorAggregate, StatementInterface {
+
+  /**
+   * Reference to the database connection object for this statement.
+   *
+   * @var \Drupal\Core\Database\Connection
+   */
+  public $dbh;
+
+  /**
+  * The actual PDO Statement.
+   *
+   * @var \PDOStatement
+   */
+  protected $clientStatement;
+
+  /**
+   * Is rowCount() execution allowed.
+   *
+   * @var bool
+   */
+  public $allowRowCount = FALSE;
+
+  public function __construct(Connection $connection, string $query, array $options) {
+    $this->dbh = $connection;
+    $this->clientStatement = $connection->getClientConnection()->prepare($query, $options);
+    $this->setFetchMode(\PDO::FETCH_OBJ);
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function execute($args = [], $options = []) {
+    if (isset($options['fetch'])) {
+      if (is_string($options['fetch'])) {
+        // \PDO::FETCH_PROPS_LATE tells __construct() to run before properties
+        // are added to the object.
+        $this->setFetchMode(\PDO::FETCH_CLASS | \PDO::FETCH_PROPS_LATE, $options['fetch']);
+      }
+      else {
+        $this->setFetchMode($options['fetch']);
+      }
+    }
+
+    $logger = $this->dbh->getLogger();
+    if (!empty($logger)) {
+      $query_start = microtime(TRUE);
+    }
+
+    $return = $this->clientStatement->execute($args);
+
+    if (!empty($logger)) {
+      $query_end = microtime(TRUE);
+      $logger->log($this, $args, $query_end - $query_start);
+    }
+
+    return $return;
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function getQueryString() {
+    return $this->clientStatement->queryString;
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function fetchCol($index = 0) {
+    return $this->fetchAll(\PDO::FETCH_COLUMN, $index);
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function fetchAllAssoc($key, $fetch = NULL) {
+    $return = [];
+    if (isset($fetch)) {
+      if (is_string($fetch)) {
+        $this->setFetchMode(\PDO::FETCH_CLASS, $fetch);
+      }
+      else {
+        $this->setFetchMode($fetch);
+      }
+    }
+
+    foreach ($this as $record) {
+      $record_key = is_object($record) ? $record->$key : $record[$key];
+      $return[$record_key] = $record;
+    }
+
+    return $return;
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function fetchAllKeyed($key_index = 0, $value_index = 1) {
+    $return = [];
+    $this->setFetchMode(\PDO::FETCH_NUM);
+    foreach ($this as $record) {
+      $return[$record[$key_index]] = $record[$value_index];
+    }
+    return $return;
+  }
+
+  /**
+   * @todo
+   */
+  public  function fetchColumn(int $column_number = 0) {
+    return $this->clientStatement->fetchColumn($column_number);
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function fetchField($index = 0) {
+    // Call \PDOStatement::fetchColumn to fetch the field.
+    return $this->fetchColumn($index);
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function fetchAssoc() {
+    // Call \PDOStatement::fetch to fetch the row.
+    return $this->fetch(\PDO::FETCH_ASSOC);
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function fetchObject(string $class_name = NULL) {
+    if ($class_name) {
+      return $this->clientStatement->fetchObject($class_name);
+    }
+    return $this->clientStatement->fetchObject();
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function rowCount() {
+    // SELECT query should not use the method.
+    if ($this->allowRowCount) {
+      return $this->clientStatement->rowCount();
+    }
+    else {
+      throw new RowCountException();
+    }
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function setFetchMode($mode, $a1 = NULL, $a2 = []) {
+    // Call \PDOStatement::setFetchMode to set fetch mode.
+    // \PDOStatement is picky about the number of arguments in some cases so we
+    // need to be pass the exact number of arguments we where given.
+    switch (func_num_args()) {
+      case 1:
+        return $this->clientStatement->setFetchMode($mode);
+
+      case 2:
+        return $this->clientStatement->setFetchMode($mode, $a1);
+
+      case 3:
+      default:
+        return $this->clientStatement->setFetchMode($mode, $a1, $a2);
+    }
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function fetch($mode = NULL, $cursor_orientation = NULL, $cursor_offset = NULL) {
+    // Call \PDOStatement::fetchAll to fetch all rows.
+    // \PDOStatement is picky about the number of arguments in some cases so we
+    // need to be pass the exact number of arguments we where given.
+    switch (func_num_args()) {
+      case 0:
+        return $this->clientStatement->fetch();
+
+      case 1:
+        return $this->clientStatement->fetch($mode);
+
+      case 2:
+        return $this->clientStatement->fetch($mode, $column_index);
+
+      case 3:
+      default:
+        return $this->clientStatement->fetch($mode, $column_index, $constructor_arguments);
+    }
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function fetchAll($mode = NULL, $column_index = NULL, $constructor_arguments = NULL) {
+    // Call \PDOStatement::fetchAll to fetch all rows.
+    // \PDOStatement is picky about the number of arguments in some cases so we
+    // need to be pass the exact number of arguments we where given.
+    switch (func_num_args()) {
+      case 0:
+        return $this->clientStatement->fetchAll();
+
+      case 1:
+        return $this->clientStatement->fetchAll($mode);
+
+      case 2:
+        return $this->clientStatement->fetchAll($mode, $column_index);
+
+      case 3:
+      default:
+        return $this->clientStatement->fetchAll($mode, $column_index, $constructor_arguments);
+    }
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function getIterator() {
+    return new \ArrayIterator($this->fetchAll());
+  }
+
+  /**
+   * Binds a parameter to the specified variable name.
+   *
+   * @param mixed $parameter
+   *   Parameter identifier. For a prepared statement using named placeholders,
+   *   this will be a parameter name of the form :name.
+   * @param mixed $variable
+   *   Name of the PHP variable to bind to the SQL statement parameter.
+   * @param int $data_type
+   *   (Optional) explicit data type for the parameter using the PDO::PARAM_*
+   *   constants. To return an INOUT parameter from a stored procedure, use the
+   *   bitwise OR operator to set the PDO::PARAM_INPUT_OUTPUT bits for the
+   *   data_type parameter.
+   * @param int $length
+   *   (Optional) length of the data type. To indicate that a parameter is an
+   *   OUT parameter from a stored procedure, you must explicitly set the
+   *   length.
+   * @param mixed $driver_options
+   *   (Optional) Driver options.
+   *
+   * @return bool
+   *   Returns TRUE on success or FALSE on failure.
+   */
+  public function bindParam ($parameter, &$variable, int $data_type = \PDO::PARAM_STR, int $length = 0, $driver_options = NULL) : bool {
+    switch (func_num_args()) {
+      case 2:
+        return $this->clientStatement->bindParam($parameter, $variable);
+
+      case 3:
+        return $this->clientStatement->bindParam($parameter, $variable, $data_type);
+
+      case 4:
+        return $this->clientStatement->bindParam($parameter, $variable, $data_type, $length);
+
+      case 5:
+        return $this->clientStatement->bindParam($parameter, $variable, $data_type, $length, $driver_options);
+
+    }
+  }
+
+}