diff --git a/core/lib/Drupal/Core/Database/Connection.php b/core/lib/Drupal/Core/Database/Connection.php
index aad323229f06b03fec929c21aebb758bcf8e4e59..39a03b98ed4dbdb2fa24f36e80e8e0de17ed363f 100644
--- a/core/lib/Drupal/Core/Database/Connection.php
+++ b/core/lib/Drupal/Core/Database/Connection.php
@@ -73,9 +73,22 @@ abstract class Connection {
    * The name of the Statement class for this connection.
    *
    * @var string
+   *
+   * @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/3177488
    */
   protected $statementClass = 'Drupal\Core\Database\Statement';
 
+  /**
+   * The name of the StatementWrapper class for this connection.
+   *
+   * @var string
+   */
+  protected $statementWrapperClass = NULL;
+
   /**
    * Whether this database connection supports transactional DDL.
    *
@@ -239,7 +252,9 @@ public function __construct(\PDO $connection, array $connection_options) {
     $this->setPrefix(isset($connection_options['prefix']) ? $connection_options['prefix'] : '');
 
     // Set a Statement class, unless the driver opted out.
+    // @todo remove this in Drupal 10 https://www.drupal.org/node/3177490
     if (!empty($this->statementClass)) {
+      @trigger_error('\Drupal\Core\Database\Connection::$statementClass 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/3177488', E_USER_DEPRECATED);
       $connection->setAttribute(\PDO::ATTR_STATEMENT_CLASS, [$this->statementClass, [$this]]);
     }
 
@@ -278,6 +293,7 @@ public function destroy() {
       // Destroy all references to this connection by setting them to NULL.
       // The Statement class attribute only accepts a new value that presents a
       // proper callable, so we reset it to PDOStatement.
+      // @todo remove this in Drupal 10 https://www.drupal.org/node/3177490
       if (!empty($this->statementClass)) {
         $this->connection->setAttribute(\PDO::ATTR_STATEMENT_CLASS, ['PDOStatement', []]);
       }
@@ -535,7 +551,11 @@ public function prepareStatement(string $query, array $options): StatementInterf
     if (!($options['allow_square_brackets'] ?? FALSE)) {
       $query = $this->quoteIdentifiers($query);
     }
-    return $this->connection->prepare($query, $options['pdo'] ?? []);
+    // @todo in Drupal 10, only return the StatementWrapper.
+    // @see https://www.drupal.org/node/3177490
+    return $this->statementWrapperClass ?
+      new $this->statementWrapperClass($this, $this->connection, $query, $options['pdo'] ?? []) :
+      $this->connection->prepare($query, $options['pdo'] ?? []);
   }
 
   /**
@@ -728,7 +748,7 @@ protected function filterComment($comment = '') {
    * query. All queries executed by Drupal are executed as PDO prepared
    * statements.
    *
-   * @param string|\Drupal\Core\Database\StatementInterface $query
+   * @param string|\Drupal\Core\Database\StatementInterface|\PDOStatement $query
    *   The query to execute. In most cases this will be a string containing
    *   an SQL query with placeholders. An already-prepared instance of
    *   StatementInterface may also be passed in order to allow calling
@@ -779,6 +799,10 @@ public function query($query, array $args = [], $options = []) {
         $stmt = $query;
         $stmt->execute(NULL, $options);
       }
+      elseif ($query instanceof \PDOStatement) {
+        $stmt = $query;
+        $stmt->execute();
+      }
       else {
         $this->expandArguments($query, $args);
         // To protect against SQL injection, Drupal only supports executing one
@@ -854,7 +878,15 @@ protected function handleQueryException(\PDOException $e, $query, array $args =
       // Wrap the exception in another exception, because PHP does not allow
       // overriding Exception::getMessage(). Its message is the extra database
       // debug information.
-      $query_string = ($query instanceof StatementInterface) ? $query->getQueryString() : $query;
+      if ($query instanceof StatementInterface) {
+        $query_string = $query->getQueryString();
+      }
+      elseif ($query instanceof \PDOStatement) {
+        $query_string = $query->queryString;
+      }
+      else {
+        $query_string = $query;
+      }
       $message = $e->getMessage() . ": " . $query_string . "; " . print_r($args, TRUE);
       // Match all SQLSTATE 23xxx errors.
       if (substr($e->getCode(), -6, -3) == '23') {
@@ -1743,7 +1775,9 @@ abstract public function nextId($existing_id = 0);
    */
   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 Connection::prepareStatement method instead. \PDO::prepare should not be called outside of driver code. See https://www.drupal.org/node/3137786', E_USER_DEPRECATED);
-    return $this->connection->prepare($statement, $driver_options);
+    return $this->statementWrapperClass ?
+      (new $this->statementWrapperClass($this, $this->connection, $statement, $driver_options))->getClientStatement() :
+      $this->connection->prepare($statement, $driver_options);
   }
 
   /**
diff --git a/core/lib/Drupal/Core/Database/Driver/mysql/Connection.php b/core/lib/Drupal/Core/Database/Driver/mysql/Connection.php
index 2894517fd4a002b58ee399992e31c1c1397714b8..016c28382427e7577617fb40819fb40093f8132c 100644
--- a/core/lib/Drupal/Core/Database/Driver/mysql/Connection.php
+++ b/core/lib/Drupal/Core/Database/Driver/mysql/Connection.php
@@ -5,7 +5,7 @@
 use Drupal\Core\Database\DatabaseAccessDeniedException;
 use Drupal\Core\Database\IntegrityConstraintViolationException;
 use Drupal\Core\Database\DatabaseExceptionWrapper;
-
+use Drupal\Core\Database\StatementWrapper;
 use Drupal\Core\Database\Database;
 use Drupal\Core\Database\DatabaseNotFoundException;
 use Drupal\Core\Database\DatabaseException;
@@ -47,6 +47,16 @@ class Connection extends DatabaseConnection {
    */
   const SQLSTATE_SYNTAX_ERROR = 42000;
 
+  /**
+   * {@inheritdoc}
+   */
+  protected $statementClass = NULL;
+
+  /**
+   * {@inheritdoc}
+   */
+  protected $statementWrapperClass = StatementWrapper::class;
+
   /**
    * Flag to indicate if the cleanup function in __destruct() should run.
    *
diff --git a/core/lib/Drupal/Core/Database/Driver/pgsql/Connection.php b/core/lib/Drupal/Core/Database/Driver/pgsql/Connection.php
index d0af8424045472328c8ab887b69d901d3629ef05..4a07ba1a58b5a99a3a3f3729e756c040c0bc5a8c 100644
--- a/core/lib/Drupal/Core/Database/Driver/pgsql/Connection.php
+++ b/core/lib/Drupal/Core/Database/Driver/pgsql/Connection.php
@@ -7,6 +7,7 @@
 use Drupal\Core\Database\DatabaseAccessDeniedException;
 use Drupal\Core\Database\DatabaseNotFoundException;
 use Drupal\Core\Database\StatementInterface;
+use Drupal\Core\Database\StatementWrapper;
 
 // cSpell:ignore ilike nextval
 
@@ -38,6 +39,16 @@ class Connection extends DatabaseConnection {
    */
   const CONNECTION_FAILURE = '08006';
 
+  /**
+   * {@inheritdoc}
+   */
+  protected $statementClass = NULL;
+
+  /**
+   * {@inheritdoc}
+   */
+  protected $statementWrapperClass = StatementWrapper::class;
+
   /**
    * A map of condition operators to PostgreSQL operators.
    *
diff --git a/core/lib/Drupal/Core/Database/Driver/pgsql/Insert.php b/core/lib/Drupal/Core/Database/Driver/pgsql/Insert.php
index 5fc6a214c61721802149c15b1845c41332dd3ca4..bbab92babfac6455ebf2bdab57596da840a1ef2a 100644
--- a/core/lib/Drupal/Core/Database/Driver/pgsql/Insert.php
+++ b/core/lib/Drupal/Core/Database/Driver/pgsql/Insert.php
@@ -37,13 +37,13 @@ public function execute() {
           fwrite($blobs[$blob_count], $insert_values[$idx]);
           rewind($blobs[$blob_count]);
 
-          $stmt->bindParam(':db_insert_placeholder_' . $max_placeholder++, $blobs[$blob_count], \PDO::PARAM_LOB);
+          $stmt->getClientStatement()->bindParam(':db_insert_placeholder_' . $max_placeholder++, $blobs[$blob_count], \PDO::PARAM_LOB);
 
           // Pre-increment is faster in PHP than increment.
           ++$blob_count;
         }
         else {
-          $stmt->bindParam(':db_insert_placeholder_' . $max_placeholder++, $insert_values[$idx]);
+          $stmt->getClientStatement()->bindParam(':db_insert_placeholder_' . $max_placeholder++, $insert_values[$idx]);
         }
       }
       // Check if values for a serial field has been passed.
@@ -80,7 +80,7 @@ public function execute() {
       // the foreach statement assigns the element to the existing reference.
       $arguments = $this->fromQuery->getArguments();
       foreach ($arguments as $key => $value) {
-        $stmt->bindParam($key, $arguments[$key]);
+        $stmt->getClientStatement()->bindParam($key, $arguments[$key]);
       }
     }
 
diff --git a/core/lib/Drupal/Core/Database/Driver/pgsql/Update.php b/core/lib/Drupal/Core/Database/Driver/pgsql/Update.php
index c4297d1e8cecdceaab49692f179f0e89f2603f18..7e49310175a34b4ad6f7bf4edc07b5fc5ca9c5a4 100644
--- a/core/lib/Drupal/Core/Database/Driver/pgsql/Update.php
+++ b/core/lib/Drupal/Core/Database/Driver/pgsql/Update.php
@@ -32,14 +32,14 @@ public function execute() {
           // We assume that an expression will never happen on a BLOB field,
           // which is a fairly safe assumption to make since in most cases
           // it would be an invalid query anyway.
-          $stmt->bindParam($placeholder, $data['arguments'][$placeholder]);
+          $stmt->getClientStatement()->bindParam($placeholder, $data['arguments'][$placeholder]);
         }
       }
       if ($data['expression'] instanceof SelectInterface) {
         $data['expression']->compile($this->connection, $this);
         $select_query_arguments = $data['expression']->arguments();
         foreach ($select_query_arguments as $placeholder => $argument) {
-          $stmt->bindParam($placeholder, $select_query_arguments[$placeholder]);
+          $stmt->getClientStatement()->bindParam($placeholder, $select_query_arguments[$placeholder]);
         }
       }
       unset($fields[$field]);
@@ -52,11 +52,11 @@ public function execute() {
         $blobs[$blob_count] = fopen('php://memory', 'a');
         fwrite($blobs[$blob_count], $value);
         rewind($blobs[$blob_count]);
-        $stmt->bindParam($placeholder, $blobs[$blob_count], \PDO::PARAM_LOB);
+        $stmt->getClientStatement()->bindParam($placeholder, $blobs[$blob_count], \PDO::PARAM_LOB);
         ++$blob_count;
       }
       else {
-        $stmt->bindParam($placeholder, $fields[$field]);
+        $stmt->getClientStatement()->bindParam($placeholder, $fields[$field]);
       }
     }
 
@@ -65,7 +65,7 @@ public function execute() {
 
       $arguments = $this->condition->arguments();
       foreach ($arguments as $placeholder => $value) {
-        $stmt->bindParam($placeholder, $arguments[$placeholder]);
+        $stmt->getClientStatement()->bindParam($placeholder, $arguments[$placeholder]);
       }
     }
 
diff --git a/core/lib/Drupal/Core/Database/Driver/pgsql/Upsert.php b/core/lib/Drupal/Core/Database/Driver/pgsql/Upsert.php
index df63e137847e2ef446dd72848a05bff8cf3348a1..4aa51e588589c11f5eec9ed410df7cbcdddc63ed 100644
--- a/core/lib/Drupal/Core/Database/Driver/pgsql/Upsert.php
+++ b/core/lib/Drupal/Core/Database/Driver/pgsql/Upsert.php
@@ -34,13 +34,13 @@ public function execute() {
           fwrite($blobs[$blob_count], $insert_values[$idx]);
           rewind($blobs[$blob_count]);
 
-          $stmt->bindParam(':db_insert_placeholder_' . $max_placeholder++, $blobs[$blob_count], \PDO::PARAM_LOB);
+          $stmt->getClientStatement()->bindParam(':db_insert_placeholder_' . $max_placeholder++, $blobs[$blob_count], \PDO::PARAM_LOB);
 
           // Pre-increment is faster in PHP than increment.
           ++$blob_count;
         }
         else {
-          $stmt->bindParam(':db_insert_placeholder_' . $max_placeholder++, $insert_values[$idx]);
+          $stmt->getClientStatement()->bindParam(':db_insert_placeholder_' . $max_placeholder++, $insert_values[$idx]);
         }
       }
       // Check if values for a serial field has been passed.
diff --git a/core/lib/Drupal/Core/Database/Driver/sqlite/Connection.php b/core/lib/Drupal/Core/Database/Driver/sqlite/Connection.php
index d59e2d07bc06137ff549cabf0446645e919db0d7..0905ca9e2ae5ae63aa0c6d9cffa6cff19aadf19a 100644
--- a/core/lib/Drupal/Core/Database/Driver/sqlite/Connection.php
+++ b/core/lib/Drupal/Core/Database/Driver/sqlite/Connection.php
@@ -17,6 +17,16 @@ class Connection extends DatabaseConnection {
    */
   const DATABASE_NOT_FOUND = 14;
 
+  /**
+   * {@inheritdoc}
+   */
+  protected $statementClass = NULL;
+
+  /**
+   * {@inheritdoc}
+   */
+  protected $statementWrapperClass = NULL;
+
   /**
    * Whether or not the active transaction (if any) will be rolled back.
    *
@@ -70,10 +80,6 @@ class Connection extends DatabaseConnection {
    * Constructs a \Drupal\Core\Database\Driver\sqlite\Connection object.
    */
   public function __construct(\PDO $connection, array $connection_options) {
-    // We don't need a specific PDOStatement class here, we simulate it in
-    // static::prepare().
-    $this->statementClass = NULL;
-
     parent::__construct($connection, $connection_options);
 
     // Attach one database for each registered prefix.
diff --git a/core/lib/Drupal/Core/Database/Statement.php b/core/lib/Drupal/Core/Database/Statement.php
index c6a120d61166b09f271335f6f86d7346955e4baa..3de4537fb2cdcfff46a81fa96f3b18b7e339d763 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('\Drupal\Core\Database\Statement 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/3177488', 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/3177488
  */
 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 0000000000000000000000000000000000000000..eff40102f46b9773e86a49080309f567b9825af8
--- /dev/null
+++ b/core/lib/Drupal/Core/Database/StatementWrapper.php
@@ -0,0 +1,377 @@
+<?php
+
+namespace Drupal\Core\Database;
+
+// cSpell:ignore maxlen driverdata INOUT
+
+/**
+ * Implementation of StatementInterface encapsulating PDOStatement.
+ */
+class StatementWrapper implements \IteratorAggregate, StatementInterface {
+
+  /**
+   * The Drupal database connection object.
+   *
+   * @var \Drupal\Core\Database\Connection
+   */
+  public $dbh;
+
+  /**
+   * The client database Statement object.
+   *
+   * For a \PDO client connection, this will be a \PDOStatement object.
+   *
+   * @var object
+   */
+  protected $clientStatement;
+
+  /**
+   * Is rowCount() execution allowed.
+   *
+   * @var bool
+   */
+  public $allowRowCount = FALSE;
+
+  /**
+   * Constructs a StatementWrapper object.
+   *
+   * @param \Drupal\Core\Database\Connection $connection
+   *   Drupal database connection object.
+   * @param object $client_connection
+   *   Client database connection object, for example \PDO.
+   * @param string $query
+   *   The SQL query string.
+   * @param array $options
+   *   Array of query options.
+   */
+  public function __construct(Connection $connection, $client_connection, string $query, array $options) {
+    $this->dbh = $connection;
+    $this->clientStatement = $client_connection->prepare($query, $options);
+    $this->setFetchMode(\PDO::FETCH_OBJ);
+  }
+
+  /**
+   * Implements the magic __get() method.
+   *
+   * @deprecated in drupal:9.1.0 and is removed from drupal:10.0.0. Access the
+   *   client-level statement object via ::getClientStatement().
+   *
+   * @see https://www.drupal.org/node/3177488
+   */
+  public function __get($name) {
+    if ($name === 'queryString') {
+      @trigger_error("StatementWrapper::\${$name} should not be accessed in drupal:9.1.0 and will error in drupal:10.0.0. Access the client-level statement object via ::getClientStatement(). See https://www.drupal.org/node/3177488", E_USER_DEPRECATED);
+      return $this->getClientStatement()->queryString;
+    }
+  }
+
+  /**
+   * Implements the magic __call() method.
+   *
+   * @deprecated in drupal:9.1.0 and is removed from drupal:10.0.0. Access the
+   *   client-level statement object via ::getClientStatement().
+   *
+   * @see https://www.drupal.org/node/3177488
+   */
+  public function __call($method, $arguments) {
+    if (is_callable([$this->getClientStatement(), $method])) {
+      @trigger_error("StatementWrapper::{$method} should not be called in drupal:9.1.0 and will error in drupal:10.0.0. Access the client-level statement object via ::getClientStatement(). See https://www.drupal.org/node/3177488", E_USER_DEPRECATED);
+      return call_user_func_array([$this->getClientStatement(), $method], $arguments);
+    }
+    throw new \BadMethodCallException($method);
+  }
+
+  /**
+   * Returns the client-level database statement object.
+   *
+   * This method should normally be used only within database driver code.
+   *
+   * @return object
+   *   The client-level database statement, for example \PDOStatement.
+   */
+  public function getClientStatement() {
+    return $this->clientStatement;
+  }
+
+  /**
+   * {@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;
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function fetchField($index = 0) {
+    // Call \PDOStatement::fetchColumn to fetch the field.
+    return $this->clientStatement->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, $cursor_orientation);
+
+      case 3:
+      default:
+        return $this->clientStatement->fetch($mode, $cursor_orientation, $cursor_offset);
+    }
+  }
+
+  /**
+   * {@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());
+  }
+
+  /**
+   * Bind a column to a PHP variable.
+   *
+   * @param mixed $column
+   *   Number of the column (1-indexed) or name of the column in the result set.
+   *   If using the column name, be aware that the name should match the case of
+   *   the column, as returned by the driver.
+   * @param mixed $param
+   *   Name of the PHP variable to which the column will be bound.
+   * @param int $type
+   *   (Optional) data type of the parameter, specified by the PDO::PARAM_*
+   *   constants.
+   * @param int $maxlen
+   *   (Optional) a hint for pre-allocation.
+   * @param mixed $driverdata
+   *   (Optional) optional parameter(s) for the driver.
+   *
+   * @return bool
+   *   Returns TRUE on success or FALSE on failure.
+   *
+   * @deprecated in drupal:9.1.0 and is removed from drupal:10.0.0.
+   *   StatementWrapper::bindColumn should not be called. Access the
+   *   client-level statement object via ::getClientStatement().
+   *
+   * @see https://www.drupal.org/node/3177488
+   */
+  public function bindColumn($column, &$param, int $type = 0, int $maxlen = 0, $driverdata = NULL): bool {
+    @trigger_error("StatementWrapper::bindColumn should not be called in drupal:9.1.0 and will error in drupal:10.0.0. Access the client-level statement object via ::getClientStatement(). See https://www.drupal.org/node/3177488", E_USER_DEPRECATED);
+    switch (func_num_args()) {
+      case 2:
+        return $this->clientStatement->bindColumn($column, $param);
+
+      case 3:
+        return $this->clientStatement->bindColumn($column, $param, $type);
+
+      case 4:
+        return $this->clientStatement->bindColumn($column, $param, $type, $maxlen);
+
+      case 5:
+        return $this->clientStatement->bindColumn($column, $param, $type, $maxlen, $driverdata);
+
+    }
+  }
+
+  /**
+   * 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.
+   *
+   * @deprecated in drupal:9.1.0 and is removed from drupal:10.0.0.
+   *   StatementWrapper::bindParam should not be called. Access the
+   *   client-level statement object via ::getClientStatement().
+   *
+   * @see https://www.drupal.org/node/3177488
+   */
+  public function bindParam($parameter, &$variable, int $data_type = \PDO::PARAM_STR, int $length = 0, $driver_options = NULL) : bool {
+    @trigger_error("StatementWrapper::bindParam should not be called in drupal:9.1.0 and will error in drupal:10.0.0. Access the client-level statement object via ::getClientStatement(). See https://www.drupal.org/node/3177488", E_USER_DEPRECATED);
+    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);
+
+    }
+  }
+
+}
diff --git a/core/tests/Drupal/KernelTests/Core/Database/StatementWrapperLegacyTest.php b/core/tests/Drupal/KernelTests/Core/Database/StatementWrapperLegacyTest.php
new file mode 100644
index 0000000000000000000000000000000000000000..56609ee8a6a317a3bbee4372aace555040f12536
--- /dev/null
+++ b/core/tests/Drupal/KernelTests/Core/Database/StatementWrapperLegacyTest.php
@@ -0,0 +1,73 @@
+<?php
+
+namespace Drupal\KernelTests\Core\Database;
+
+use Drupal\Core\Database\StatementWrapper;
+
+/**
+ * Tests the deprecations of the StatementWrapper class.
+ *
+ * @coversDefaultClass \Drupal\Core\Database\StatementWrapper
+ * @group legacy
+ * @group Database
+ */
+class StatementWrapperLegacyTest extends DatabaseTestBase {
+
+  protected $statement;
+
+  /**
+   * {@inheritdoc}
+   */
+  protected function setUp(): void {
+    parent::setUp();
+    $this->statement = $this->connection->prepareStatement('SELECT * FROM {test}', []);
+    if (!$this->statement instanceof StatementWrapper) {
+      $this->markTestSkipped('This test only works for drivers implementing Drupal\Core\Database\StatementWrapper.');
+    }
+  }
+
+  /**
+   * @covers ::getQueryString
+   */
+  public function testQueryString() {
+    $this->expectDeprecation('StatementWrapper::$queryString should not be accessed in drupal:9.1.0 and will error in drupal:10.0.0. Access the client-level statement object via ::getClientStatement(). See https://www.drupal.org/node/3177488');
+    $this->assertStringContainsString('SELECT * FROM ', $this->statement->queryString);
+    $this->assertStringContainsString('SELECT * FROM ', $this->statement->getQueryString());
+  }
+
+  /**
+   * Tests calling a non existing \PDOStatement method.
+   */
+  public function testMissingMethod() {
+    $this->expectException('\BadMethodCallException');
+    $this->statement->boo();
+  }
+
+  /**
+   * Tests calling an existing \PDOStatement method.
+   */
+  public function testClientStatementMethod() {
+    $this->expectDeprecation('StatementWrapper::columnCount should not be called in drupal:9.1.0 and will error in drupal:10.0.0. Access the client-level statement object via ::getClientStatement(). See https://www.drupal.org/node/3177488');
+    $this->statement->execute();
+    $this->assertEquals(4, $this->statement->columnCount());
+  }
+
+  /**
+   * @covers ::bindParam
+   */
+  public function testBindParam() {
+    $this->expectDeprecation('StatementWrapper::bindParam should not be called in drupal:9.1.0 and will error in drupal:10.0.0. Access the client-level statement object via ::getClientStatement(). See https://www.drupal.org/node/3177488');
+    $test = NULL;
+    $this->assertTrue($this->statement->bindParam(':name', $test));
+  }
+
+  /**
+   * @covers ::bindColumn
+   */
+  public function testBindColumn() {
+    $this->expectDeprecation('StatementWrapper::bindColumn should not be called in drupal:9.1.0 and will error in drupal:10.0.0. Access the client-level statement object via ::getClientStatement(). See https://www.drupal.org/node/3177488');
+    $test = NULL;
+    $this->assertTrue($this->statement->bindColumn(1, $test));
+  }
+
+}
diff --git a/core/tests/Drupal/Tests/Core/Database/ConnectionTest.php b/core/tests/Drupal/Tests/Core/Database/ConnectionTest.php
index c4e4a1c26835a02dc72534f308816fec5322f33a..ab8ad7b28bbda4ea5451acf6b1acabfe07798d46 100644
--- a/core/tests/Drupal/Tests/Core/Database/ConnectionTest.php
+++ b/core/tests/Drupal/Tests/Core/Database/ConnectionTest.php
@@ -4,6 +4,7 @@
 
 use Composer\Autoload\ClassLoader;
 use Drupal\Core\Database\Statement;
+use Drupal\Core\Database\StatementWrapper;
 use Drupal\Tests\Core\Database\Stub\StubConnection;
 use Drupal\Tests\Core\Database\Stub\StubPDO;
 use Drupal\Tests\Core\Database\Stub\Driver;
@@ -604,6 +605,21 @@ public function testNamespaceDefault() {
     $this->assertSame('Drupal\Tests\Core\Database\Stub', $connection->getConnectionOptions()['namespace']);
   }
 
+  /**
+   * Tests deprecation of the Statement class.
+   *
+   * @group legacy
+   */
+  public function testStatementDeprecation() {
+    if (PHP_VERSION_ID >= 80000) {
+      $this->markTestSkipped('Drupal\Core\Database\Statement is incompatible with PHP 8.0. Remove in https://www.drupal.org/node/3177490');
+    }
+    $this->expectDeprecation('\Drupal\Core\Database\Statement 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/3177488');
+    $mock_statement = $this->getMockBuilder(Statement::class)
+      ->disableOriginalConstructor()
+      ->getMock();
+  }
+
   /**
    * Test rtrim() of query strings.
    *
@@ -613,7 +629,7 @@ public function testQueryTrim($expected, $query, $options) {
     $mock_pdo = $this->getMockBuilder(StubPdo::class)
       ->setMethods(['execute', 'prepare', 'setAttribute'])
       ->getMock();
-    $mock_statement = $this->getMockBuilder(Statement::class)
+    $mock_statement = $this->getMockBuilder(StatementWrapper::class)
       ->disableOriginalConstructor()
       ->getMock();
 
@@ -684,4 +700,18 @@ public function testLegacyDatabaseDriverInRootDriversDirectory() {
     $this->assertEquals($namespace, $connection->getConnectionOptions()['namespace']);
   }
 
+  /**
+   * Tests the deprecation of \Drupal\Core\Database\Connection::$statementClass.
+   *
+   * @group legacy
+   */
+  public function testPdoStatementClass() {
+    if (PHP_VERSION_ID >= 80000) {
+      $this->markTestSkipped('Drupal\Core\Database\Statement is incompatible with PHP 8.0. Remove in https://www.drupal.org/node/3177490');
+    }
+    $this->expectDeprecation('\Drupal\Core\Database\Connection::$statementClass 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/3177488');
+    $mock_pdo = $this->createMock(StubPDO::class);
+    new StubConnection($mock_pdo, ['namespace' => 'Drupal\\Tests\\Core\\Database\\Stub\\Driver'], ['"', '"'], Statement::class);
+  }
+
 }
diff --git a/core/tests/Drupal/Tests/Core/Database/Stub/StubConnection.php b/core/tests/Drupal/Tests/Core/Database/Stub/StubConnection.php
index 702f4b0ba00065eb6375b121ae2d529c186b20b9..146ba0891cb5a6e638f2fc3b27117eaf5d235e0a 100644
--- a/core/tests/Drupal/Tests/Core/Database/Stub/StubConnection.php
+++ b/core/tests/Drupal/Tests/Core/Database/Stub/StubConnection.php
@@ -5,6 +5,7 @@
 use Drupal\Core\Database\Connection;
 use Drupal\Core\Database\Log;
 use Drupal\Core\Database\StatementEmpty;
+use Drupal\Core\Database\StatementWrapper;
 
 /**
  * A stub of the abstract Connection class for testing purposes.
@@ -13,6 +14,16 @@
  */
 class StubConnection extends Connection {
 
+  /**
+   * {@inheritdoc}
+   */
+  protected $statementClass = NULL;
+
+  /**
+   * {@inheritdoc}
+   */
+  protected $statementWrapperClass = StatementWrapper::class;
+
   /**
    * Public property so we can test driver loading mechanism.
    *
@@ -30,9 +41,15 @@ class StubConnection extends Connection {
    *   An array of options for the connection.
    * @param string[]|null $identifier_quotes
    *   The identifier quote characters. Defaults to an empty strings.
+   * @param string|null $statement_class
+   *   A class to use as a statement class for deprecation testing.
    */
-  public function __construct(\PDO $connection, array $connection_options, $identifier_quotes = ['', '']) {
+  public function __construct(\PDO $connection, array $connection_options, $identifier_quotes = ['', ''], $statement_class = NULL) {
     $this->identifierQuotes = $identifier_quotes;
+    if ($statement_class) {
+      $this->statementClass = $statement_class;
+      $this->statementWrapperClass = NULL;
+    }
     parent::__construct($connection, $connection_options);
   }