diff --git a/core/lib/Drupal/Core/Database/Driver/mysql/Schema.php b/core/lib/Drupal/Core/Database/Driver/mysql/Schema.php
index 1911d6a..d9f7088 100644
--- a/core/lib/Drupal/Core/Database/Driver/mysql/Schema.php
+++ b/core/lib/Drupal/Core/Database/Driver/mysql/Schema.php
@@ -608,6 +608,17 @@ public function fieldExists($table, $column) {
     }
   }
 
+  /**
+   * {@inheritdoc}
+   */
+  public function getPrimaryKeyColumns($table) {
+    if (!$this->tableExists($table)) {
+      return FALSE;
+    }
+    $result = $this->connection->query("SHOW KEYS FROM {" . $table . "} WHERE Key_name = 'PRIMARY'")->fetchAllAssoc('Column_name');
+    return array_keys($result);
+  }
+
 }
 
 /**
diff --git a/core/lib/Drupal/Core/Database/Driver/pgsql/Schema.php b/core/lib/Drupal/Core/Database/Driver/pgsql/Schema.php
index 4586d01..37825a0 100644
--- a/core/lib/Drupal/Core/Database/Driver/pgsql/Schema.php
+++ b/core/lib/Drupal/Core/Database/Driver/pgsql/Schema.php
@@ -884,6 +884,17 @@ protected function hashBase64($data) {
     return strtr($hash, ['+' => '_', '/' => '_', '=' => '']);
   }
 
+  /**
+   * {@inheritdoc}
+   */
+  public function getPrimaryKeyColumns($table) {
+    if (!$this->tableExists($table)) {
+      return FALSE;
+    }
+    $result = $this->connection->query("SELECT a.attname, format_type(a.atttypid, a.atttypmod) AS data_type FROM pg_index i JOIN pg_attribute a ON a.attrelid = i.indrelid AND a.attnum = ANY(i.indkey) WHERE i.indrelid = '{" . $table . "}'::regclass AND i.indisprimary")->fetchAllAssoc('attname');
+    return array_keys($result);
+  }
+
 }
 
 /**
diff --git a/core/lib/Drupal/Core/Database/Driver/sqlite/Schema.php b/core/lib/Drupal/Core/Database/Driver/sqlite/Schema.php
index 90c4848..6ef7a45 100644
--- a/core/lib/Drupal/Core/Database/Driver/sqlite/Schema.php
+++ b/core/lib/Drupal/Core/Database/Driver/sqlite/Schema.php
@@ -737,4 +737,15 @@ public function findTables($table_expression) {
     return $tables;
   }
 
+  /**
+   * {@inheritdoc}
+   */
+  public function getPrimaryKeyColumns($table) {
+    if (!$this->tableExists($table)) {
+      return FALSE;
+    }
+    $schema = $this->introspectSchema($table);
+    return $schema['primary key'];
+  }
+
 }
diff --git a/core/lib/Drupal/Core/Database/Schema.php b/core/lib/Drupal/Core/Database/Schema.php
index 80a68f6..a394b87 100644
--- a/core/lib/Drupal/Core/Database/Schema.php
+++ b/core/lib/Drupal/Core/Database/Schema.php
@@ -658,4 +658,16 @@ protected function escapeDefaultValue($value) {
     return is_string($value) ? $this->connection->quote($value) : $value;
   }
 
+  /**
+   * Gets the primary key columns of a table, from the database.
+   *
+   * @param $table
+   *   The name of the table.
+   *
+   * @return string[]|false
+   *   A simple array with the names of the columns composing the table's
+   *   primary key, or FALSE if the table does not exist.
+   */
+  abstract public function getPrimaryKeyColumns($table);
+
 }
diff --git a/core/tests/Drupal/KernelTests/Core/Database/SchemaTest.php b/core/tests/Drupal/KernelTests/Core/Database/SchemaTest.php
index e239098..bcaf057 100644
--- a/core/tests/Drupal/KernelTests/Core/Database/SchemaTest.php
+++ b/core/tests/Drupal/KernelTests/Core/Database/SchemaTest.php
@@ -149,7 +149,8 @@ public function testSchema() {
     db_field_set_default('test_table', 'test_field', 0);
     db_add_field('test_table', 'test_serial', ['type' => 'serial', 'not null' => TRUE], ['primary key' => ['test_serial']]);
 
-    $this->assertPrimaryKeyColumns('test_table', ['test_serial']);
+    // Test the primary key columns.
+    $this->assertSame(['test_serial'], Database::getConnection()->schema()->getPrimaryKeyColumns('test_table'));
 
     $this->assertTrue($this->tryInsert(), 'Insert with a serial succeeded.');
     $max1 = db_query('SELECT MAX(test_serial) FROM {test_table}')->fetchField();
@@ -163,7 +164,8 @@ public function testSchema() {
     // Test adding a new column and form a composite primary key with it.
     db_add_field('test_table', 'test_composite_primary_key', ['type' => 'int', 'not null' => TRUE, 'default' => 0], ['primary key' => ['test_serial', 'test_composite_primary_key']]);
 
-    $this->assertPrimaryKeyColumns('test_table', ['test_serial', 'test_composite_primary_key']);
+    // Test the primary key columns.
+    $this->assertSame(['test_serial', 'test_composite_primary_key'], Database::getConnection()->schema()->getPrimaryKeyColumns('test_table'));
 
     // Test renaming of keys and constraints.
     db_drop_table('test_table');
@@ -826,46 +828,4 @@ public function testFindTables() {
     Database::setActiveConnection('default');
   }
 
-  /**
-   * Tests the primary keys of a table.
-   *
-   * @param string $table_name
-   *   The name of the table to check.
-   * @param array $primary_key
-   *   The expected key column specifier for a table's primary key.
-   */
-  protected function assertPrimaryKeyColumns($table_name, array $primary_key = []) {
-    $db_type = Database::getConnection()->databaseType();
-
-    switch ($db_type) {
-      case 'mysql':
-        $result = Database::getConnection()->query("SHOW KEYS FROM {" . $table_name . "} WHERE Key_name = 'PRIMARY'")->fetchAllAssoc('Column_name');
-        $this->assertSame($primary_key, array_keys($result));
-
-        break;
-      case 'pgsql':
-        $result = Database::getConnection()->query("SELECT a.attname, format_type(a.atttypid, a.atttypmod) AS data_type
-          FROM pg_index i
-          JOIN pg_attribute a ON a.attrelid = i.indrelid AND a.attnum = ANY(i.indkey)
-          WHERE i.indrelid = '{" . $table_name . "}'::regclass AND i.indisprimary")
-          ->fetchAllAssoc('attname');
-        $this->assertSame($primary_key, array_keys($result));
-
-        break;
-      case 'sqlite':
-        // For SQLite we need access to the protected
-        // \Drupal\Core\Database\Driver\sqlite\Schema::introspectSchema() method
-        // because we have no other way of getting the table prefixes needed for
-        // running a straight PRAGMA query.
-        $schema_object = Database::getConnection()->schema();
-        $reflection = new \ReflectionMethod($schema_object, 'introspectSchema');
-        $reflection->setAccessible(TRUE);
-
-        $table_info = $reflection->invoke($schema_object, $table_name);
-        $this->assertSame($primary_key, $table_info['primary key']);
-
-        break;
-    }
-  }
-
 }