diff --git a/core/lib/Drupal/Core/Database/Connection.php b/core/lib/Drupal/Core/Database/Connection.php
index 5b68cd9a8d..81a0363612 100644
--- a/core/lib/Drupal/Core/Database/Connection.php
+++ b/core/lib/Drupal/Core/Database/Connection.php
@@ -2,6 +2,8 @@
 
 namespace Drupal\Core\Database;
 
+use Drupal\Core\Database\Query\Condition;
+
 /**
  * Base Database API class.
  *
@@ -865,6 +867,11 @@ public function getDriverClass($class) {
       }
       $driver_class = $this->connectionOptions['namespace'] . '\\' . $class;
       $this->driverClasses[$class] = class_exists($driver_class) ? $driver_class : $class;
+      if ($this->driverClasses[$class] === 'Condition') {
+        // @todo Deprecate the fallback for contrib and custom drivers in 9.1.x
+        // in https://www.drupal.org/project/drupal/issues/3120036.
+        $this->driverClasses[$class] = Condition::class;
+      }
     }
     return $this->driverClasses[$class];
   }
@@ -1026,6 +1033,22 @@ public function schema() {
     return $this->schema;
   }
 
+  /**
+   * Prepares and returns a CONDITION query object.
+   *
+   * @param string $conjunction
+   *   The operator to use to combine conditions: 'AND' or 'OR'.
+   *
+   * @return \Drupal\Core\Database\Query\Condition
+   *   A new Condition query object.
+   *
+   * @see \Drupal\Core\Database\Query\Condition
+   */
+  public function condition($conjunction) {
+    $class = $this->getDriverClass('Condition');
+    return new $class($conjunction);
+  }
+
   /**
    * Escapes a database name string.
    *
diff --git a/core/lib/Drupal/Core/Database/Driver/mysql/Schema.php b/core/lib/Drupal/Core/Database/Driver/mysql/Schema.php
index 1c7670a5a4..d909f7c50b 100644
--- a/core/lib/Drupal/Core/Database/Driver/mysql/Schema.php
+++ b/core/lib/Drupal/Core/Database/Driver/mysql/Schema.php
@@ -2,7 +2,6 @@
 
 namespace Drupal\Core\Database\Driver\mysql;
 
-use Drupal\Core\Database\Query\Condition;
 use Drupal\Core\Database\SchemaException;
 use Drupal\Core\Database\SchemaObjectExistsException;
 use Drupal\Core\Database\SchemaObjectDoesNotExistException;
@@ -75,7 +74,7 @@ protected function getPrefixInfo($table = 'default', $add_prefix = TRUE) {
   protected function buildTableNameCondition($table_name, $operator = '=', $add_prefix = TRUE) {
     $table_info = $this->getPrefixInfo($table_name, $add_prefix);
 
-    $condition = new Condition('AND');
+    $condition = $this->connection->condition('AND');
     $condition->condition('table_schema', $table_info['database']);
     $condition->condition('table_name', $table_info['table'], $operator);
     return $condition;
diff --git a/core/lib/Drupal/Core/Database/Query/Condition.php b/core/lib/Drupal/Core/Database/Query/Condition.php
index e7038e23c9..2515bbfdc8 100644
--- a/core/lib/Drupal/Core/Database/Query/Condition.php
+++ b/core/lib/Drupal/Core/Database/Query/Condition.php
@@ -399,7 +399,7 @@ protected function mapConditionOperator($operator) {
    * {@inheritdoc}
    */
   public function conditionGroupFactory($conjunction = 'AND') {
-    return new Condition($conjunction);
+    return new static($conjunction);
   }
 
   /**
diff --git a/core/lib/Drupal/Core/Database/Query/Delete.php b/core/lib/Drupal/Core/Database/Query/Delete.php
index def88385e0..658b8342cc 100644
--- a/core/lib/Drupal/Core/Database/Query/Delete.php
+++ b/core/lib/Drupal/Core/Database/Query/Delete.php
@@ -36,7 +36,7 @@ public function __construct(Connection $connection, $table, array $options = [])
     parent::__construct($connection, $options);
     $this->table = $table;
 
-    $this->condition = new Condition('AND');
+    $this->condition = $this->connection->condition('AND');
   }
 
   /**
diff --git a/core/lib/Drupal/Core/Database/Query/Merge.php b/core/lib/Drupal/Core/Database/Query/Merge.php
index 6d89532472..5a9bc3db89 100644
--- a/core/lib/Drupal/Core/Database/Query/Merge.php
+++ b/core/lib/Drupal/Core/Database/Query/Merge.php
@@ -138,7 +138,7 @@ public function __construct(Connection $connection, $table, array $options = [])
     parent::__construct($connection, $options);
     $this->table = $table;
     $this->conditionTable = $table;
-    $this->condition = new Condition('AND');
+    $this->condition = $this->connection->condition('AND');
   }
 
   /**
diff --git a/core/lib/Drupal/Core/Database/Query/QueryConditionTrait.php b/core/lib/Drupal/Core/Database/Query/QueryConditionTrait.php
index 9053a17772..83be429fa5 100644
--- a/core/lib/Drupal/Core/Database/Query/QueryConditionTrait.php
+++ b/core/lib/Drupal/Core/Database/Query/QueryConditionTrait.php
@@ -108,7 +108,7 @@ public function compiled() {
    * {@inheritdoc}
    */
   public function conditionGroupFactory($conjunction = 'AND') {
-    return new Condition($conjunction);
+    return $this->connection->condition($conjunction);
   }
 
   /**
diff --git a/core/lib/Drupal/Core/Database/Query/Select.php b/core/lib/Drupal/Core/Database/Query/Select.php
index fb74d9f1c1..78462d0b63 100644
--- a/core/lib/Drupal/Core/Database/Query/Select.php
+++ b/core/lib/Drupal/Core/Database/Query/Select.php
@@ -134,8 +134,8 @@ public function __construct(Connection $connection, $table, $alias = NULL, $opti
     $options['return'] = Database::RETURN_STATEMENT;
     parent::__construct($connection, $options);
     $conjunction = isset($options['conjunction']) ? $options['conjunction'] : 'AND';
-    $this->condition = new Condition($conjunction);
-    $this->having = new Condition($conjunction);
+    $this->condition = $this->connection->condition($conjunction);
+    $this->having = $this->connection->condition($conjunction);
     $this->addJoin(NULL, $table, $alias);
   }
 
diff --git a/core/lib/Drupal/Core/Database/Query/SelectExtender.php b/core/lib/Drupal/Core/Database/Query/SelectExtender.php
index dc3531df2f..e428c21d86 100644
--- a/core/lib/Drupal/Core/Database/Query/SelectExtender.php
+++ b/core/lib/Drupal/Core/Database/Query/SelectExtender.php
@@ -521,7 +521,7 @@ public function __call($method, $args) {
    * {@inheritdoc}
    */
   public function conditionGroupFactory($conjunction = 'AND') {
-    return new Condition($conjunction);
+    return $this->connection->condition($conjunction);
   }
 
   /**
diff --git a/core/lib/Drupal/Core/Database/Query/Update.php b/core/lib/Drupal/Core/Database/Query/Update.php
index 5de5d682ec..2b08e20d00 100644
--- a/core/lib/Drupal/Core/Database/Query/Update.php
+++ b/core/lib/Drupal/Core/Database/Query/Update.php
@@ -65,7 +65,7 @@ public function __construct(Connection $connection, $table, array $options = [])
     parent::__construct($connection, $options);
     $this->table = $table;
 
-    $this->condition = new Condition('AND');
+    $this->condition = $this->connection->condition('AND');
   }
 
   /**
diff --git a/core/lib/Drupal/Core/Database/Schema.php b/core/lib/Drupal/Core/Database/Schema.php
index 38fde96345..1987177a3e 100644
--- a/core/lib/Drupal/Core/Database/Schema.php
+++ b/core/lib/Drupal/Core/Database/Schema.php
@@ -2,7 +2,6 @@
 
 namespace Drupal\Core\Database;
 
-use Drupal\Core\Database\Query\Condition;
 use Drupal\Core\Database\Query\PlaceholderInterface;
 
 /**
@@ -150,7 +149,7 @@ protected function buildTableNameCondition($table_name, $operator = '=', $add_pr
     // Retrieve the table name and schema
     $table_info = $this->getPrefixInfo($table_name, $add_prefix);
 
-    $condition = new Condition('AND');
+    $condition = $this->connection->condition('AND');
     $condition->condition('table_catalog', $info['database']);
     $condition->condition('table_schema', $table_info['schema']);
     $condition->condition('table_name', $table_info['table'], $operator);
diff --git a/core/modules/system/tests/modules/database_statement_monitoring_test/src/LoggedStatementsTrait.php b/core/modules/system/tests/modules/database_statement_monitoring_test/src/LoggedStatementsTrait.php
index ceb98ea05d..d4f760e093 100644
--- a/core/modules/system/tests/modules/database_statement_monitoring_test/src/LoggedStatementsTrait.php
+++ b/core/modules/system/tests/modules/database_statement_monitoring_test/src/LoggedStatementsTrait.php
@@ -2,6 +2,8 @@
 
 namespace Drupal\database_statement_monitoring_test;
 
+use Drupal\Core\Database\Query\Condition;
+
 /**
  * Trait for Connection classes that can store logged statements.
  */
@@ -48,7 +50,13 @@ public function getDriverClass($class) {
     // based on object, which would break.
     $namespace = (new \ReflectionClass(get_parent_class($this)))->getNamespaceName();
     $driver_class = $namespace . '\\' . $class;
-    return class_exists($driver_class) ? $driver_class : $class;
+    if (class_exists($driver_class)) {
+      return $driver_class;
+    }
+    elseif ($class == 'Condition') {
+      return Condition::class;
+    }
+    return $class;
   }
 
   /**
diff --git a/core/tests/Drupal/KernelTests/Core/Database/ConnectionTest.php b/core/tests/Drupal/KernelTests/Core/Database/ConnectionTest.php
index 7e0a5f89ac..ed2945bb7b 100644
--- a/core/tests/Drupal/KernelTests/Core/Database/ConnectionTest.php
+++ b/core/tests/Drupal/KernelTests/Core/Database/ConnectionTest.php
@@ -12,6 +12,42 @@
  */
 class ConnectionTest extends DatabaseTestBase {
 
+  /**
+   * Test that the method ::condition() returns an Condition object.
+   */
+  public function testCondition() {
+    $db = Database::getConnection('default', 'default');
+    $namespace = (new \ReflectionObject($db))->getNamespaceName() . "\\Condition";
+    if (!class_exists($namespace)) {
+      $namespace = 'Drupal\Core\Database\Query\Condition';
+    }
+    $condition = $db->condition('AND');
+    $this->assertIdentical($namespace, get_class($condition));
+  }
+
+  /**
+   * Test that the core Condition can be overridden.
+   */
+  public function testContribCondition() {
+    $this->mockCondition = $this->getMockBuilder('Drupal\Core\Database\Query\Condition')
+      ->setMockClassName('MockCondition')
+      ->setConstructorArgs([NULL])
+      ->disableOriginalConstructor()
+      ->getMock();
+    $contrib_namespace = 'Drupal\Driver\Database\mock';
+    $mocked_namespace = $contrib_namespace . '\\Condition';
+    class_alias('MockCondition', $mocked_namespace);
+
+    $this->options['namespace'] = $mocked_namespace;
+    $this->options['prefix']['default'] = '';
+
+    $this->mockPdo = $this->createMock('Drupal\Tests\Core\Database\Stub\StubPDO');
+
+    $connection = new Connection($this->mockPdo, $this->options);
+    $condition = $connection->condition();
+    $this->assertIdentical($mocked_namespace, get_class($condition));
+  }
+
   /**
    * Tests that connections return appropriate connection objects.
    */
diff --git a/core/tests/Drupal/Tests/Core/Database/OrderByTest.php b/core/tests/Drupal/Tests/Core/Database/OrderByTest.php
index 9c28331ca8..9ee37adb0c 100644
--- a/core/tests/Drupal/Tests/Core/Database/OrderByTest.php
+++ b/core/tests/Drupal/Tests/Core/Database/OrderByTest.php
@@ -2,6 +2,7 @@
 
 namespace Drupal\Tests\Core\Database;
 
+use Drupal\Core\Database\Query\Condition;
 use Drupal\Core\Database\Query\Select;
 use Drupal\Tests\UnitTestCase;
 
@@ -28,7 +29,11 @@ protected function setUp() {
       // Prevent deprecation message being triggered by
       // Connection::identifierQuote().
       ->setMethods(['identifierQuote'])
+      ->setMethods(['condition'])
       ->getMockForAbstractClass();
+    $connection->expects($this->any())
+      ->method('condition')
+      ->willReturn(new Condition('AND'));
     $this->query = new Select($connection, 'test', NULL);
   }