first commit

core/modules/pgsql/src/Driver/Database/pgsql/Connection.php

@@ -0,0 +1,547 @@
+<?php
+
+namespace Drupal\pgsql\Driver\Database\pgsql;
+
+use Drupal\Core\Database\Connection as DatabaseConnection;
+use Drupal\Core\Database\Database;
+use Drupal\Core\Database\DatabaseAccessDeniedException;
+use Drupal\Core\Database\DatabaseNotFoundException;
+use Drupal\Core\Database\ExceptionHandler;
+use Drupal\Core\Database\Query\Condition;
+use Drupal\Core\Database\StatementInterface;
+use Drupal\Core\Database\StatementWrapperIterator;
+use Drupal\Core\Database\SupportsTemporaryTablesInterface;
+use Drupal\Core\Database\Transaction\TransactionManagerInterface;
+
+// cSpell:ignore ilike nextval
+
+/**
+ * @addtogroup database
+ * @{
+ */
+
+/**
+ * PostgreSQL implementation of \Drupal\Core\Database\Connection.
+ */
+class Connection extends DatabaseConnection implements SupportsTemporaryTablesInterface {
+
+  /**
+   * The name by which to obtain a lock for retrieve the next insert id.
+   */
+  const POSTGRESQL_NEXTID_LOCK = 1000;
+
+  /**
+   * Error code for "Unknown database" error.
+   */
+  const DATABASE_NOT_FOUND = 7;
+
+  /**
+   * Error code for "Connection failure" errors.
+   *
+   * Technically this is an internal error code that will only be shown in the
+   * PDOException message. It will need to get extracted.
+   */
+  const CONNECTION_FAILURE = '08006';
+
+  /**
+   * {@inheritdoc}
+   */
+  protected $statementWrapperClass = StatementWrapperIterator::class;
+
+  /**
+   * A map of condition operators to PostgreSQL operators.
+   *
+   * In PostgreSQL, 'LIKE' is case-sensitive. ILIKE should be used for
+   * case-insensitive statements.
+   */
+  protected static $postgresqlConditionOperatorMap = [
+    'LIKE' => ['operator' => 'ILIKE'],
+    'LIKE BINARY' => ['operator' => 'LIKE'],
+    'NOT LIKE' => ['operator' => 'NOT ILIKE'],
+    'REGEXP' => ['operator' => '~*'],
+    'NOT REGEXP' => ['operator' => '!~*'],
+  ];
+
+  /**
+   * {@inheritdoc}
+   */
+  protected $transactionalDDLSupport = TRUE;
+
+  /**
+   * {@inheritdoc}
+   */
+  protected $identifierQuotes = ['"', '"'];
+
+  /**
+   * An array of transaction savepoints.
+   *
+   * The main use for this array is to store information about transaction
+   * savepoints opened to to mimic MySql's InnoDB functionality, which provides
+   * an inherent savepoint before any query in a transaction.
+   *
+   * @see ::addSavepoint()
+   * @see ::releaseSavepoint()
+   * @see ::rollbackSavepoint()
+   *
+   * @var array<string,Transaction>
+   */
+  protected array $savepoints = [];
+
+  /**
+   * Constructs a connection object.
+   */
+  public function __construct(\PDO $connection, array $connection_options) {
+    // Sanitize the schema name here, so we do not have to do it in other
+    // functions.
+    if (isset($connection_options['schema']) && ($connection_options['schema'] !== 'public')) {
+      $connection_options['schema'] = preg_replace('/[^A-Za-z0-9_]+/', '', $connection_options['schema']);
+    }
+
+    // We need to set the connectionOptions before the parent, because setPrefix
+    // needs this.
+    $this->connectionOptions = $connection_options;
+
+    parent::__construct($connection, $connection_options);
+
+    // Force PostgreSQL to use the UTF-8 character set by default.
+    $this->connection->exec("SET NAMES 'UTF8'");
+
+    // Execute PostgreSQL init_commands.
+    if (isset($connection_options['init_commands'])) {
+      $this->connection->exec(implode('; ', $connection_options['init_commands']));
+    }
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  protected function setPrefix($prefix) {
+    assert(is_string($prefix), 'The \'$prefix\' argument to ' . __METHOD__ . '() must be a string');
+    $this->prefix = $prefix;
+
+    // Add the schema name if it is not set to public, otherwise it will use the
+    // default schema name.
+    $quoted_schema = '';
+    if (isset($this->connectionOptions['schema']) && ($this->connectionOptions['schema'] !== 'public')) {
+      $quoted_schema = $this->identifierQuotes[0] . $this->connectionOptions['schema'] . $this->identifierQuotes[1] . '.';
+    }
+
+    $this->tablePlaceholderReplacements = [
+      $quoted_schema . $this->identifierQuotes[0] . str_replace('.', $this->identifierQuotes[1] . '.' . $this->identifierQuotes[0], $prefix),
+      $this->identifierQuotes[1],
+    ];
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public static function open(array &$connection_options = []) {
+    // Default to TCP connection on port 5432.
+    if (empty($connection_options['port'])) {
+      $connection_options['port'] = 5432;
+    }
+
+    // PostgreSQL in trust mode doesn't require a password to be supplied.
+    if (empty($connection_options['password'])) {
+      $connection_options['password'] = NULL;
+    }
+    // If the password contains a backslash it is treated as an escape character
+    // http://bugs.php.net/bug.php?id=53217
+    // so backslashes in the password need to be doubled up.
+    // The bug was reported against pdo_pgsql 1.0.2, backslashes in passwords
+    // will break on this doubling up when the bug is fixed, so check the version
+    // elseif (phpversion('pdo_pgsql') < 'version_this_was_fixed_in') {
+    else {
+      $connection_options['password'] = str_replace('\\', '\\\\', $connection_options['password']);
+    }
+
+    $connection_options['database'] = (!empty($connection_options['database']) ? $connection_options['database'] : 'template1');
+    $dsn = 'pgsql:host=' . $connection_options['host'] . ' dbname=' . $connection_options['database'] . ' port=' . $connection_options['port'];
+
+    // Allow PDO options to be overridden.
+    $connection_options += [
+      'pdo' => [],
+    ];
+    $connection_options['pdo'] += [
+      \PDO::ATTR_ERRMODE => \PDO::ERRMODE_EXCEPTION,
+      // Prepared statements are most effective for performance when queries
+      // are recycled (used several times). However, if they are not re-used,
+      // prepared statements become inefficient. Since most of Drupal's
+      // prepared queries are not re-used, it should be faster to emulate
+      // the preparation than to actually ready statements for re-use. If in
+      // doubt, reset to FALSE and measure performance.
+      \PDO::ATTR_EMULATE_PREPARES => TRUE,
+      // Convert numeric values to strings when fetching.
+      \PDO::ATTR_STRINGIFY_FETCHES => TRUE,
+    ];
+
+    try {
+      $pdo = new \PDO($dsn, $connection_options['username'], $connection_options['password'], $connection_options['pdo']);
+    }
+    catch (\PDOException $e) {
+      if (static::getSQLState($e) == static::CONNECTION_FAILURE) {
+        if (str_contains($e->getMessage(), 'password authentication failed for user')) {
+          throw new DatabaseAccessDeniedException($e->getMessage(), $e->getCode(), $e);
+        }
+        elseif (str_contains($e->getMessage(), 'database') && str_contains($e->getMessage(), 'does not exist')) {
+          throw new DatabaseNotFoundException($e->getMessage(), $e->getCode(), $e);
+        }
+      }
+      throw $e;
+    }
+
+    return $pdo;
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function query($query, array $args = [], $options = []) {
+    $options += $this->defaultOptions();
+
+    // The PDO PostgreSQL driver has a bug which doesn't type cast booleans
+    // correctly when parameters are bound using associative arrays.
+    // @see http://bugs.php.net/bug.php?id=48383
+    foreach ($args as &$value) {
+      if (is_bool($value)) {
+        $value = (int) $value;
+      }
+    }
+
+    // We need to wrap queries with a savepoint if:
+    // - Currently in a transaction.
+    // - A 'mimic_implicit_commit' does not exist already.
+    // - The query is not a savepoint query.
+    $wrap_with_savepoint = $this->inTransaction() &&
+      !$this->transactionManager()->has('mimic_implicit_commit') &&
+      !(is_string($query) && (
+        stripos($query, 'ROLLBACK TO SAVEPOINT ') === 0 ||
+        stripos($query, 'RELEASE SAVEPOINT ') === 0 ||
+        stripos($query, 'SAVEPOINT ') === 0
+      )
+    );
+    if ($wrap_with_savepoint) {
+      // Create a savepoint so we can rollback a failed query. This is so we can
+      // mimic MySQL and SQLite transactions which don't fail if a single query
+      // fails. This is important for tables that are created on demand. For
+      // example, \Drupal\Core\Cache\DatabaseBackend.
+      $this->addSavepoint();
+      try {
+        $return = parent::query($query, $args, $options);
+        $this->releaseSavepoint();
+      }
+      catch (\Exception $e) {
+        $this->rollbackSavepoint();
+        throw $e;
+      }
+    }
+    else {
+      $return = parent::query($query, $args, $options);
+    }
+
+    return $return;
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function prepareStatement(string $query, array $options, bool $allow_row_count = FALSE): StatementInterface {
+    // mapConditionOperator converts some operations (LIKE, REGEXP, etc.) to
+    // PostgreSQL equivalents (ILIKE, ~*, etc.). However PostgreSQL doesn't
+    // 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, $allow_row_count);
+  }
+
+  public function queryRange($query, $from, $count, array $args = [], array $options = []) {
+    return $this->query($query . ' LIMIT ' . (int) $count . ' OFFSET ' . (int) $from, $args, $options);
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function queryTemporary($query, array $args = [], array $options = []) {
+    $tablename = 'db_temporary_' . uniqid();
+    $this->query('CREATE TEMPORARY TABLE {' . $tablename . '} AS ' . $query, $args, $options);
+    return $tablename;
+  }
+
+  public function driver() {
+    return 'pgsql';
+  }
+
+  public function databaseType() {
+    return 'pgsql';
+  }
+
+  /**
+   * Overrides \Drupal\Core\Database\Connection::createDatabase().
+   *
+   * @param string $database
+   *   The name of the database to create.
+   *
+   * @throws \Drupal\Core\Database\DatabaseNotFoundException
+   */
+  public function createDatabase($database) {
+    // Escape the database name.
+    $database = Database::getConnection()->escapeDatabase($database);
+    $db_created = FALSE;
+
+    // Try to determine the proper locales for character classification and
+    // collation. If we could determine locales other than 'en_US', try creating
+    // the database with these first.
+    $ctype = setlocale(LC_CTYPE, 0);
+    $collate = setlocale(LC_COLLATE, 0);
+    if ($ctype && $collate) {
+      try {
+        $this->connection->exec("CREATE DATABASE $database WITH TEMPLATE template0 ENCODING='UTF8' LC_CTYPE='$ctype.UTF-8' LC_COLLATE='$collate.UTF-8'");
+        $db_created = TRUE;
+      }
+      catch (\Exception $e) {
+        // It might be that the server is remote and does not support the
+        // locale and collation of the webserver, so we will try again.
+      }
+    }
+
+    // Otherwise fall back to creating the database using the 'en_US' locales.
+    if (!$db_created) {
+      try {
+        $this->connection->exec("CREATE DATABASE $database WITH TEMPLATE template0 ENCODING='UTF8' LC_CTYPE='en_US.UTF-8' LC_COLLATE='en_US.UTF-8'");
+      }
+      catch (\Exception $e) {
+        // If the database can't be created with the 'en_US' locale either,
+        // we're finally throwing an exception.
+        throw new DatabaseNotFoundException($e->getMessage());
+      }
+    }
+  }
+
+  public function mapConditionOperator($operator) {
+    return static::$postgresqlConditionOperatorMap[$operator] ?? NULL;
+  }
+
+  /**
+   * Creates the appropriate sequence name for a given table and serial field.
+   *
+   * This method should only be called by the driver's code.
+   *
+   * @param string $table
+   *   The table name to use for the sequence.
+   * @param string $field
+   *   The field name to use for the sequence.
+   *
+   * @return string
+   *   A table prefix-parsed string for the sequence name.
+   *
+   * @internal
+   */
+  public function makeSequenceName($table, $field) {
+    $sequence_name = $this->prefixTables('{' . $table . '}_' . $field . '_seq');
+    // Remove identifier quotes as we are constructing a new name from a
+    // prefixed and quoted table name.
+    return str_replace($this->identifierQuotes, '', $sequence_name);
+  }
+
+  /**
+   * Retrieve a the next id in a sequence.
+   *
+   * PostgreSQL has built in sequences. We'll use these instead of inserting
+   * and updating a sequences table.
+   */
+  public function nextId($existing = 0) {
+    @trigger_error('Drupal\Core\Database\Connection::nextId() is deprecated in drupal:10.2.0 and is removed from drupal:11.0.0. Modules should use instead the keyvalue storage for the last used id. See https://www.drupal.org/node/3349345', E_USER_DEPRECATED);
+    // Retrieve the name of the sequence. This information cannot be cached
+    // because the prefix may change, for example, like it does in tests.
+    $sequence_name = $this->makeSequenceName('sequences', 'value');
+
+    // When PostgreSQL gets a value too small then it will lock the table,
+    // retry the INSERT and if it's still too small then alter the sequence.
+    $id = $this->query("SELECT nextval('" . $sequence_name . "')")->fetchField();
+    if ($id > $existing) {
+      return $id;
+    }
+
+    // PostgreSQL advisory locks are simply locks to be used by an
+    // application such as Drupal. This will prevent other Drupal processes
+    // from altering the sequence while we are.
+    $this->query("SELECT pg_advisory_lock(" . self::POSTGRESQL_NEXTID_LOCK . ")");
+
+    // While waiting to obtain the lock, the sequence may have been altered
+    // so lets try again to obtain an adequate value.
+    $id = $this->query("SELECT nextval('" . $sequence_name . "')")->fetchField();
+    if ($id > $existing) {
+      $this->query("SELECT pg_advisory_unlock(" . self::POSTGRESQL_NEXTID_LOCK . ")");
+      return $id;
+    }
+
+    // Reset the sequence to a higher value than the existing id.
+    $this->query("ALTER SEQUENCE " . $sequence_name . " RESTART WITH " . ($existing + 1));
+
+    // Retrieve the next id. We know this will be as high as we want it.
+    $id = $this->query("SELECT nextval('" . $sequence_name . "')")->fetchField();
+
+    $this->query("SELECT pg_advisory_unlock(" . self::POSTGRESQL_NEXTID_LOCK . ")");
+
+    return $id;
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function getFullQualifiedTableName($table) {
+    $options = $this->getConnectionOptions();
+    $schema = $options['schema'] ?? 'public';
+
+    // The fully qualified table name in PostgreSQL is in the form of
+    // <database>.<schema>.<table>.
+    return $options['database'] . '.' . $schema . '.' . $this->getPrefix() . $table;
+  }
+
+  /**
+   * Add a new savepoint with a unique name.
+   *
+   * The main use for this method is to mimic InnoDB functionality, which
+   * provides an inherent savepoint before any query in a transaction.
+   *
+   * @param $savepoint_name
+   *   A string representing the savepoint name. By default,
+   *   "mimic_implicit_commit" is used.
+   */
+  public function addSavepoint($savepoint_name = 'mimic_implicit_commit') {
+    if ($this->inTransaction()) {
+      $this->savepoints[$savepoint_name] = $this->startTransaction($savepoint_name);
+    }
+  }
+
+  /**
+   * Release a savepoint by name.
+   *
+   * @param $savepoint_name
+   *   A string representing the savepoint name. By default,
+   *   "mimic_implicit_commit" is used.
+   */
+  public function releaseSavepoint($savepoint_name = 'mimic_implicit_commit') {
+    if ($this->inTransaction() && $this->transactionManager()->has($savepoint_name)) {
+      unset($this->savepoints[$savepoint_name]);
+    }
+  }
+
+  /**
+   * Rollback a savepoint by name if it exists.
+   *
+   * @param $savepoint_name
+   *   A string representing the savepoint name. By default,
+   *   "mimic_implicit_commit" is used.
+   */
+  public function rollbackSavepoint($savepoint_name = 'mimic_implicit_commit') {
+    if ($this->inTransaction() && $this->transactionManager()->has($savepoint_name)) {
+      $this->savepoints[$savepoint_name]->rollBack();
+      unset($this->savepoints[$savepoint_name]);
+    }
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function hasJson(): bool {
+    try {
+      return (bool) $this->query('SELECT JSON_TYPEOF(\'1\')');
+    }
+    catch (\Exception $e) {
+      return FALSE;
+    }
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function exceptionHandler() {
+    return new ExceptionHandler();
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function select($table, $alias = NULL, array $options = []) {
+    return new Select($this, $table, $alias, $options);
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function insert($table, array $options = []) {
+    return new Insert($this, $table, $options);
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function merge($table, array $options = []) {
+    return new Merge($this, $table, $options);
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function upsert($table, array $options = []) {
+    return new Upsert($this, $table, $options);
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function update($table, array $options = []) {
+    return new Update($this, $table, $options);
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function delete($table, array $options = []) {
+    return new Delete($this, $table, $options);
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function truncate($table, array $options = []) {
+    return new Truncate($this, $table, $options);
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function schema() {
+    if (empty($this->schema)) {
+      $this->schema = new Schema($this);
+    }
+    return $this->schema;
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function condition($conjunction) {
+    return new Condition($conjunction);
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  protected function driverTransactionManager(): TransactionManagerInterface {
+    return new TransactionManager($this);
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function startTransaction($name = '') {
+    return $this->transactionManager()->push($name);
+  }
+
+}
+
+/**
+ * @} End of "addtogroup database".
+ */

core/modules/pgsql/src/Driver/Database/pgsql/Delete.php

@@ -0,0 +1,39 @@
+<?php
+
+namespace Drupal\pgsql\Driver\Database\pgsql;
+
+use Drupal\Core\Database\Query\Delete as QueryDelete;
+
+/**
+ * PostgreSQL implementation of \Drupal\Core\Database\Query\Delete.
+ */
+class Delete extends QueryDelete {
+
+  /**
+   * {@inheritdoc}
+   */
+  public function __construct(Connection $connection, string $table, array $options = []) {
+    // @todo Remove the __construct in Drupal 11.
+    // @see https://www.drupal.org/project/drupal/issues/3256524
+    parent::__construct($connection, $table, $options);
+    unset($this->queryOptions['return']);
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function execute() {
+    $this->connection->addSavepoint();
+    try {
+      $result = parent::execute();
+    }
+    catch (\Exception $e) {
+      $this->connection->rollbackSavepoint();
+      throw $e;
+    }
+    $this->connection->releaseSavepoint();
+
+    return $result;
+  }
+
+}

core/modules/pgsql/src/Driver/Database/pgsql/Insert.php

@@ -0,0 +1,156 @@
+<?php
+
+namespace Drupal\pgsql\Driver\Database\pgsql;
+
+use Drupal\Core\Database\DatabaseExceptionWrapper;
+use Drupal\Core\Database\Query\Insert as QueryInsert;
+
+// cSpell:ignore nextval setval
+
+/**
+ * @ingroup database
+ * @{
+ */
+
+/**
+ * PostgreSQL implementation of \Drupal\Core\Database\Query\Insert.
+ */
+class Insert extends QueryInsert {
+
+  /**
+   * {@inheritdoc}
+   */
+  public function __construct(Connection $connection, string $table, array $options = []) {
+    // @todo Remove the __construct in Drupal 11.
+    // @see https://www.drupal.org/project/drupal/issues/3256524
+    parent::__construct($connection, $table, $options);
+    unset($this->queryOptions['return']);
+  }
+
+  public function execute() {
+    if (!$this->preExecute()) {
+      return NULL;
+    }
+
+    $stmt = $this->connection->prepareStatement((string) $this, $this->queryOptions);
+
+    // Fetch the list of blobs and sequences used on that table.
+    $table_information = $this->connection->schema()->queryTableInformation($this->table);
+
+    $max_placeholder = 0;
+    $blobs = [];
+    $blob_count = 0;
+    foreach ($this->insertValues as $insert_values) {
+      foreach ($this->insertFields as $idx => $field) {
+        if (isset($table_information->blob_fields[$field]) && $insert_values[$idx] !== NULL) {
+          $blobs[$blob_count] = fopen('php://memory', 'a');
+          fwrite($blobs[$blob_count], $insert_values[$idx]);
+          rewind($blobs[$blob_count]);
+
+          $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->getClientStatement()->bindParam(':db_insert_placeholder_' . $max_placeholder++, $insert_values[$idx]);
+        }
+      }
+      // Check if values for a serial field has been passed.
+      if (!empty($table_information->serial_fields)) {
+        foreach ($table_information->serial_fields as $index => $serial_field) {
+          $serial_key = array_search($serial_field, $this->insertFields);
+          if ($serial_key !== FALSE) {
+            $serial_value = $insert_values[$serial_key];
+
+            // Sequences must be greater than or equal to 1.
+            if ($serial_value === NULL || !$serial_value) {
+              $serial_value = 1;
+            }
+            // Set the sequence to the bigger value of either the passed
+            // value or the max value of the column. It can happen that another
+            // thread calls nextval() which could lead to a serial number being
+            // used twice. However, trying to insert a value into a serial
+            // column should only be done in very rare cases and is not thread
+            // safe by definition.
+            $this->connection->query("SELECT setval('" . $table_information->sequences[$index] . "', GREATEST(MAX(" . $serial_field . "), :serial_value)) FROM {" . $this->table . "}", [':serial_value' => (int) $serial_value]);
+          }
+        }
+      }
+    }
+    if (!empty($this->fromQuery)) {
+      // bindParam stores only a reference to the variable that is followed when
+      // the statement is executed. We pass $arguments[$key] instead of $value
+      // because the second argument to bindParam is passed by reference and
+      // the foreach statement assigns the element to the existing reference.
+      $arguments = $this->fromQuery->getArguments();
+      foreach ($arguments as $key => $value) {
+        $stmt->getClientStatement()->bindParam($key, $arguments[$key]);
+      }
+    }
+
+    // Create a savepoint so we can rollback a failed query. This is so we can
+    // mimic MySQL and SQLite transactions which don't fail if a single query
+    // fails. This is important for tables that are created on demand. For
+    // example, \Drupal\Core\Cache\DatabaseBackend.
+    $this->connection->addSavepoint();
+    try {
+      $stmt->execute(NULL, $this->queryOptions);
+      if (isset($table_information->serial_fields[0])) {
+        $last_insert_id = $stmt->fetchField();
+      }
+      $this->connection->releaseSavepoint();
+    }
+    catch (\Exception $e) {
+      $this->connection->rollbackSavepoint();
+      $this->connection->exceptionHandler()->handleExecutionException($e, $stmt, [], $this->queryOptions);
+    }
+
+    // Re-initialize the values array so that we can re-use this query.
+    $this->insertValues = [];
+
+    return $last_insert_id ?? NULL;
+  }
+
+  public function __toString() {
+    // Create a sanitized comment string to prepend to the query.
+    $comments = $this->connection->makeComment($this->comments);
+
+    // Default fields are always placed first for consistency.
+    $insert_fields = array_merge($this->defaultFields, $this->insertFields);
+
+    $insert_fields = array_map(function ($f) {
+      return $this->connection->escapeField($f);
+    }, $insert_fields);
+
+    // If we're selecting from a SelectQuery, finish building the query and
+    // pass it back, as any remaining options are irrelevant.
+    if (!empty($this->fromQuery)) {
+      $insert_fields_string = $insert_fields ? ' (' . implode(', ', $insert_fields) . ') ' : ' ';
+      $query = $comments . 'INSERT INTO {' . $this->table . '}' . $insert_fields_string . $this->fromQuery;
+    }
+    else {
+      $query = $comments . 'INSERT INTO {' . $this->table . '} (' . implode(', ', $insert_fields) . ') VALUES ';
+
+      $values = $this->getInsertPlaceholderFragment($this->insertValues, $this->defaultFields);
+      $query .= implode(', ', $values);
+    }
+    try {
+      // Fetch the list of blobs and sequences used on that table.
+      $table_information = $this->connection->schema()->queryTableInformation($this->table);
+      if (isset($table_information->serial_fields[0])) {
+        // Use RETURNING syntax to get the last insert ID in the same INSERT
+        // query, see https://www.postgresql.org/docs/12/dml-returning.html.
+        $query .= ' RETURNING ' . $table_information->serial_fields[0];
+      }
+    }
+    catch (DatabaseExceptionWrapper $e) {
+      // If we fail to get the table information it is probably because the
+      // table does not exist yet so adding the returning statement is pointless
+      // because the query will fail. This happens for tables created on demand,
+      // for example, cache tables.
+    }
+    return $query;
+  }
+
+}

core/modules/pgsql/src/Driver/Database/pgsql/Install/Tasks.php

@@ -0,0 +1,328 @@
+<?php
+
+namespace Drupal\pgsql\Driver\Database\pgsql\Install;
+
+use Drupal\Core\Database\Database;
+use Drupal\Core\Database\Install\Tasks as InstallTasks;
+use Drupal\Core\Database\DatabaseNotFoundException;
+
+// cspell:ignore trgm
+
+/**
+ * Specifies installation tasks for PostgreSQL databases.
+ */
+class Tasks extends InstallTasks {
+
+  /**
+   * Minimum required PostgreSQL version.
+   *
+   * The contrib extension pg_trgm is supposed to be installed.
+   *
+   * @see https://www.postgresql.org/docs/12/pgtrgm.html
+   */
+  const PGSQL_MINIMUM_VERSION = '12';
+
+  /**
+   * {@inheritdoc}
+   */
+  protected $pdoDriver = 'pgsql';
+
+  /**
+   * Constructs a \Drupal\pgsql\Driver\Database\pgsql\Install\Tasks object.
+   */
+  public function __construct() {
+    $this->tasks[] = [
+      'function' => 'checkEncoding',
+      'arguments' => [],
+    ];
+    $this->tasks[] = [
+      'function' => 'checkBinaryOutput',
+      'arguments' => [],
+    ];
+    $this->tasks[] = [
+      'function' => 'checkStandardConformingStrings',
+      'arguments' => [],
+    ];
+    $this->tasks[] = [
+      'function' => 'checkExtensions',
+      'arguments' => [],
+    ];
+    $this->tasks[] = [
+      'function' => 'initializeDatabase',
+      'arguments' => [],
+    ];
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function name() {
+    return t('PostgreSQL');
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function minimumVersion() {
+    return static::PGSQL_MINIMUM_VERSION;
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  protected function connect() {
+    try {
+      // This doesn't actually test the connection.
+      Database::setActiveConnection();
+      // Now actually do a check.
+      Database::getConnection();
+      $this->pass('Drupal can CONNECT to the database ok.');
+    }
+    catch (\Exception $e) {
+      // Attempt to create the database if it is not found.
+      if ($e instanceof DatabaseNotFoundException) {
+        // Remove the database string from connection info.
+        $connection_info = Database::getConnectionInfo();
+        $database = $connection_info['default']['database'];
+        unset($connection_info['default']['database']);
+
+        // In order to change the Database::$databaseInfo array, need to remove
+        // the active connection, then re-add it with the new info.
+        Database::removeConnection('default');
+        Database::addConnectionInfo('default', 'default', $connection_info['default']);
+
+        try {
+          // Now, attempt the connection again; if it's successful, attempt to
+          // create the database.
+          Database::getConnection()->createDatabase($database);
+          Database::closeConnection();
+
+          // Now, restore the database config.
+          Database::removeConnection('default');
+          $connection_info['default']['database'] = $database;
+          Database::addConnectionInfo('default', 'default', $connection_info['default']);
+
+          // Check the database connection.
+          Database::getConnection();
+          $this->pass('Drupal can CONNECT to the database ok.');
+        }
+        catch (DatabaseNotFoundException $e) {
+          // Still no dice; probably a permission issue. Raise the error to the
+          // installer.
+          $this->fail(t('Database %database not found. The server reports the following message when attempting to create the database: %error.', ['%database' => $database, '%error' => $e->getMessage()]));
+        }
+      }
+      else {
+        // Database connection failed for some other reason than a non-existent
+        // database.
+        $this->fail(t('Failed to connect to your database server. The server reports the following message: %error.<ul><li>Is the database server running?</li><li>Does the database exist, and have you entered the correct database name?</li><li>Have you entered the correct username and password?</li><li>Have you entered the correct database hostname and port number?</li></ul>', ['%error' => $e->getMessage()]));
+        return FALSE;
+      }
+    }
+    return TRUE;
+  }
+
+  /**
+   * Check encoding is UTF8.
+   */
+  protected function checkEncoding() {
+    try {
+      if (Database::getConnection()->query('SHOW server_encoding')->fetchField() == 'UTF8') {
+        $this->pass(t('Database is encoded in UTF-8'));
+      }
+      else {
+        $this->fail(t('The %driver database must use %encoding encoding to work with Drupal. Recreate the database with %encoding encoding. See <a href="INSTALL.pgsql.txt">INSTALL.pgsql.txt</a> for more details.', [
+          '%encoding' => 'UTF8',
+          '%driver' => $this->name(),
+        ]));
+      }
+    }
+    catch (\Exception $e) {
+      $this->fail(t('Drupal could not determine the encoding of the database was set to UTF-8'));
+    }
+  }
+
+  /**
+   * Check Binary Output.
+   *
+   * Unserializing does not work on Postgresql 9 when bytea_output is 'hex'.
+   */
+  public function checkBinaryOutput() {
+    $database_connection = Database::getConnection();
+    if (!$this->checkBinaryOutputSuccess()) {
+      // First try to alter the database. If it fails, raise an error telling
+      // the user to do it themselves.
+      $connection_options = $database_connection->getConnectionOptions();
+      // It is safe to include the database name directly here, because this
+      // code is only called when a connection to the database is already
+      // established, thus the database name is guaranteed to be a correct
+      // value.
+      $query = "ALTER DATABASE \"{$connection_options['database']}\" SET bytea_output = 'escape';";
+      try {
+        $database_connection->query($query);
+      }
+      catch (\Exception $e) {
+        // Ignore possible errors when the user doesn't have the necessary
+        // privileges to ALTER the database.
+      }
+
+      // Close the database connection so that the configuration parameter
+      // is applied to the current connection.
+      Database::closeConnection();
+
+      // Recheck, if it fails, finally just rely on the end user to do the
+      // right thing.
+      if (!$this->checkBinaryOutputSuccess()) {
+        $replacements = [
+          '%setting' => 'bytea_output',
+          '%current_value' => 'hex',
+          '%needed_value' => 'escape',
+          '@query' => $query,
+        ];
+        $this->fail(t("The %setting setting is currently set to '%current_value', but needs to be '%needed_value'. Change this by running the following query: <code>@query</code>", $replacements));
+      }
+    }
+  }
+
+  /**
+   * Verify that a binary data roundtrip returns the original string.
+   */
+  protected function checkBinaryOutputSuccess() {
+    $bytea_output = Database::getConnection()->query("SHOW bytea_output")->fetchField();
+    return ($bytea_output == 'escape');
+  }
+
+  /**
+   * Ensures standard_conforming_strings setting is 'on'.
+   *
+   * When standard_conforming_strings setting is 'on' string literals ('...')
+   * treat backslashes literally, as specified in the SQL standard. This allows
+   * Drupal to convert between bytea, text and varchar columns.
+   */
+  public function checkStandardConformingStrings() {
+    $database_connection = Database::getConnection();
+    if (!$this->checkStandardConformingStringsSuccess()) {
+      // First try to alter the database. If it fails, raise an error telling
+      // the user to do it themselves.
+      $connection_options = $database_connection->getConnectionOptions();
+      // It is safe to include the database name directly here, because this
+      // code is only called when a connection to the database is already
+      // established, thus the database name is guaranteed to be a correct
+      // value.
+      $query = "ALTER DATABASE \"" . $connection_options['database'] . "\" SET standard_conforming_strings = 'on';";
+      try {
+        $database_connection->query($query);
+      }
+      catch (\Exception $e) {
+        // Ignore possible errors when the user doesn't have the necessary
+        // privileges to ALTER the database.
+      }
+
+      // Close the database connection so that the configuration parameter
+      // is applied to the current connection.
+      Database::closeConnection();
+
+      // Recheck, if it fails, finally just rely on the end user to do the
+      // right thing.
+      if (!$this->checkStandardConformingStringsSuccess()) {
+        $replacements = [
+          '%setting' => 'standard_conforming_strings',
+          '%current_value' => 'off',
+          '%needed_value' => 'on',
+          '@query' => $query,
+        ];
+        $this->fail(t("The %setting setting is currently set to '%current_value', but needs to be '%needed_value'. Change this by running the following query: <code>@query</code>", $replacements));
+      }
+    }
+  }
+
+  /**
+   * Verifies the standard_conforming_strings setting.
+   */
+  protected function checkStandardConformingStringsSuccess() {
+    $standard_conforming_strings = Database::getConnection()->query("SHOW standard_conforming_strings")->fetchField();
+    return ($standard_conforming_strings == 'on');
+  }
+
+  /**
+   * Generic function to check postgresql extensions.
+   */
+  public function checkExtensions() {
+    $connection = Database::getConnection();
+    try {
+      // Enable pg_trgm for PostgreSQL 13 or higher.
+      // @todo Remove this if-statement in D11 when the minimum required version
+      // for PostgreSQL becomes 13 or higher. https://www.drupal.org/i/3357409
+      if (version_compare($connection->version(), '13.0', '>=')) {
+        $connection->query('CREATE EXTENSION IF NOT EXISTS pg_trgm');
+      }
+
+      if ($connection->schema()->extensionExists('pg_trgm')) {
+        $this->pass(t('PostgreSQL has the pg_trgm extension enabled.'));
+      }
+      else {
+        $this->fail(t('The <a href=":pg_trgm">pg_trgm</a> PostgreSQL extension is not present. The extension is required by Drupal 10 to improve performance when using PostgreSQL. See <a href=":requirements">Drupal database server requirements</a> for more information.', [
+          ':pg_trgm' => 'https://www.postgresql.org/docs/current/pgtrgm.html',
+          ':requirements' => 'https://www.drupal.org/docs/system-requirements/database-server-requirements',
+        ]));
+      }
+    }
+    catch (\Exception $e) {
+      $this->fail(t('Drupal could not check for the pg_trgm extension: @error.', ['@error' => $e->getMessage()]));
+    }
+  }
+
+  /**
+   * Make PostgreSQL Drupal friendly.
+   */
+  public function initializeDatabase() {
+    // We create some functions using global names instead of prefixing them
+    // like we do with table names. This is so that we don't double up if more
+    // than one instance of Drupal is running on a single database. We therefore
+    // avoid trying to create them again in that case.
+    // At the same time checking for the existence of the function fixes
+    // concurrency issues, when both try to update at the same time.
+    try {
+      $connection = Database::getConnection();
+      // When testing, two installs might try to run the CREATE FUNCTION queries
+      // at the same time. Do not let that happen.
+      $connection->query('SELECT pg_advisory_lock(1)');
+      // Don't use {} around pg_proc table.
+      if (!$connection->query("SELECT COUNT(*) FROM pg_proc WHERE proname = 'rand'")->fetchField()) {
+        $connection->query('CREATE OR REPLACE FUNCTION "rand"() RETURNS float AS
+          \'SELECT random();\'
+          LANGUAGE \'sql\'',
+          [],
+          ['allow_delimiter_in_query' => TRUE]
+        );
+      }
+
+      if (!$connection->query("SELECT COUNT(*) FROM pg_proc WHERE proname = 'substring_index'")->fetchField()) {
+        $connection->query('CREATE OR REPLACE FUNCTION "substring_index"(text, text, integer) RETURNS text AS
+          \'SELECT array_to_string((string_to_array($1, $2)) [1:$3], $2);\'
+          LANGUAGE \'sql\'',
+          [],
+          ['allow_delimiter_in_query' => TRUE, 'allow_square_brackets' => TRUE]
+        );
+      }
+      $connection->query('SELECT pg_advisory_unlock(1)');
+
+      $this->pass(t('PostgreSQL has initialized itself.'));
+    }
+    catch (\Exception $e) {
+      $this->fail(t('Drupal could not be correctly setup with the existing database due to the following error: @error.', ['@error' => $e->getMessage()]));
+    }
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function getFormOptions(array $database) {
+    $form = parent::getFormOptions($database);
+    if (empty($form['advanced_options']['port']['#default_value'])) {
+      $form['advanced_options']['port']['#default_value'] = '5432';
+    }
+    return $form;
+  }
+
+}

core/modules/pgsql/src/Driver/Database/pgsql/Merge.php

@@ -0,0 +1,22 @@
+<?php
+
+namespace Drupal\pgsql\Driver\Database\pgsql;
+
+use Drupal\Core\Database\Query\Merge as QueryMerge;
+
+/**
+ * PostgreSQL implementation of \Drupal\Core\Database\Query\Merge.
+ */
+class Merge extends QueryMerge {
+
+  /**
+   * {@inheritdoc}
+   */
+  public function __construct(Connection $connection, string $table, array $options = []) {
+    // @todo Remove the __construct in Drupal 11.
+    // @see https://www.drupal.org/project/drupal/issues/3256524
+    parent::__construct($connection, $table, $options);
+    unset($this->queryOptions['return']);
+  }
+
+}

core/modules/pgsql/src/Driver/Database/pgsql/Select.php

@@ -0,0 +1,170 @@
+<?php
+
+namespace Drupal\pgsql\Driver\Database\pgsql;
+
+use Drupal\Core\Database\Query\Select as QuerySelect;
+
+/**
+ * @addtogroup database
+ * @{
+ */
+
+/**
+ * PostgreSQL implementation of \Drupal\Core\Database\Query\Select.
+ */
+class Select extends QuerySelect {
+
+  /**
+   * {@inheritdoc}
+   */
+  public function __construct(Connection $connection, $table, $alias = NULL, array $options = []) {
+    // @todo Remove the __construct in Drupal 11.
+    // @see https://www.drupal.org/project/drupal/issues/3256524
+    parent::__construct($connection, $table, $alias, $options);
+    unset($this->queryOptions['return']);
+  }
+
+  public function orderRandom() {
+    $alias = $this->addExpression('RANDOM()', 'random_field');
+    $this->orderBy($alias);
+    return $this;
+  }
+
+  /**
+   * Overrides SelectQuery::orderBy().
+   *
+   * PostgreSQL adheres strictly to the SQL-92 standard and requires that when
+   * using DISTINCT or GROUP BY conditions, fields and expressions that are
+   * ordered on also need to be selected. This is a best effort implementation
+   * to handle the cases that can be automated by adding the field if it is not
+   * yet selected.
+   *
+   * @code
+   *   $query = \Drupal::database()->select('example', 'e');
+   *   $query->join('example_revision', 'er', '[e].[vid] = [er].[vid]');
+   *   $query
+   *     ->distinct()
+   *     ->fields('e')
+   *     ->orderBy('timestamp');
+   * @endcode
+   *
+   * In this query, it is not possible (without relying on the schema) to know
+   * whether timestamp belongs to example_revision and needs to be added or
+   * belongs to node and is already selected. Queries like this will need to be
+   * corrected in the original query by adding an explicit call to
+   * SelectQuery::addField() or SelectQuery::fields().
+   *
+   * Since this has a small performance impact, both by the additional
+   * processing in this function and in the database that needs to return the
+   * additional fields, this is done as an override instead of implementing it
+   * directly in SelectQuery::orderBy().
+   */
+  public function orderBy($field, $direction = 'ASC') {
+    // Only allow ASC and DESC, default to ASC.
+    // Emulate MySQL default behavior to sort NULL values first for ascending,
+    // and last for descending.
+    // @see http://www.postgresql.org/docs/9.3/static/queries-order.html
+    $direction = strtoupper($direction) == 'DESC' ? 'DESC NULLS LAST' : 'ASC NULLS FIRST';
+    $this->order[$field] = $direction;
+
+    if ($this->hasTag('entity_query')) {
+      return $this;
+    }
+
+    // If there is a table alias specified, split it up.
+    if (str_contains($field, '.')) {
+      [$table, $table_field] = explode('.', $field);
+    }
+    // Figure out if the field has already been added.
+    foreach ($this->fields as $existing_field) {
+      if (!empty($table)) {
+        // If table alias is given, check if field and table exists.
+        if ($existing_field['table'] == $table && $existing_field['field'] == $table_field) {
+          return $this;
+        }
+      }
+      else {
+        // If there is no table, simply check if the field exists as a field or
+        // an aliased field.
+        if ($existing_field['alias'] == $field) {
+          return $this;
+        }
+      }
+    }
+
+    // Also check expression aliases.
+    foreach ($this->expressions as $expression) {
+      if ($expression['alias'] == $this->connection->escapeAlias($field)) {
+        return $this;
+      }
+    }
+
+    // If a table loads all fields, it can not be added again. It would
+    // result in an ambiguous alias error because that field would be loaded
+    // twice: Once through table_alias.* and once directly. If the field
+    // actually belongs to a different table, it must be added manually.
+    foreach ($this->tables as $table) {
+      if (!empty($table['all_fields'])) {
+        return $this;
+      }
+    }
+
+    // If $field contains characters which are not allowed in a field name
+    // it is considered an expression, these can't be handled automatically
+    // either.
+    if ($this->connection->escapeField($field) != $field) {
+      return $this;
+    }
+
+    // This is a case that can be handled automatically, add the field.
+    $this->addField(NULL, $field);
+    return $this;
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function addExpression($expression, $alias = NULL, $arguments = []) {
+    if (empty($alias)) {
+      $alias = 'expression';
+    }
+
+    // This implements counting in the same manner as the parent method.
+    $alias_candidate = $alias;
+    $count = 2;
+    while (!empty($this->expressions[$alias_candidate])) {
+      $alias_candidate = $alias . '_' . $count++;
+    }
+    $alias = $alias_candidate;
+
+    $this->expressions[$alias] = [
+      'expression' => $expression,
+      'alias' => $this->connection->escapeAlias($alias_candidate),
+      'arguments' => $arguments,
+    ];
+
+    return $alias;
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function execute() {
+    $this->connection->addSavepoint();
+    try {
+      $result = parent::execute();
+    }
+    catch (\Exception $e) {
+      $this->connection->rollbackSavepoint();
+      throw $e;
+    }
+    $this->connection->releaseSavepoint();
+
+    return $result;
+  }
+
+}
+
+/**
+ * @} End of "addtogroup database".
+ */

core/modules/pgsql/src/Driver/Database/pgsql/TransactionManager.php

@@ -0,0 +1,46 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Drupal\pgsql\Driver\Database\pgsql;
+
+use Drupal\Core\Database\Transaction\ClientConnectionTransactionState;
+use Drupal\Core\Database\Transaction\TransactionManagerBase;
+
+/**
+ * PostgreSql implementation of TransactionManagerInterface.
+ */
+class TransactionManager extends TransactionManagerBase {
+
+  /**
+   * {@inheritdoc}
+   */
+  protected function beginClientTransaction(): bool {
+    return $this->connection->getClientConnection()->beginTransaction();
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  protected function rollbackClientTransaction(): bool {
+    $clientRollback = $this->connection->getClientConnection()->rollBack();
+    $this->setConnectionTransactionState($clientRollback ?
+      ClientConnectionTransactionState::RolledBack :
+      ClientConnectionTransactionState::RollbackFailed
+    );
+    return $clientRollback;
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  protected function commitClientTransaction(): bool {
+    $clientCommit = $this->connection->getClientConnection()->commit();
+    $this->setConnectionTransactionState($clientCommit ?
+      ClientConnectionTransactionState::Committed :
+      ClientConnectionTransactionState::CommitFailed
+    );
+    return $clientCommit;
+  }
+
+}

core/modules/pgsql/src/Driver/Database/pgsql/Truncate.php

@@ -0,0 +1,39 @@
+<?php
+
+namespace Drupal\pgsql\Driver\Database\pgsql;
+
+use Drupal\Core\Database\Query\Truncate as QueryTruncate;
+
+/**
+ * PostgreSQL implementation of \Drupal\Core\Database\Query\Truncate.
+ */
+class Truncate extends QueryTruncate {
+
+  /**
+   * {@inheritdoc}
+   */
+  public function __construct(Connection $connection, string $table, array $options = []) {
+    // @todo Remove the __construct in Drupal 11.
+    // @see https://www.drupal.org/project/drupal/issues/3256524
+    parent::__construct($connection, $table, $options);
+    unset($this->queryOptions['return']);
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function execute() {
+    $this->connection->addSavepoint();
+    try {
+      $result = parent::execute();
+    }
+    catch (\Exception $e) {
+      $this->connection->rollbackSavepoint();
+      throw $e;
+    }
+    $this->connection->releaseSavepoint();
+
+    return $result;
+  }
+
+}

core/modules/pgsql/src/Driver/Database/pgsql/Update.php

@@ -0,0 +1,93 @@
+<?php
+
+namespace Drupal\pgsql\Driver\Database\pgsql;
+
+use Drupal\Core\Database\Query\Update as QueryUpdate;
+use Drupal\Core\Database\Query\SelectInterface;
+
+/**
+ * PostgreSQL implementation of \Drupal\Core\Database\Query\Update.
+ */
+class Update extends QueryUpdate {
+
+  /**
+   * {@inheritdoc}
+   */
+  public function __construct(Connection $connection, string $table, array $options = []) {
+    // @todo Remove the __construct in Drupal 11.
+    // @see https://www.drupal.org/project/drupal/issues/3256524
+    parent::__construct($connection, $table, $options);
+    unset($this->queryOptions['return']);
+  }
+
+  public function execute() {
+    $max_placeholder = 0;
+    $blobs = [];
+    $blob_count = 0;
+
+    // Because we filter $fields the same way here and in __toString(), the
+    // placeholders will all match up properly.
+    $stmt = $this->connection->prepareStatement((string) $this, $this->queryOptions, TRUE);
+
+    // Fetch the list of blobs and sequences used on that table.
+    $table_information = $this->connection->schema()->queryTableInformation($this->table);
+
+    // Expressions take priority over literal fields, so we process those first
+    // and remove any literal fields that conflict.
+    $fields = $this->fields;
+    foreach ($this->expressionFields as $field => $data) {
+      if (!empty($data['arguments'])) {
+        foreach ($data['arguments'] as $placeholder => $argument) {
+          // 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->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->getClientStatement()->bindParam($placeholder, $select_query_arguments[$placeholder]);
+        }
+      }
+      unset($fields[$field]);
+    }
+
+    foreach ($fields as $field => $value) {
+      $placeholder = ':db_update_placeholder_' . ($max_placeholder++);
+
+      if (isset($table_information->blob_fields[$field]) && $value !== NULL) {
+        $blobs[$blob_count] = fopen('php://memory', 'a');
+        fwrite($blobs[$blob_count], $value);
+        rewind($blobs[$blob_count]);
+        $stmt->getClientStatement()->bindParam($placeholder, $blobs[$blob_count], \PDO::PARAM_LOB);
+        ++$blob_count;
+      }
+      else {
+        $stmt->getClientStatement()->bindParam($placeholder, $fields[$field]);
+      }
+    }
+
+    if (count($this->condition)) {
+      $this->condition->compile($this->connection, $this);
+
+      $arguments = $this->condition->arguments();
+      foreach ($arguments as $placeholder => $value) {
+        $stmt->getClientStatement()->bindParam($placeholder, $arguments[$placeholder]);
+      }
+    }
+
+    $this->connection->addSavepoint();
+    try {
+      $stmt->execute(NULL, $this->queryOptions);
+      $this->connection->releaseSavepoint();
+      return $stmt->rowCount();
+    }
+    catch (\Exception $e) {
+      $this->connection->rollbackSavepoint();
+      $this->connection->exceptionHandler()->handleExecutionException($e, $stmt, [], $this->queryOptions);
+    }
+  }
+
+}

core/modules/pgsql/src/Driver/Database/pgsql/Upsert.php

@@ -0,0 +1,136 @@
+<?php
+
+namespace Drupal\pgsql\Driver\Database\pgsql;
+
+use Drupal\Core\Database\Query\Upsert as QueryUpsert;
+
+// cSpell:ignore nextval setval
+
+/**
+ * PostgreSQL implementation of \Drupal\Core\Database\Query\Upsert.
+ */
+class Upsert extends QueryUpsert {
+
+  /**
+   * {@inheritdoc}
+   */
+  public function __construct(Connection $connection, string $table, array $options = []) {
+    // @todo Remove the __construct in Drupal 11.
+    // @see https://www.drupal.org/project/drupal/issues/3256524
+    parent::__construct($connection, $table, $options);
+    unset($this->queryOptions['return']);
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function execute() {
+    if (!$this->preExecute()) {
+      return NULL;
+    }
+
+    $stmt = $this->connection->prepareStatement((string) $this, $this->queryOptions, TRUE);
+
+    // Fetch the list of blobs and sequences used on that table.
+    $table_information = $this->connection->schema()->queryTableInformation($this->table);
+
+    $max_placeholder = 0;
+    $blobs = [];
+    $blob_count = 0;
+    foreach ($this->insertValues as $insert_values) {
+      foreach ($this->insertFields as $idx => $field) {
+        if (isset($table_information->blob_fields[$field]) && $insert_values[$idx] !== NULL) {
+          $blobs[$blob_count] = fopen('php://memory', 'a');
+          fwrite($blobs[$blob_count], $insert_values[$idx]);
+          rewind($blobs[$blob_count]);
+
+          $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->getClientStatement()->bindParam(':db_insert_placeholder_' . $max_placeholder++, $insert_values[$idx]);
+        }
+      }
+      // Check if values for a serial field has been passed.
+      if (!empty($table_information->serial_fields)) {
+        foreach ($table_information->serial_fields as $index => $serial_field) {
+          $serial_key = array_search($serial_field, $this->insertFields);
+          if ($serial_key !== FALSE) {
+            $serial_value = $insert_values[$serial_key];
+
+            // Sequences must be greater than or equal to 1.
+            if ($serial_value === NULL || !$serial_value) {
+              $serial_value = 1;
+            }
+            // Set the sequence to the bigger value of either the passed
+            // value or the max value of the column. It can happen that another
+            // thread calls nextval() which could lead to a serial number being
+            // used twice. However, trying to insert a value into a serial
+            // column should only be done in very rare cases and is not thread
+            // safe by definition.
+            $this->connection->query("SELECT setval('" . $table_information->sequences[$index] . "', GREATEST(MAX(" . $serial_field . "), :serial_value)) FROM {" . $this->table . "}", [':serial_value' => (int) $serial_value]);
+          }
+        }
+      }
+    }
+
+    $options = $this->queryOptions;
+    if (!empty($table_information->sequences)) {
+      $options['sequence_name'] = $table_information->sequences[0];
+    }
+
+    // Re-initialize the values array so that we can re-use this query.
+    $this->insertValues = [];
+
+    // Create a savepoint so we can rollback a failed query. This is so we can
+    // mimic MySQL and SQLite transactions which don't fail if a single query
+    // fails. This is important for tables that are created on demand. For
+    // example, \Drupal\Core\Cache\DatabaseBackend.
+    $this->connection->addSavepoint();
+    try {
+      $stmt->execute(NULL, $options);
+      $this->connection->releaseSavepoint();
+      return $stmt->rowCount();
+    }
+    catch (\Exception $e) {
+      $this->connection->rollbackSavepoint();
+      $this->connection->exceptionHandler()->handleExecutionException($e, $stmt, [], $options);
+    }
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function __toString() {
+    // Create a sanitized comment string to prepend to the query.
+    $comments = $this->connection->makeComment($this->comments);
+
+    // Default fields are always placed first for consistency.
+    $insert_fields = array_merge($this->defaultFields, $this->insertFields);
+    $insert_fields = array_map(function ($field) {
+      return $this->connection->escapeField($field);
+    }, $insert_fields);
+
+    $query = $comments . 'INSERT INTO {' . $this->table . '} (' . implode(', ', $insert_fields) . ') VALUES ';
+
+    $values = $this->getInsertPlaceholderFragment($this->insertValues, $this->defaultFields);
+    $query .= implode(', ', $values);
+
+    // Updating the unique / primary key is not necessary.
+    unset($insert_fields[$this->key]);
+
+    $update = [];
+    foreach ($insert_fields as $field) {
+      // The "excluded." prefix causes the field to refer to the value for field
+      // that would have been inserted had there been no conflict.
+      $update[] = "$field = EXCLUDED.$field";
+    }
+
+    $query .= ' ON CONFLICT (' . $this->connection->escapeField($this->key) . ') DO UPDATE SET ' . implode(', ', $update);
+
+    return $query;
+  }
+
+}

core/modules/pgsql/src/Update10101.php

@@ -0,0 +1,234 @@
+<?php
+
+namespace Drupal\pgsql;
+
+use Drupal\Core\Database\Connection;
+use Drupal\Core\Database\DatabaseExceptionWrapper;
+use Drupal\Core\DependencyInjection\ContainerInjectionInterface;
+use Drupal\Core\Entity\EntityLastInstalledSchemaRepositoryInterface;
+use Drupal\Core\Entity\EntityTypeManagerInterface;
+use Drupal\Core\Entity\Sql\SqlContentEntityStorage;
+use Drupal\Core\Extension\ModuleExtensionList;
+use Drupal\Core\Extension\ModuleHandlerInterface;
+use Drupal\Core\StringTranslation\PluralTranslatableMarkup;
+use Symfony\Component\DependencyInjection\ContainerInterface;
+
+// cSpell:ignore relkind objid regclass
+
+/**
+ * An update class for sequence ownership.
+ * @see https://www.drupal.org/i/3028706
+ *
+ * @internal
+ */
+class Update10101 implements ContainerInjectionInterface {
+
+  /**
+   * Sequence owner update constructor.
+   *
+   * @param \Drupal\Core\Entity\EntityTypeManagerInterface $entityTypeManager
+   *   The entity type manager.
+   * @param \Drupal\Core\Entity\EntityLastInstalledSchemaRepositoryInterface $entityLastInstalledSchemaRepository
+   *   The last installed schema repository service.
+   * @param \Drupal\Core\Database\Connection $connection
+   *   The database connection.
+   * @param \Drupal\Core\Extension\ModuleExtensionList $moduleExtensionList
+   *   The module extension list.
+   * @param \Drupal\Core\Extension\ModuleHandlerInterface $moduleHandler
+   *   The module handler service.
+   */
+  public function __construct(
+    protected EntityTypeManagerInterface $entityTypeManager,
+    protected EntityLastInstalledSchemaRepositoryInterface $entityLastInstalledSchemaRepository,
+    protected Connection $connection,
+    protected ModuleExtensionList $moduleExtensionList,
+    protected ModuleHandlerInterface $moduleHandler,
+  ) {
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public static function create(ContainerInterface $container) {
+    return new static(
+      $container->get('entity_type.manager'),
+      $container->get('entity.last_installed_schema.repository'),
+      $container->get('database'),
+      $container->get('extension.list.module'),
+      $container->get('module_handler')
+    );
+  }
+
+  /**
+   * Update *all* existing sequences to include the owner tables.
+   *
+   * @param array $sandbox
+   *   Stores information for batch updates.
+   *
+   * @return \Drupal\Core\StringTranslation\PluralTranslatableMarkup|null
+   *   Returns the amount of orphaned sequences fixed.
+   */
+  public function update(array &$sandbox): ?PluralTranslatableMarkup {
+    if ($this->connection->databaseType() !== 'pgsql') {
+      // This database update is a no-op for all other core database drivers.
+      $sandbox['#finished'] = 1;
+      return NULL;
+    }
+
+    if (!isset($sandbox['progress'])) {
+      $sandbox['fixed'] = 0;
+      $sandbox['progress'] = 0;
+      $sandbox['tables'] = [];
+
+      // Discovers all tables defined with hook_schema().
+      // @todo We need to add logic to do the same for on-demand tables. See
+      //   https://www.drupal.org/i/3358777
+      $modules = $this->moduleExtensionList->getList();
+      foreach ($modules as $extension) {
+        $module = $extension->getName();
+        $this->moduleHandler->loadInclude($module, 'install');
+        $schema = $this->moduleHandler->invoke($module, 'schema');
+        if (!empty($schema)) {
+          foreach ($schema as $table_name => $table_info) {
+            foreach ($table_info['fields'] as $column_name => $column_info) {
+              if (str_starts_with($column_info['type'], 'serial')) {
+                $sandbox['tables'][] = [
+                  'table' => $table_name,
+                  'column' => $column_name,
+                ];
+              }
+            }
+          }
+        }
+      }
+
+      // Discovers all content entity types with integer entity keys that are
+      // most likely serial columns.
+      $entity_types = $this->entityTypeManager->getDefinitions();
+      /** @var \Drupal\Core\Entity\EntityTypeInterface $entity_type */
+      foreach ($entity_types as $entity_type) {
+        $storage_class = $entity_type->getStorageClass();
+        if (is_subclass_of($storage_class, SqlContentEntityStorage::class)) {
+          $id_key = $entity_type->getKey('id');
+          $revision_key = $entity_type->getKey('revision');
+
+          $original_storage_definitions = $this->entityLastInstalledSchemaRepository->getLastInstalledFieldStorageDefinitions($entity_type->id());
+          if ($original_storage_definitions[$id_key]->getType() === 'integer') {
+            $sandbox['tables'][] = [
+              'table' => $entity_type->getBaseTable(),
+              'column' => $id_key,
+            ];
+          }
+
+          if ($entity_type->isRevisionable() &&
+            $original_storage_definitions[$revision_key]->getType() === 'integer') {
+            $sandbox['tables'][] = [
+              'table' => $entity_type->getRevisionTable(),
+              'column' => $revision_key,
+            ];
+          }
+        }
+      }
+      $sandbox['max'] = count($sandbox['tables']);
+    }
+    else {
+      // Adds ownership of orphan sequences to tables.
+      $to_process = array_slice($sandbox['tables'], $sandbox['progress'], 50);
+
+      // Ensures that a sequence is not owned first, then ensures that the a
+      // sequence exists at all before trying to alter it.
+      foreach ($to_process as $table_info) {
+        if ($this->connection->schema()->tableExists($table_info['table'])) {
+          $owned = (bool) $this->getSequenceName($table_info['table'], $table_info['column']);
+
+          if (!$owned) {
+            $sequence_name = $this->connection
+              ->makeSequenceName($table_info['table'], $table_info['column']);
+            $exists = $this->sequenceExists($sequence_name);
+            if ($exists) {
+              $transaction = $this->connection->startTransaction($sequence_name);
+              try {
+                $this->updateSequenceOwnership($sequence_name, $table_info['table'], $table_info['column']);
+
+                $sandbox['fixed']++;
+              }
+              catch (DatabaseExceptionWrapper $e) {
+                $transaction->rollBack();
+              }
+            }
+          }
+        }
+        $sandbox['progress']++;
+      }
+    }
+
+    if ($sandbox['max'] && $sandbox['progress'] < $sandbox['max']) {
+      $sandbox['#finished'] = $sandbox['progress'] / $sandbox['max'];
+      return NULL;
+    }
+    else {
+      $sandbox['#finished'] = 1;
+      return new PluralTranslatableMarkup(
+        $sandbox['fixed'],
+        '1 orphaned sequence fixed.',
+        '@count orphaned sequences fixed'
+      );
+    }
+  }
+
+  /**
+   * Alters the ownership of a sequence.
+   *
+   * This is used for updating orphaned sequences.
+   *
+   * @param string $sequence_name
+   *   The appropriate sequence name for a given table and serial field.
+   * @param string $table
+   *   The unquoted or prefixed table name.
+   * @param string $column
+   *   The column name for the sequence.
+   *
+   * @see https://www.drupal.org/i/3028706
+   */
+  private function updateSequenceOwnership(string $sequence_name, string $table, string $column): void {
+    $this->connection->query('ALTER SEQUENCE IF EXISTS ' . $sequence_name . ' OWNED BY {' . $table . '}.[' . $column . ']');
+  }
+
+  /**
+   * Retrieves a sequence name that is owned by the table and column.
+   *
+   * @param string $table
+   *   A table name that is not prefixed or quoted.
+   * @param string $column
+   *   The column name.
+   *
+   * @return string|null
+   *   The name of the sequence or NULL if it does not exist.
+   */
+  public function getSequenceName(string $table, string $column): ?string {
+    return $this->connection
+      ->query("SELECT pg_get_serial_sequence(:table, :column)", [
+        ':table' => $this->connection->getPrefix() . $table,
+        ':column' => $column,
+      ])
+      ->fetchField();
+  }
+
+  /**
+   * Checks if a sequence exists.
+   *
+   * @param string $name
+   *   The fully-qualified sequence name.
+   *
+   * @return bool
+   *   TRUE if the sequence exists by the name.
+   *
+   * @see \Drupal\pgsql\Driver\Database\pgsql\Connection::makeSequenceName()
+   */
+  private function sequenceExists(string $name): bool {
+    return (bool) \Drupal::database()
+      ->query("SELECT c.relname FROM pg_class as c WHERE c.relkind = 'S' AND c.relname = :name", [':name' => $name])
+      ->fetchField();
+  }
+
+}