diff --git a/exposed-core/api/exposed-core.api b/exposed-core/api/exposed-core.api
index c957c1399e..da691569b1 100644
--- a/exposed-core/api/exposed-core.api
+++ b/exposed-core/api/exposed-core.api
@@ -2968,6 +2968,7 @@ public abstract class org/jetbrains/exposed/sql/statements/api/ExposedDatabaseMe
 	public abstract fun cleanCache ()V
 	public abstract fun columns ([Lorg/jetbrains/exposed/sql/Table;)Ljava/util/Map;
 	public abstract fun existingIndices ([Lorg/jetbrains/exposed/sql/Table;)Ljava/util/Map;
+	public abstract fun existingPrimaryKeys ([Lorg/jetbrains/exposed/sql/Table;)Ljava/util/Map;
 	public abstract fun getCurrentScheme ()Ljava/lang/String;
 	public final fun getDatabase ()Ljava/lang/String;
 	public abstract fun getDatabaseDialectName ()Ljava/lang/String;
@@ -3202,6 +3203,7 @@ public abstract class org/jetbrains/exposed/sql/vendors/DataTypeProvider {
 
 public abstract interface class org/jetbrains/exposed/sql/vendors/DatabaseDialect {
 	public static final field Companion Lorg/jetbrains/exposed/sql/vendors/DatabaseDialect$Companion;
+	public abstract fun addPrimaryKey (Lorg/jetbrains/exposed/sql/Table;Ljava/lang/String;[Lorg/jetbrains/exposed/sql/Column;)Ljava/lang/String;
 	public abstract fun allTablesNames ()Ljava/util/List;
 	public abstract fun catalog (Lorg/jetbrains/exposed/sql/Transaction;)Ljava/lang/String;
 	public abstract fun checkTableMapping (Lorg/jetbrains/exposed/sql/Table;)Z
@@ -3213,6 +3215,7 @@ public abstract interface class org/jetbrains/exposed/sql/vendors/DatabaseDialec
 	public abstract fun dropIndex (Ljava/lang/String;Ljava/lang/String;ZZ)Ljava/lang/String;
 	public abstract fun dropSchema (Lorg/jetbrains/exposed/sql/Schema;Z)Ljava/lang/String;
 	public abstract fun existingIndices ([Lorg/jetbrains/exposed/sql/Table;)Ljava/util/Map;
+	public abstract fun existingPrimaryKeys ([Lorg/jetbrains/exposed/sql/Table;)Ljava/util/Map;
 	public abstract fun getDataTypeProvider ()Lorg/jetbrains/exposed/sql/vendors/DataTypeProvider;
 	public abstract fun getDatabase ()Ljava/lang/String;
 	public abstract fun getDefaultReferenceOption ()Lorg/jetbrains/exposed/sql/ReferenceOption;
@@ -3256,6 +3259,7 @@ public final class org/jetbrains/exposed/sql/vendors/DatabaseDialect$DefaultImpl
 	public static fun dropDatabase (Lorg/jetbrains/exposed/sql/vendors/DatabaseDialect;Ljava/lang/String;)Ljava/lang/String;
 	public static fun dropSchema (Lorg/jetbrains/exposed/sql/vendors/DatabaseDialect;Lorg/jetbrains/exposed/sql/Schema;Z)Ljava/lang/String;
 	public static fun existingIndices (Lorg/jetbrains/exposed/sql/vendors/DatabaseDialect;[Lorg/jetbrains/exposed/sql/Table;)Ljava/util/Map;
+	public static fun existingPrimaryKeys (Lorg/jetbrains/exposed/sql/vendors/DatabaseDialect;[Lorg/jetbrains/exposed/sql/Table;)Ljava/util/Map;
 	public static fun getDefaultReferenceOption (Lorg/jetbrains/exposed/sql/vendors/DatabaseDialect;)Lorg/jetbrains/exposed/sql/ReferenceOption;
 	public static fun getLikePatternSpecialChars (Lorg/jetbrains/exposed/sql/vendors/DatabaseDialect;)Ljava/util/Map;
 	public static fun getNeedsQuotesWhenSymbolsInNames (Lorg/jetbrains/exposed/sql/vendors/DatabaseDialect;)Z
@@ -3554,6 +3558,19 @@ public class org/jetbrains/exposed/sql/vendors/PostgreSQLNGDialect : org/jetbrai
 public final class org/jetbrains/exposed/sql/vendors/PostgreSQLNGDialect$Companion : org/jetbrains/exposed/sql/vendors/VendorDialect$DialectNameProvider {
 }
 
+public final class org/jetbrains/exposed/sql/vendors/PrimaryKeyMetadata {
+	public fun <init> (Ljava/lang/String;Ljava/util/List;)V
+	public final fun component1 ()Ljava/lang/String;
+	public final fun component2 ()Ljava/util/List;
+	public final fun copy (Ljava/lang/String;Ljava/util/List;)Lorg/jetbrains/exposed/sql/vendors/PrimaryKeyMetadata;
+	public static synthetic fun copy$default (Lorg/jetbrains/exposed/sql/vendors/PrimaryKeyMetadata;Ljava/lang/String;Ljava/util/List;ILjava/lang/Object;)Lorg/jetbrains/exposed/sql/vendors/PrimaryKeyMetadata;
+	public fun equals (Ljava/lang/Object;)Z
+	public final fun getColumnNames ()Ljava/util/List;
+	public final fun getName ()Ljava/lang/String;
+	public fun hashCode ()I
+	public fun toString ()Ljava/lang/String;
+}
+
 public class org/jetbrains/exposed/sql/vendors/SQLServerDialect : org/jetbrains/exposed/sql/vendors/VendorDialect {
 	public static final field Companion Lorg/jetbrains/exposed/sql/vendors/SQLServerDialect$Companion;
 	public fun <init> ()V
@@ -3598,6 +3615,7 @@ public final class org/jetbrains/exposed/sql/vendors/SQLiteDialect$Companion : o
 
 public abstract class org/jetbrains/exposed/sql/vendors/VendorDialect : org/jetbrains/exposed/sql/vendors/DatabaseDialect {
 	public fun <init> (Ljava/lang/String;Lorg/jetbrains/exposed/sql/vendors/DataTypeProvider;Lorg/jetbrains/exposed/sql/vendors/FunctionProvider;)V
+	public fun addPrimaryKey (Lorg/jetbrains/exposed/sql/Table;Ljava/lang/String;[Lorg/jetbrains/exposed/sql/Column;)Ljava/lang/String;
 	public fun allTablesNames ()Ljava/util/List;
 	public fun catalog (Lorg/jetbrains/exposed/sql/Transaction;)Ljava/lang/String;
 	public fun checkTableMapping (Lorg/jetbrains/exposed/sql/Table;)Z
@@ -3610,6 +3628,7 @@ public abstract class org/jetbrains/exposed/sql/vendors/VendorDialect : org/jetb
 	public fun dropIndex (Ljava/lang/String;Ljava/lang/String;ZZ)Ljava/lang/String;
 	public fun dropSchema (Lorg/jetbrains/exposed/sql/Schema;Z)Ljava/lang/String;
 	public fun existingIndices ([Lorg/jetbrains/exposed/sql/Table;)Ljava/util/Map;
+	public fun existingPrimaryKeys ([Lorg/jetbrains/exposed/sql/Table;)Ljava/util/Map;
 	protected fun fillConstraintCacheForTables (Ljava/util/List;)V
 	public final fun filterCondition (Lorg/jetbrains/exposed/sql/Index;)Ljava/lang/String;
 	public final fun getAllTablesNames ()Ljava/util/List;
diff --git a/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/SchemaUtils.kt b/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/SchemaUtils.kt
index e29b1ec9fe..d709cd9034 100644
--- a/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/SchemaUtils.kt
+++ b/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/SchemaUtils.kt
@@ -5,6 +5,7 @@ import org.jetbrains.exposed.sql.transactions.TransactionManager
 import org.jetbrains.exposed.sql.vendors.*
 import java.math.BigDecimal
 
+@Suppress("TooManyFunctions")
 object SchemaUtils {
     private inline fun <R> logTimeSpent(message: String, withLogs: Boolean, block: () -> R): R {
         return if (withLogs) {
@@ -19,8 +20,9 @@ object SchemaUtils {
 
     private class TableDepthGraph(val tables: Iterable<Table>) {
         val graph = fetchAllTables().let { tables ->
-            if (tables.isEmpty()) emptyMap()
-            else {
+            if (tables.isEmpty()) {
+                emptyMap()
+            } else {
                 tables.associateWith { t ->
                     t.columns.mapNotNull { c ->
                         c.referee?.let { it.table to c.columnType.nullable }
@@ -125,7 +127,9 @@ object SchemaUtils {
     )
     fun createFKey(reference: Column<*>): List<String> {
         val foreignKey = reference.foreignKey
-        require(foreignKey != null && (foreignKey.deleteRule != null || foreignKey.updateRule != null)) { "$reference does not reference anything" }
+        require(foreignKey != null && (foreignKey.deleteRule != null || foreignKey.updateRule != null)) {
+            "$reference does not reference anything"
+        }
         return createFKey(foreignKey)
     }
 
@@ -196,6 +200,10 @@ object SchemaUtils {
             currentDialect.tableColumns(*tables)
         }
 
+        val existingPrimaryKeys = logTimeSpent("Extracting primary keys", withLogs) {
+            currentDialect.existingPrimaryKeys(*tables)
+        }
+
         val dbSupportsAlterTableWithAddColumn = TransactionManager.current().db.supportsAlterTableWithAddColumn
 
         for (table in tables) {
@@ -222,37 +230,56 @@ object SchemaUtils {
                         val columnType = col.columnType
                         val incorrectNullability = existingCol.nullable != columnType.nullable
                         // Exposed doesn't support changing sequences on columns
-                        val incorrectAutoInc = existingCol.autoIncrement != columnType.isAutoInc && col.autoIncColumnType?.autoincSeq == null
-                        val incorrectDefaults =
-                            existingCol.defaultDbValue != col.dbDefaultValue?.let { dataTypeProvider.dbDefaultToString(col, it) }
+                        val incorrectAutoInc = existingCol.autoIncrement != columnType.isAutoInc &&
+                            col.autoIncColumnType?.autoincSeq == null
+                        val incorrectDefaults = existingCol.defaultDbValue != col.dbDefaultValue?.let {
+                            dataTypeProvider.dbDefaultToString(col, it)
+                        }
                         val incorrectCaseSensitiveName = existingCol.name.inProperCase() != col.nameInDatabaseCase()
                         ColumnDiff(incorrectNullability, incorrectAutoInc, incorrectDefaults, incorrectCaseSensitiveName)
                     }
                     .filterValues { it.hasDifferences() }
 
                 redoColumns.flatMapTo(statements) { (col, changedState) -> col.modifyStatements(changedState) }
+
+                // add missing primary key
+                val missingPK = table.primaryKey?.takeIf { pk -> pk.columns.none { it in missingTableColumns } }
+                if (missingPK != null && existingPrimaryKeys[table] == null) {
+                    val missingPKName = missingPK.name.takeIf { table.isCustomPKNameDefined() }
+                    statements.add(
+                        currentDialect.addPrimaryKey(table, missingPKName, pkColumns = missingPK.columns)
+                    )
+                }
             }
         }
 
         if (dbSupportsAlterTableWithAddColumn) {
-            val existingColumnConstraint = logTimeSpent("Extracting column constraints", withLogs) {
-                currentDialect.columnConstraints(*tables)
-            }
+            statements.addAll(addMissingColumnConstraints(*tables, withLogs = withLogs))
+        }
 
-            val foreignKeyConstraints = tables.flatMap { table ->
-                table.foreignKeys.map { it to existingColumnConstraint[table to it.from]?.firstOrNull() }
-            }
+        return statements
+    }
 
-            for ((foreignKey, existingConstraint) in foreignKeyConstraints) {
-                if (existingConstraint == null) {
-                    statements.addAll(createFKey(foreignKey))
-                } else if (existingConstraint.targetTable != foreignKey.targetTable ||
-                    foreignKey.deleteRule != existingConstraint.deleteRule ||
-                    foreignKey.updateRule != existingConstraint.updateRule
-                ) {
-                    statements.addAll(existingConstraint.dropStatement())
-                    statements.addAll(createFKey(foreignKey))
-                }
+    private fun addMissingColumnConstraints(vararg tables: Table, withLogs: Boolean): List<String> {
+        val existingColumnConstraint = logTimeSpent("Extracting column constraints", withLogs) {
+            currentDialect.columnConstraints(*tables)
+        }
+
+        val foreignKeyConstraints = tables.flatMap { table ->
+            table.foreignKeys.map { it to existingColumnConstraint[table to it.from]?.firstOrNull() }
+        }
+
+        val statements = ArrayList<String>()
+
+        for ((foreignKey, existingConstraint) in foreignKeyConstraints) {
+            if (existingConstraint == null) {
+                statements.addAll(createFKey(foreignKey))
+            } else if (existingConstraint.targetTable != foreignKey.targetTable ||
+                foreignKey.deleteRule != existingConstraint.deleteRule ||
+                foreignKey.updateRule != existingConstraint.updateRule
+            ) {
+                statements.addAll(existingConstraint.dropStatement())
+                statements.addAll(createFKey(foreignKey))
             }
         }
 
@@ -300,7 +327,9 @@ object SchemaUtils {
                     "${currentDialect.name} requires autoCommit to be enabled for CREATE DATABASE",
                     exception
                 )
-            } else throw exception
+            } else {
+                throw exception
+            }
         }
     }
 
@@ -327,7 +356,9 @@ object SchemaUtils {
                     "${currentDialect.name} requires autoCommit to be enabled for DROP DATABASE",
                     exception
                 )
-            } else throw exception
+            } else {
+                throw exception
+            }
         }
     }
 
@@ -367,7 +398,8 @@ object SchemaUtils {
             }
             val executedStatements = createStatements + alterStatements
             logTimeSpent("Checking mapping consistence", withLogs) {
-                val modifyTablesStatements = checkMappingConsistence(tables = tables, withLogs).filter { it !in executedStatements }
+                val modifyTablesStatements = checkMappingConsistence(tables = tables, withLogs)
+                    .filter { it !in executedStatements }
                 execStatements(inBatch, modifyTablesStatements)
                 commit()
             }
@@ -389,7 +421,8 @@ object SchemaUtils {
         }
         val executedStatements = createStatements + alterStatements
         val modifyTablesStatements = logTimeSpent("Checking mapping consistence", withLogs) {
-            checkMappingConsistence(tables = tablesToAlter.toTypedArray(), withLogs).filter { it !in executedStatements }
+            checkMappingConsistence(tables = tablesToAlter.toTypedArray(), withLogs)
+                .filter { it !in executedStatements }
         }
         return executedStatements + modifyTablesStatements
     }
@@ -426,7 +459,9 @@ object SchemaUtils {
         }
 
         val excessiveIndices =
-            currentDialect.existingIndices(*tables).flatMap { it.value }.groupBy { Triple(it.table, it.unique, it.columns.joinToString { it.name }) }
+            currentDialect.existingIndices(*tables)
+                .flatMap { it.value }
+                .groupBy { Triple(it.table, it.unique, it.columns.joinToString { it.name }) }
                 .filter { it.value.size > 1 }
         if (excessiveIndices.isNotEmpty()) {
             exposedLogger.warn("List of excessive indices:")
@@ -486,7 +521,8 @@ object SchemaUtils {
                 nameDiffers.add(mappedIndex)
             }
 
-            notMappedIndices.getOrPut(table.nameInDatabaseCase()) { hashSetOf() }.addAll(existingTableIndices.subtract(mappedIndices))
+            notMappedIndices.getOrPut(table.nameInDatabaseCase()) { hashSetOf() }
+                .addAll(existingTableIndices.subtract(mappedIndices))
 
             missingIndices.addAll(mappedIndices.subtract(existingTableIndices))
         }
@@ -601,7 +637,11 @@ object SchemaUtils {
     fun dropSchema(vararg schemas: Schema, cascade: Boolean = false, inBatch: Boolean = false) {
         if (schemas.isEmpty()) return
         with(TransactionManager.current()) {
-            val schemasForDeletion = if (currentDialect.supportsIfNotExists) schemas.distinct() else schemas.distinct().filter { it.exists() }
+            val schemasForDeletion = if (currentDialect.supportsIfNotExists) {
+                schemas.distinct()
+            } else {
+                schemas.distinct().filter { it.exists() }
+            }
             val dropStatements = schemasForDeletion.flatMap { it.dropStatement(cascade) }
 
             execStatements(inBatch, dropStatements)
diff --git a/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/statements/api/ExposedDatabaseMetadata.kt b/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/statements/api/ExposedDatabaseMetadata.kt
index 572018165c..19e6225062 100644
--- a/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/statements/api/ExposedDatabaseMetadata.kt
+++ b/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/statements/api/ExposedDatabaseMetadata.kt
@@ -4,6 +4,7 @@ import org.jetbrains.exposed.sql.ForeignKeyConstraint
 import org.jetbrains.exposed.sql.Index
 import org.jetbrains.exposed.sql.Table
 import org.jetbrains.exposed.sql.vendors.ColumnMetadata
+import org.jetbrains.exposed.sql.vendors.PrimaryKeyMetadata
 import java.math.BigDecimal
 
 abstract class ExposedDatabaseMetadata(val database: String) {
@@ -33,6 +34,8 @@ abstract class ExposedDatabaseMetadata(val database: String) {
 
     abstract fun existingIndices(vararg tables: Table): Map<Table, List<Index>>
 
+    abstract fun existingPrimaryKeys(vararg tables: Table): Map<Table, PrimaryKeyMetadata?>
+
     abstract fun tableConstraints(tables: List<Table>): Map<String, List<ForeignKeyConstraint>>
 
     abstract fun cleanCache()
diff --git a/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/vendors/Default.kt b/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/vendors/Default.kt
index dd272ade4e..8e5a163478 100644
--- a/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/vendors/Default.kt
+++ b/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/vendors/Default.kt
@@ -836,6 +836,16 @@ data class ColumnMetadata(
     val defaultDbValue: String?,
 )
 
+/**
+ * Represents metadata information about a specific table's primary key.
+ */
+data class PrimaryKeyMetadata(
+    /** Name of the primary key. */
+    val name: String,
+    /** Names of the primary key's columns. */
+    val columnNames: List<String>
+)
+
 /**
  * Common interface for all database dialects.
  */
@@ -911,11 +921,16 @@ interface DatabaseDialect {
     fun tableColumns(vararg tables: Table): Map<Table, List<ColumnMetadata>> = emptyMap()
 
     /** Returns a map with the foreign key constraints of all the defined columns sets in each of the specified [tables]. */
-    fun columnConstraints(vararg tables: Table): Map<Pair<Table, LinkedHashSet<Column<*>>>, List<ForeignKeyConstraint>> = emptyMap()
+    fun columnConstraints(
+        vararg tables: Table
+    ): Map<Pair<Table, LinkedHashSet<Column<*>>>, List<ForeignKeyConstraint>> = emptyMap()
 
     /** Returns a map with all the defined indices in each of the specified [tables]. */
     fun existingIndices(vararg tables: Table): Map<Table, List<Index>> = emptyMap()
 
+    /** Returns a map with the primary key metadata in each of the specified [tables]. */
+    fun existingPrimaryKeys(vararg tables: Table): Map<Table, PrimaryKeyMetadata?> = emptyMap()
+
     /** Returns `true` if the dialect supports `SELECT FOR UPDATE` statements, `false` otherwise. */
     fun supportsSelectForUpdate(): Boolean
 
@@ -942,6 +957,9 @@ interface DatabaseDialect {
     /** Returns the SQL command that modifies the specified [column]. */
     fun modifyColumn(column: Column<*>, columnDiff: ColumnDiff): List<String>
 
+    /** Returns the SQL command that adds a primary key specified [pkName] to an existing [table]. */
+    fun addPrimaryKey(table: Table, pkName: String?, vararg pkColumns: Column<*>): String
+
     fun createDatabase(name: String) = "CREATE DATABASE IF NOT EXISTS ${name.inProperCase()}"
 
     fun dropDatabase(name: String) = "DROP DATABASE IF EXISTS ${name.inProperCase()}"
@@ -994,7 +1012,11 @@ sealed class ForUpdateOption(open val querySuffix: String) {
             NO_WAIT("NOWAIT"), SKIP_LOCKED("SKIP LOCKED")
         }
 
-        abstract class ForUpdateBase(querySuffix: String, private val mode: MODE? = null, private vararg val ofTables: Table) : ForUpdateOption("") {
+        abstract class ForUpdateBase(
+            querySuffix: String,
+            private val mode: MODE? = null,
+            private vararg val ofTables: Table
+        ) : ForUpdateOption("") {
             private val preparedQuerySuffix = buildString {
                 append(querySuffix)
                 ofTables.takeIf { it.isNotEmpty() }?.let { tables ->
@@ -1008,17 +1030,29 @@ sealed class ForUpdateOption(open val querySuffix: String) {
             final override val querySuffix: String = preparedQuerySuffix
         }
 
-        class ForUpdate(mode: MODE? = null, vararg ofTables: Table) : ForUpdateBase("FOR UPDATE", mode, ofTables = ofTables)
+        class ForUpdate(
+            mode: MODE? = null,
+            vararg ofTables: Table
+        ) : ForUpdateBase("FOR UPDATE", mode, ofTables = ofTables)
 
-        open class ForNoKeyUpdate(mode: MODE? = null, vararg ofTables: Table) : ForUpdateBase("FOR NO KEY UPDATE", mode, ofTables = ofTables) {
+        open class ForNoKeyUpdate(
+            mode: MODE? = null,
+            vararg ofTables: Table
+        ) : ForUpdateBase("FOR NO KEY UPDATE", mode, ofTables = ofTables) {
             companion object : ForNoKeyUpdate()
         }
 
-        open class ForShare(mode: MODE? = null, vararg ofTables: Table) : ForUpdateBase("FOR SHARE", mode, ofTables = ofTables) {
+        open class ForShare(
+            mode: MODE? = null,
+            vararg ofTables: Table
+        ) : ForUpdateBase("FOR SHARE", mode, ofTables = ofTables) {
             companion object : ForShare()
         }
 
-        open class ForKeyShare(mode: MODE? = null, vararg ofTables: Table) : ForUpdateBase("FOR KEY SHARE", mode, ofTables = ofTables) {
+        open class ForKeyShare(
+            mode: MODE? = null,
+            vararg ofTables: Table
+        ) : ForUpdateBase("FOR KEY SHARE", mode, ofTables = ofTables) {
             companion object : ForKeyShare()
         }
     }
@@ -1107,7 +1141,9 @@ abstract class VendorDialect(
     override fun tableColumns(vararg tables: Table): Map<Table, List<ColumnMetadata>> =
         TransactionManager.current().connection.metadata { columns(*tables) }
 
-    override fun columnConstraints(vararg tables: Table): Map<Pair<Table, LinkedHashSet<Column<*>>>, List<ForeignKeyConstraint>> {
+    override fun columnConstraints(
+        vararg tables: Table
+    ): Map<Pair<Table, LinkedHashSet<Column<*>>>, List<ForeignKeyConstraint>> {
         val constraints = HashMap<Pair<Table, LinkedHashSet<Column<*>>>, MutableList<ForeignKeyConstraint>>()
 
         val tablesToLoad = tables.filter { !columnConstraintsCache.containsKey(it.nameInDatabaseCase()) }
@@ -1124,7 +1160,12 @@ abstract class VendorDialect(
     override fun existingIndices(vararg tables: Table): Map<Table, List<Index>> =
         TransactionManager.current().db.metadata { existingIndices(*tables) }
 
-    private val supportsSelectForUpdate: Boolean by lazy { TransactionManager.current().db.metadata { supportsSelectForUpdate } }
+    override fun existingPrimaryKeys(vararg tables: Table): Map<Table, PrimaryKeyMetadata?> =
+        TransactionManager.current().db.metadata { existingPrimaryKeys(*tables) }
+
+    private val supportsSelectForUpdate: Boolean by lazy {
+        TransactionManager.current().db.metadata { supportsSelectForUpdate }
+    }
 
     override fun supportsSelectForUpdate(): Boolean = supportsSelectForUpdate
 
@@ -1232,6 +1273,13 @@ abstract class VendorDialect(
 
     override fun modifyColumn(column: Column<*>, columnDiff: ColumnDiff): List<String> =
         listOf("ALTER TABLE ${TransactionManager.current().identity(column.table)} MODIFY COLUMN ${column.descriptionDdl(true)}")
+
+    override fun addPrimaryKey(table: Table, pkName: String?, vararg pkColumns: Column<*>): String {
+        val transaction = TransactionManager.current()
+        val columns = pkColumns.joinToString(prefix = "(", postfix = ")") { transaction.identity(it) }
+        val constraint = pkName?.let { " CONSTRAINT ${identifierManager.quoteIfNecessary(it)} " } ?: " "
+        return "ALTER TABLE ${transaction.identity(table)} ADD${constraint}PRIMARY KEY $columns"
+    }
 }
 
 private val explicitDialect = ThreadLocal<DatabaseDialect?>()
diff --git a/exposed-jdbc/api/exposed-jdbc.api b/exposed-jdbc/api/exposed-jdbc.api
index 55864e20aa..ce7bcb56f3 100644
--- a/exposed-jdbc/api/exposed-jdbc.api
+++ b/exposed-jdbc/api/exposed-jdbc.api
@@ -37,6 +37,7 @@ public final class org/jetbrains/exposed/sql/statements/jdbc/JdbcDatabaseMetadat
 	public fun cleanCache ()V
 	public fun columns ([Lorg/jetbrains/exposed/sql/Table;)Ljava/util/Map;
 	public fun existingIndices ([Lorg/jetbrains/exposed/sql/Table;)Ljava/util/Map;
+	public fun existingPrimaryKeys ([Lorg/jetbrains/exposed/sql/Table;)Ljava/util/Map;
 	public fun getCurrentScheme ()Ljava/lang/String;
 	public fun getDatabaseDialectName ()Ljava/lang/String;
 	public fun getDatabaseProductVersion ()Ljava/lang/String;
diff --git a/exposed-jdbc/src/main/kotlin/org/jetbrains/exposed/sql/statements/jdbc/JdbcDatabaseMetadataImpl.kt b/exposed-jdbc/src/main/kotlin/org/jetbrains/exposed/sql/statements/jdbc/JdbcDatabaseMetadataImpl.kt
index 6becca8e4f..27bfefa0d1 100644
--- a/exposed-jdbc/src/main/kotlin/org/jetbrains/exposed/sql/statements/jdbc/JdbcDatabaseMetadataImpl.kt
+++ b/exposed-jdbc/src/main/kotlin/org/jetbrains/exposed/sql/statements/jdbc/JdbcDatabaseMetadataImpl.kt
@@ -80,7 +80,10 @@ class JdbcDatabaseMetadataImpl(database: String, val metadata: DatabaseMetaData)
         _currentScheme = null
     }
 
-    private inner class CachableMapWithDefault<K, V>(private val map: MutableMap<K, V> = mutableMapOf(), val default: (K) -> V) : Map<K, V> by map {
+    private inner class CachableMapWithDefault<K, V>(
+        private val map: MutableMap<K, V> = mutableMapOf(),
+        val default: (K) -> V
+    ) : Map<K, V> by map {
         override fun get(key: K): V? = map.getOrPut(key) { default(key) }
         override fun containsKey(key: K): Boolean = true
         override fun isEmpty(): Boolean = false
@@ -239,6 +242,21 @@ class JdbcDatabaseMetadataImpl(database: String, val metadata: DatabaseMetaData)
         return HashMap(existingIndicesCache)
     }
 
+    override fun existingPrimaryKeys(vararg tables: Table): Map<Table, PrimaryKeyMetadata?> {
+        return tables.associateWith { table ->
+            metadata.getPrimaryKeys(databaseName, currentScheme, table.nameInDatabaseCase()).let { rs ->
+                val columnNames = mutableListOf<String>()
+                var pkName = ""
+                while (rs.next()) {
+                    rs.getString("PK_NAME")?.let { pkName = it }
+                    columnNames += rs.getString("COLUMN_NAME")
+                }
+                rs.close()
+                if (pkName.isEmpty()) null else PrimaryKeyMetadata(pkName, columnNames)
+            }
+        }
+    }
+
     @Synchronized
     override fun tableConstraints(tables: List<Table>): Map<String, List<ForeignKeyConstraint>> {
         val allTables = SchemaUtils.sortTablesByReferences(tables).associateBy { it.nameInDatabaseCase() }
diff --git a/exposed-tests/src/test/kotlin/org/jetbrains/exposed/sql/tests/postgresql/PostgresqlTests.kt b/exposed-tests/src/test/kotlin/org/jetbrains/exposed/sql/tests/postgresql/PostgresqlTests.kt
index b78720fa69..5544351818 100644
--- a/exposed-tests/src/test/kotlin/org/jetbrains/exposed/sql/tests/postgresql/PostgresqlTests.kt
+++ b/exposed-tests/src/test/kotlin/org/jetbrains/exposed/sql/tests/postgresql/PostgresqlTests.kt
@@ -5,10 +5,14 @@ import org.jetbrains.exposed.sql.*
 import org.jetbrains.exposed.sql.tests.DatabaseTestsBase
 import org.jetbrains.exposed.sql.tests.RepeatableTestRule
 import org.jetbrains.exposed.sql.tests.TestDB
+import org.jetbrains.exposed.sql.tests.shared.assertFailAndRollback
+import org.jetbrains.exposed.sql.tests.shared.assertFalse
+import org.jetbrains.exposed.sql.tests.shared.assertTrue
 import org.jetbrains.exposed.sql.vendors.ForUpdateOption
 import org.jetbrains.exposed.sql.vendors.ForUpdateOption.PostgreSQL
 import org.junit.Rule
 import org.junit.Test
+import java.sql.ResultSet
 import kotlin.test.assertEquals
 
 class PostgresqlTests : DatabaseTestsBase() {
@@ -61,6 +65,57 @@ class PostgresqlTests : DatabaseTestsBase() {
         }
     }
 
+    @Test
+    fun testPrimaryKeyCreatedInPostgresql() {
+        val tableName = "tester"
+        val tester1 = object : Table(tableName) {
+            val age = integer("age")
+        }
+
+        val tester2 = object : Table(tableName) {
+            val age = integer("age")
+
+            override val primaryKey = PrimaryKey(age)
+        }
+
+        val tester3 = object : IntIdTable(tableName) {
+            val age = integer("age")
+        }
+
+        fun <T : Any> Transaction.assertPrimaryKey(transform: (ResultSet) -> T): T? {
+            return exec(
+                """
+                SELECT ct.relname as TABLE_NAME, ci.relname AS PK_NAME
+                FROM pg_catalog.pg_class ct
+                JOIN pg_index i ON (ct.oid = i.indrelid AND indisprimary)
+                JOIN pg_catalog.pg_class ci ON (ci.oid = i.indexrelid)
+                WHERE ct.relname IN ('$tableName')
+                """.trimIndent()
+            ) { rs ->
+                transform(rs)
+            }
+        }
+        withDb(listOf(TestDB.POSTGRESQLNG, TestDB.POSTGRESQL)) {
+            val defaultPKName = "tester_pkey"
+            SchemaUtils.createMissingTablesAndColumns(tester1)
+            assertPrimaryKey {
+                assertFalse(it.next())
+            }
+
+            SchemaUtils.createMissingTablesAndColumns(tester2)
+            assertPrimaryKey {
+                assertTrue(it.next())
+                assertEquals(defaultPKName, it.getString("PK_NAME"))
+            }
+
+            assertFailAndRollback("Multiple primary keys are not allowed") {
+                SchemaUtils.createMissingTablesAndColumns(tester3)
+            }
+
+            SchemaUtils.drop(tester1)
+        }
+    }
+
     private fun Transaction.withTable(statement: Transaction.() -> Unit) {
         SchemaUtils.create(table)
         try {
diff --git a/exposed-tests/src/test/kotlin/org/jetbrains/exposed/sql/tests/shared/ddl/CreateMissingTablesAndColumnsTests.kt b/exposed-tests/src/test/kotlin/org/jetbrains/exposed/sql/tests/shared/ddl/CreateMissingTablesAndColumnsTests.kt
index 7b11a1dc0a..c85d1b1120 100644
--- a/exposed-tests/src/test/kotlin/org/jetbrains/exposed/sql/tests/shared/ddl/CreateMissingTablesAndColumnsTests.kt
+++ b/exposed-tests/src/test/kotlin/org/jetbrains/exposed/sql/tests/shared/ddl/CreateMissingTablesAndColumnsTests.kt
@@ -16,10 +16,13 @@ import org.jetbrains.exposed.sql.tests.shared.assertFailAndRollback
 import org.jetbrains.exposed.sql.tests.shared.assertTrue
 import org.jetbrains.exposed.sql.vendors.MysqlDialect
 import org.jetbrains.exposed.sql.vendors.OracleDialect
+import org.jetbrains.exposed.sql.vendors.PrimaryKeyMetadata
 import org.junit.Test
 import java.math.BigDecimal
 import java.util.*
 import kotlin.properties.Delegates
+import kotlin.test.assertNotNull
+import kotlin.test.assertNull
 
 class CreateMissingTablesAndColumnsTests : DatabaseTestsBase() {
 
@@ -184,7 +187,6 @@ class CreateMissingTablesAndColumnsTests : DatabaseTestsBase() {
         }
 
         withDb {
-//        withDb(db = listOf(TestDB.H2)) {
             SchemaUtils.createMissingTablesAndColumns(t1)
 
             val missingStatements = SchemaUtils.addMissingColumnsStatements(t2)
@@ -210,7 +212,6 @@ class CreateMissingTablesAndColumnsTests : DatabaseTestsBase() {
         }
 
         withDb {
-//        withDb(db = listOf(TestDB.H2)) {
             SchemaUtils.createMissingTablesAndColumns(t1)
 
             val missingStatements = SchemaUtils.addMissingColumnsStatements(t2)
@@ -240,7 +241,8 @@ class CreateMissingTablesAndColumnsTests : DatabaseTestsBase() {
         }
     }
 
-    @Test fun addAutoPrimaryKey() {
+    @Test
+    fun addAutoPrimaryKey() {
         val tableName = "Foo"
         val initialTable = object : Table(tableName) {
             val bar = text("bar")
@@ -255,6 +257,37 @@ class CreateMissingTablesAndColumnsTests : DatabaseTestsBase() {
         }
     }
 
+    @Test
+    fun testAddNewPrimaryKeyOnExistingColumn() {
+        val tableName = "tester"
+        val noPKTable = object : Table(tableName) {
+            val bar = integer("bar")
+        }
+
+        val singlePKTable = object : Table(tableName) {
+            val bar = integer("bar")
+
+            override val primaryKey = PrimaryKey(bar)
+        }
+
+        withDb(excludeSettings = listOf(TestDB.SQLITE)) {
+            SchemaUtils.createMissingTablesAndColumns(noPKTable)
+            var primaryKey: PrimaryKeyMetadata? = currentDialectTest.existingPrimaryKeys(singlePKTable)[singlePKTable]
+            assertNull(primaryKey)
+
+            val expected = "ALTER TABLE ${tableName.inProperCase()} ADD PRIMARY KEY (${noPKTable.bar.nameInDatabaseCase()})"
+            val statements = SchemaUtils.statementsRequiredToActualizeScheme(singlePKTable)
+            assertEquals(expected, statements.single())
+
+            SchemaUtils.createMissingTablesAndColumns(singlePKTable)
+            primaryKey = currentDialectTest.existingPrimaryKeys(singlePKTable)[singlePKTable]
+            assertNotNull(primaryKey)
+            assertEquals("bar".inProperCase(), primaryKey.columnNames.single())
+
+            SchemaUtils.drop(noPKTable)
+        }
+    }
+
     @Test
     fun `columns with default values that haven't changed shouldn't trigger change`() {
         var table by Delegates.notNull<Table>()