JetBrains · bog-walk · Feb 13, 2025 · Feb 5, 2025 · Feb 12, 2025 · Feb 12, 2025
diff --git a/exposed-core/api/exposed-core.api b/exposed-core/api/exposed-core.api
@@ -2410,9 +2410,6 @@ public abstract class org/jetbrains/exposed/sql/Transaction : org/jetbrains/expo
 	public final fun getDebug ()Z
 	public final fun getDuration ()J
 	public final fun getId ()Ljava/lang/String;
-	public final fun getMaxAttempts ()I
-	public final fun getMaxRetryDelay ()J
-	public final fun getMinRetryDelay ()J
 	public final fun getQueryTimeout ()Ljava/lang/Integer;
 	public final fun getStatementCount ()I
 	public final fun getStatementStats ()Ljava/util/HashMap;
@@ -2422,9 +2419,6 @@ public abstract class org/jetbrains/exposed/sql/Transaction : org/jetbrains/expo
 	public final fun identity (Lorg/jetbrains/exposed/sql/Table;)Ljava/lang/String;
 	public final fun setDebug (Z)V
 	public final fun setDuration (J)V
-	public final fun setMaxAttempts (I)V
-	public final fun setMaxRetryDelay (J)V
-	public final fun setMinRetryDelay (J)V
 	public final fun setQueryTimeout (Ljava/lang/Integer;)V
 	public final fun setStatementCount (I)V
 	public final fun setWarnLongQueriesDuration (Ljava/lang/Long;)V
@@ -3365,15 +3359,28 @@ public abstract interface class org/jetbrains/exposed/sql/statements/api/ResultA
 	public abstract fun releaseResult ()V
 }
 
+public final class org/jetbrains/exposed/sql/transactions/CoreTransactionManager {
+	public static final field INSTANCE Lorg/jetbrains/exposed/sql/transactions/CoreTransactionManager;
+	public final fun closeAndUnregisterDatabase (Lorg/jetbrains/exposed/sql/DatabaseApi;)V
+	public final fun currentTransaction ()Lorg/jetbrains/exposed/sql/Transaction;
+	public final fun currentTransactionOrNull ()Lorg/jetbrains/exposed/sql/Transaction;
+	public final fun getCurrentThreadManager ()Lorg/jetbrains/exposed/sql/transactions/TransactionManagerApi;
+	public final fun getDatabaseManager (Lorg/jetbrains/exposed/sql/DatabaseApi;)Lorg/jetbrains/exposed/sql/transactions/TransactionManagerApi;
+	public final fun getDefaultDatabase ()Lorg/jetbrains/exposed/sql/DatabaseApi;
+	public final fun getDefaultDatabaseOrFirst ()Lorg/jetbrains/exposed/sql/DatabaseApi;
+	public final fun registerDatabaseManager (Lorg/jetbrains/exposed/sql/DatabaseApi;Lorg/jetbrains/exposed/sql/transactions/TransactionManagerApi;)V
+	public final fun resetCurrentThreadManager (Lorg/jetbrains/exposed/sql/transactions/TransactionManagerApi;)V
+	public final fun setDefaultDatabase (Lorg/jetbrains/exposed/sql/DatabaseApi;)V
+}
+
 public abstract interface class org/jetbrains/exposed/sql/transactions/TransactionInterface {
 	public abstract fun getDb ()Lorg/jetbrains/exposed/sql/DatabaseApi;
 	public abstract fun getOuterTransaction ()Lorg/jetbrains/exposed/sql/Transaction;
 	public abstract fun getReadOnly ()Z
 	public abstract fun getTransactionIsolation ()I
 }
 
-public abstract interface class org/jetbrains/exposed/sql/transactions/TransactionManager {
-	public static final field Companion Lorg/jetbrains/exposed/sql/transactions/TransactionManager$Companion;
+public abstract interface class org/jetbrains/exposed/sql/transactions/TransactionManagerApi {
 	public abstract fun bindTransactionToThread (Lorg/jetbrains/exposed/sql/Transaction;)V
 	public abstract fun currentOrNull ()Lorg/jetbrains/exposed/sql/Transaction;
 	public abstract fun getDefaultIsolationLevel ()I
@@ -3389,23 +3396,8 @@ public abstract interface class org/jetbrains/exposed/sql/transactions/Transacti
 	public abstract fun setDefaultReadOnly (Z)V
 }
 
-public final class org/jetbrains/exposed/sql/transactions/TransactionManager$Companion {
-	public final fun closeAndUnregister (Lorg/jetbrains/exposed/sql/DatabaseApi;)V
-	public final fun current ()Lorg/jetbrains/exposed/sql/Transaction;
-	public final fun currentOrNew (I)Lorg/jetbrains/exposed/sql/Transaction;
-	public final fun currentOrNull ()Lorg/jetbrains/exposed/sql/Transaction;
-	public final fun getCurrentDefaultDatabase ()Ljava/util/concurrent/atomic/AtomicReference;
-	public final fun getDefaultDatabase ()Lorg/jetbrains/exposed/sql/DatabaseApi;
-	public final fun getManager ()Lorg/jetbrains/exposed/sql/transactions/TransactionManager;
-	public final fun isInitialized ()Z
-	public final fun managerFor (Lorg/jetbrains/exposed/sql/DatabaseApi;)Lorg/jetbrains/exposed/sql/transactions/TransactionManager;
-	public final fun registerManager (Lorg/jetbrains/exposed/sql/DatabaseApi;Lorg/jetbrains/exposed/sql/transactions/TransactionManager;)V
-	public final fun resetCurrent (Lorg/jetbrains/exposed/sql/transactions/TransactionManager;)V
-	public final fun setDefaultDatabase (Lorg/jetbrains/exposed/sql/DatabaseApi;)V
-}
-
-public final class org/jetbrains/exposed/sql/transactions/TransactionManager$DefaultImpls {
-	public static synthetic fun newTransaction$default (Lorg/jetbrains/exposed/sql/transactions/TransactionManager;IZLorg/jetbrains/exposed/sql/Transaction;ILjava/lang/Object;)Lorg/jetbrains/exposed/sql/Transaction;
+public final class org/jetbrains/exposed/sql/transactions/TransactionManagerApi$DefaultImpls {
+	public static synthetic fun newTransaction$default (Lorg/jetbrains/exposed/sql/transactions/TransactionManagerApi;IZLorg/jetbrains/exposed/sql/Transaction;ILjava/lang/Object;)Lorg/jetbrains/exposed/sql/Transaction;
 }
 
 public final class org/jetbrains/exposed/sql/transactions/TransactionScopeKt {

diff --git a/exposed-core/build.gradle.kts b/exposed-core/build.gradle.kts
@@ -1,3 +1,5 @@
+import org.jetbrains.kotlin.gradle.tasks.KotlinCompilationTask
+
 plugins {
     kotlin("jvm")
 
@@ -18,3 +20,7 @@ dependencies {
     api(libs.kotlinx.coroutines)
     api(libs.slf4j)
 }
+
+tasks.named<KotlinCompilationTask<*>>("compileKotlin").configure {
+    compilerOptions.optIn.add("org.jetbrains.exposed.sql.InternalApi")
+}
diff --git a/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/AbstractQuery.kt b/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/AbstractQuery.kt
@@ -3,7 +3,7 @@ package org.jetbrains.exposed.sql
 import org.jetbrains.exposed.sql.statements.Statement
 import org.jetbrains.exposed.sql.statements.StatementType
 import org.jetbrains.exposed.sql.statements.api.ResultApi
-import org.jetbrains.exposed.sql.transactions.TransactionManager
+import org.jetbrains.exposed.sql.transactions.CoreTransactionManager
 import org.jetbrains.exposed.sql.vendors.ForUpdateOption
 import org.jetbrains.exposed.sql.vendors.currentDialect
 
@@ -230,7 +230,7 @@ abstract class AbstractQuery<T : AbstractQuery<T>>(
             }
             if (set.source != Table.Dual || currentDialect.supportsDualTableConcept) {
                 append(" FROM ")
-                set.source.describe(TransactionManager.current(), this)
+                set.source.describe(CoreTransactionManager.currentTransaction(), this)
             }
 
             where?.let {

diff --git a/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/Column.kt b/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/Column.kt
@@ -1,7 +1,7 @@
 package org.jetbrains.exposed.sql
 
 import org.jetbrains.exposed.exceptions.throwUnsupportedException
-import org.jetbrains.exposed.sql.transactions.TransactionManager
+import org.jetbrains.exposed.sql.transactions.CoreTransactionManager
 import org.jetbrains.exposed.sql.vendors.*
 
 private val comparator: Comparator<Column<*>> = compareBy({ it.table.tableName }, { it.name })
@@ -43,7 +43,7 @@ class Column<T>(
     internal var extraDefinitions = mutableListOf<Any>()
 
     /** Appends the SQL representation of this column to the specified [queryBuilder]. */
-    override fun toQueryBuilder(queryBuilder: QueryBuilder): Unit = TransactionManager.current().fullIdentity(this@Column, queryBuilder)
+    override fun toQueryBuilder(queryBuilder: QueryBuilder): Unit = CoreTransactionManager.currentTransaction().fullIdentity(this@Column, queryBuilder)
 
     /** Returns the column name in proper case. */
     fun nameInDatabaseCase(): String = name.inProperCase()
@@ -68,7 +68,7 @@ class Column<T>(
         }
 
     override fun createStatement(): List<String> {
-        val alterTablePrefix = "ALTER TABLE ${TransactionManager.current().identity(table)} ADD"
+        val alterTablePrefix = "ALTER TABLE ${CoreTransactionManager.currentTransaction().identity(table)} ADD"
         val isH2withCustomPKConstraint = currentDialect is H2Dialect && isLastColumnInPK
         val isOracle = currentDialect is OracleDialect
         val columnDefinition = when {
@@ -93,7 +93,7 @@ class Column<T>(
     override fun modifyStatement(): List<String> = currentDialect.modifyColumn(this, ColumnDiff.AllChanged)
 
     override fun dropStatement(): List<String> {
-        val tr = TransactionManager.current()
+        val tr = CoreTransactionManager.currentTransaction()
         return listOf("ALTER TABLE ${tr.identity(table)} DROP COLUMN ${tr.identity(this)}")
     }
 
@@ -102,7 +102,7 @@ class Column<T>(
     /** Returns the SQL representation of this column. */
     @Suppress("ComplexMethod")
     fun descriptionDdl(modify: Boolean = false): String = buildString {
-        val tr = TransactionManager.current()
+        val tr = CoreTransactionManager.currentTransaction()
         val column = this@Column
         append(tr.identity(column))
         append(" ")

diff --git a/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/Constraints.kt b/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/Constraints.kt
@@ -1,6 +1,6 @@
 package org.jetbrains.exposed.sql
 
-import org.jetbrains.exposed.sql.transactions.TransactionManager
+import org.jetbrains.exposed.sql.transactions.CoreTransactionManager
 import org.jetbrains.exposed.sql.vendors.DatabaseDialect
 import org.jetbrains.exposed.sql.vendors.MysqlDialect
 import org.jetbrains.exposed.sql.vendors.SQLiteDialect
@@ -66,7 +66,7 @@ data class ForeignKeyConstraint(
     ) : this(mapOf(from to target), onUpdate, onDelete, name)
 
     private val tx: Transaction
-        get() = TransactionManager.current()
+        get() = CoreTransactionManager.currentTransaction()
 
     /** The columns of the referenced parent table. */
     val target: LinkedHashSet<Column<*>> = LinkedHashSet(references.values)
@@ -222,7 +222,7 @@ data class CheckConstraint(
     companion object {
         internal fun from(table: Table, name: String, op: Op<Boolean>): CheckConstraint {
             require(name.isNotBlank()) { "Check constraint name cannot be blank" }
-            val tr = TransactionManager.current()
+            val tr = CoreTransactionManager.currentTransaction()
             val identifierManager = tr.db.identifierManager
             val tableName = tr.identity(table)
             val checkOpSQL = op.toString().replace("$tableName.", "")

diff --git a/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/Expression.kt b/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/Expression.kt
@@ -1,7 +1,7 @@
 package org.jetbrains.exposed.sql
 
 import org.jetbrains.exposed.sql.statements.DefaultValueMarker
-import org.jetbrains.exposed.sql.transactions.TransactionManager
+import org.jetbrains.exposed.sql.transactions.CoreTransactionManager
 
 /**
  * An object to which SQL expressions and values can be appended.
@@ -70,7 +70,7 @@ class QueryBuilder(
     fun <T> registerArgument(column: Column<*>, argument: T) {
         when (argument) {
             is Expression<*> -> append(argument)
-            DefaultValueMarker -> append(TransactionManager.current().db.dialect.dataTypeProvider.processForDefaultValue(column.dbDefaultValue!!))
+            DefaultValueMarker -> append(CoreTransactionManager.currentTransaction().db.dialect.dataTypeProvider.processForDefaultValue(column.dbDefaultValue!!))
             else -> registerArgument(column.columnType, argument)
         }
     }

diff --git a/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/ResultRow.kt b/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/ResultRow.kt
@@ -4,7 +4,7 @@ import org.jetbrains.exposed.dao.id.CompositeID
 import org.jetbrains.exposed.dao.id.CompositeIdTable
 import org.jetbrains.exposed.dao.id.EntityID
 import org.jetbrains.exposed.sql.statements.api.ResultApi
-import org.jetbrains.exposed.sql.transactions.TransactionManager
+import org.jetbrains.exposed.sql.transactions.CoreTransactionManager
 import org.jetbrains.exposed.sql.vendors.withDialect
 
 /** A row of data representing a single record retrieved from a database result set. */
@@ -13,7 +13,7 @@ class ResultRow(
     val fieldIndex: Map<Expression<*>, Int>,
     private val data: Array<Any?> = arrayOfNulls<Any?>(fieldIndex.size)
 ) {
-    private val database: DatabaseApi? = TransactionManager.currentOrNull()?.db
+    private val database: DatabaseApi? = CoreTransactionManager.currentTransactionOrNull()?.db
 
     private val lookUpCache = ResultRowCache()
 
@@ -74,7 +74,7 @@ class ResultRow(
         if (checkNullability) {
             if (rawValue == null && expression is Column<*> && expression.dbDefaultValue != null && !expression.columnType.nullable) {
                 exposedLogger.warn(
-                    "Column ${TransactionManager.current().fullIdentity(expression)} is marked as not null, " +
+                    "Column ${CoreTransactionManager.currentTransaction().fullIdentity(expression)} is marked as not null, " +
                         "has default db value, but returns null. Possible have to re-read it from DB."
                 )
             }

diff --git a/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/SQLLog.kt b/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/SQLLog.kt
@@ -4,7 +4,7 @@ import org.jetbrains.exposed.sql.statements.StatementContext
 import org.jetbrains.exposed.sql.statements.StatementInterceptor
 import org.jetbrains.exposed.sql.statements.api.PreparedStatementApi
 import org.jetbrains.exposed.sql.statements.expandArgs
-import org.jetbrains.exposed.sql.transactions.TransactionManager
+import org.jetbrains.exposed.sql.transactions.CoreTransactionManager
 import org.slf4j.LoggerFactory
 
 /** Base class representing a provider of log messages. */
@@ -33,7 +33,7 @@ object Slf4jSqlDebugLogger : SqlLogger {
      */
     override fun log(context: StatementContext, transaction: Transaction) {
         if (exposedLogger.isDebugEnabled) {
-            exposedLogger.debug(context.expandArgs(TransactionManager.current()))
+            exposedLogger.debug(context.expandArgs(CoreTransactionManager.currentTransaction()))
         }
     }
 }

diff --git a/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/Schema.kt b/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/Schema.kt
@@ -1,9 +1,8 @@
 package org.jetbrains.exposed.sql
 
 import org.jetbrains.exposed.exceptions.UnsupportedByDialectException
-import org.jetbrains.exposed.sql.transactions.TransactionManager
+import org.jetbrains.exposed.sql.transactions.CoreTransactionManager
 import org.jetbrains.exposed.sql.vendors.currentDialect
-import java.lang.StringBuilder
 
 /**
  * Represents a database schema.
@@ -27,7 +26,7 @@ data class Schema(
 ) {
     /** This schema's name in proper database casing. */
     val identifier
-        get() = TransactionManager.current().db.identifierManager.cutIfNecessaryAndQuote(name)
+        get() = CoreTransactionManager.currentTransaction().db.identifierManager.cutIfNecessaryAndQuote(name)
 
     /** The SQL statements that create this schema. */
     val ddl: List<String>

diff --git a/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/Sequence.kt b/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/Sequence.kt
@@ -1,7 +1,7 @@
 package org.jetbrains.exposed.sql
 
 import org.jetbrains.exposed.exceptions.UnsupportedByDialectException
-import org.jetbrains.exposed.sql.transactions.TransactionManager
+import org.jetbrains.exposed.sql.transactions.CoreTransactionManager
 import org.jetbrains.exposed.sql.vendors.currentDialect
 
 /**
@@ -26,8 +26,8 @@ class Sequence(
     val cache: Long? = null
 ) {
     /** This name of this sequence in proper database casing. */
-    val identifier
-        get() = TransactionManager.current().db.identifierManager.cutIfNecessaryAndQuote(name)
+    val identifier: String
+        get() = CoreTransactionManager.currentTransaction().db.identifierManager.cutIfNecessaryAndQuote(name)
 
     override fun toString(): String = "Sequence(identifier=$identifier)"
 

diff --git a/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/Table.kt b/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/Table.kt
@@ -9,7 +9,7 @@ import org.jetbrains.exposed.dao.id.IdTable
 import org.jetbrains.exposed.exceptions.DuplicateColumnException
 import org.jetbrains.exposed.sql.SqlExpressionBuilder.wrap
 import org.jetbrains.exposed.sql.statements.api.ExposedBlob
-import org.jetbrains.exposed.sql.transactions.TransactionManager
+import org.jetbrains.exposed.sql.transactions.CoreTransactionManager
 import org.jetbrains.exposed.sql.vendors.*
 import java.math.BigDecimal
 import java.util.*
@@ -1661,7 +1661,7 @@ open class Table(name: String = "") : ColumnSet(), DdlAware {
 
     internal fun primaryKeyConstraint(): String? {
         return primaryKey?.let { primaryKey ->
-            val tr = TransactionManager.current()
+            val tr = CoreTransactionManager.currentTransaction()
             val constraint = tr.db.identifierManager.cutIfNecessaryAndQuote(primaryKey.name)
             return primaryKey.columns
                 .joinToString(prefix = "CONSTRAINT $constraint PRIMARY KEY (", postfix = ")", transform = tr::identity)
@@ -1679,7 +1679,7 @@ open class Table(name: String = "") : ColumnSet(), DdlAware {
             if (currentDialect.supportsIfNotExists) {
                 append("IF NOT EXISTS ")
             }
-            append(TransactionManager.current().identity(this@Table))
+            append(CoreTransactionManager.currentTransaction().identity(this@Table))
 
             if (columns.isNotEmpty()) {
                 columns.joinTo(this, prefix = " (") { column ->
@@ -1769,7 +1769,7 @@ open class Table(name: String = "") : ColumnSet(), DdlAware {
             if (currentDialect.supportsIfNotExists) {
                 append("IF EXISTS ")
             }
-            append(TransactionManager.current().identity(this@Table))
+            append(CoreTransactionManager.currentTransaction().identity(this@Table))
             if (currentDialectIfAvailable is OracleDialect) {
                 append(" CASCADE CONSTRAINTS")
             } else if (currentDialectIfAvailable is PostgreSQLDialect && TableUtils.checkCycle(this@Table)) {

diff --git a/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/Transaction.kt b/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/Transaction.kt
@@ -1,7 +1,6 @@
 package org.jetbrains.exposed.sql
 
 import org.jetbrains.exposed.sql.transactions.TransactionInterface
-import org.jetbrains.exposed.sql.transactions.transactionManager
 import org.jetbrains.exposed.sql.vendors.inProperCase
 import java.util.*
 import java.util.concurrent.ConcurrentHashMap
@@ -51,25 +50,6 @@ abstract class Transaction : UserDataHolder(), TransactionInterface {
     /** Whether tracked values like [statementCount] and [duration] should be stored in [statementStats] for debugging. */
     var debug = false
 
-    /**
-     * The maximum amount of attempts that will be made to perform this `transaction` block.
-     *
-     * If this value is set to 1 and an SQLException happens, the exception will be thrown without performing a retry.
-     *
-     * @throws IllegalArgumentException If the amount of attempts is set to a value less than 1.
-     */
-    var maxAttempts: Int = db.transactionManager.defaultMaxAttempts
-        set(value) {
-            require(value > 0) { "maxAttempts must be set to perform at least 1 attempt." }
-            field = value
-        }
-
-    /** The minimum number of milliseconds to wait before retrying this `transaction` if an SQLException happens. */
-    var minRetryDelay: Long = db.transactionManager.defaultMinRetryDelay
-
-    /** The maximum number of milliseconds to wait before retrying this `transaction` if an SQLException happens. */
-    var maxRetryDelay: Long = db.transactionManager.defaultMaxRetryDelay
-
     /**
      * The number of seconds the JDBC driver should wait for a statement to execute in [Transaction] transaction before timing out.
      * Note Not all JDBC drivers implement this limit. Please check the driver documentation.

diff --git a/...sed-core/src/main/kotlin/org/jetbrains/exposed/sql/statements/BaseBatchInsertStatement.kt b/...sed-core/src/main/kotlin/org/jetbrains/exposed/sql/statements/BaseBatchInsertStatement.kt
@@ -6,7 +6,7 @@ import org.jetbrains.exposed.sql.InternalApi
 import org.jetbrains.exposed.sql.ResultRow
 import org.jetbrains.exposed.sql.Table
 import org.jetbrains.exposed.sql.isAutoInc
-import org.jetbrains.exposed.sql.transactions.TransactionManager
+import org.jetbrains.exposed.sql.transactions.CoreTransactionManager
 
 /**
  * Base class representing the SQL statement that batch inserts new rows into a table.
@@ -27,7 +27,7 @@ abstract class BaseBatchInsertStatement(
     override operator fun <S> set(column: Column<S>, value: S) {
         @OptIn(InternalApi::class)
         if (data.size > 1 && column !in data[data.size - 2] && !column.isDefaultable()) {
-            val fullIdentity = TransactionManager.current().fullIdentity(column)
+            val fullIdentity = CoreTransactionManager.currentTransaction().fullIdentity(column)
             throw BatchDataInconsistentException("Can't set $value for $fullIdentity because previous insertion can't be defaulted for that column.")
         }
         super.set(column, value)
@@ -66,7 +66,7 @@ abstract class BaseBatchInsertStatement(
 
     @InternalApi
     open fun validateLastBatch() {
-        val tr = TransactionManager.current()
+        val tr = CoreTransactionManager.currentTransaction()
         val cantBeDefaulted = (allColumnsInDataSet - values.keys).filterNot { it.isDefaultable() }
         if (cantBeDefaulted.isNotEmpty()) {
             val columnList = cantBeDefaulted.joinToString { tr.fullIdentity(it) }

diff --git a/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/statements/UpsertStatement.kt b/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/statements/UpsertStatement.kt
@@ -1,7 +1,7 @@
 package org.jetbrains.exposed.sql.statements
 
 import org.jetbrains.exposed.sql.*
-import org.jetbrains.exposed.sql.transactions.TransactionManager
+import org.jetbrains.exposed.sql.transactions.CoreTransactionManager
 import org.jetbrains.exposed.sql.vendors.*
 
 /**
@@ -94,7 +94,7 @@ sealed interface UpsertBuilder {
         override val columnType: IColumnType<T & Any>
     ) : ExpressionWithColumnType<T>() {
         override fun toQueryBuilder(queryBuilder: QueryBuilder) {
-            val transaction = TransactionManager.current()
+            val transaction = CoreTransactionManager.currentTransaction()
             val functionProvider = getFunctionProvider(transaction.db.dialect)
             functionProvider.insertValue(transaction.identity(column), queryBuilder)
         }

diff --git a/...sed-core/src/main/kotlin/org/jetbrains/exposed/sql/statements/api/IdentifierManagerApi.kt b/...sed-core/src/main/kotlin/org/jetbrains/exposed/sql/statements/api/IdentifierManagerApi.kt
@@ -1,6 +1,6 @@
 package org.jetbrains.exposed.sql.statements.api
 
-import org.jetbrains.exposed.sql.transactions.TransactionManager
+import org.jetbrains.exposed.sql.transactions.CoreTransactionManager
 import org.jetbrains.exposed.sql.vendors.ANSI_SQL_2003_KEYWORDS
 import org.jetbrains.exposed.sql.vendors.VENDORS_KEYWORDS
 import org.jetbrains.exposed.sql.vendors.currentDialect
@@ -76,7 +76,7 @@ abstract class IdentifierManagerApi {
         level = DeprecationLevel.WARNING
     )
     private val shouldPreserveKeywordCasing by lazy {
-        TransactionManager.currentOrNull()?.db?.config?.preserveKeywordCasing == true
+        CoreTransactionManager.currentTransactionOrNull()?.db?.config?.preserveKeywordCasing == true
     }
 
     /** Returns whether an SQL token should be wrapped in quotations and caches the returned value. */

diff --git a/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/transactions/TransactionManager.kt b/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/transactions/TransactionManager.kt
diff --git a/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/transactions/TransactionManagerApi.kt b/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/transactions/TransactionManagerApi.kt
@@ -0,0 +1,147 @@
+package org.jetbrains.exposed.sql.transactions
+
+import org.jetbrains.exposed.sql.DatabaseApi
+import org.jetbrains.exposed.sql.InternalApi
+import org.jetbrains.exposed.sql.Transaction
+import java.util.concurrent.ConcurrentHashMap
+import java.util.concurrent.ConcurrentLinkedDeque
+import java.util.concurrent.atomic.AtomicReference
+
+private object NotInitializedTransactionManager : TransactionManagerApi {
+    override var defaultIsolationLevel: Int = -1
+
+    override var defaultReadOnly: Boolean = false
+
+    override var defaultMaxAttempts: Int = -1
+
+    override var defaultMinRetryDelay: Long = 0
+
+    override var defaultMaxRetryDelay: Long = 0
+
+    override fun newTransaction(isolation: Int, readOnly: Boolean, outerTransaction: Transaction?): Transaction =
+        error("Please call Database.connect() before using this code")
+
+    override fun currentOrNull(): Transaction = error("Please call Database.connect() before using this code")
+
+    override fun bindTransactionToThread(transaction: Transaction?) {
+        error("Please call Database.connect() before using this code")
+    }
+}
+
+/**
+ * Represents the manager registered to a database, which is responsible for creating new transactions
+ * and storing data related to the database and its transactions.
+ */
+interface TransactionManagerApi {
+    /** The default transaction isolation level. Unless specified, the database-specific level will be used. */
+    var defaultIsolationLevel: Int
+
+    /** Whether transactions should be performed in read-only mode. Unless specified, the database default will be used. */
+    var defaultReadOnly: Boolean
+
+    /** The default maximum amount of attempts that will be made to perform a transaction. */
+    var defaultMaxAttempts: Int
+
+    /** The default minimum number of milliseconds to wait before retrying a transaction if an exception is thrown. */
+    var defaultMinRetryDelay: Long
+
+    /** The default maximum number of milliseconds to wait before retrying a transaction if an exception is thrown. */
+    var defaultMaxRetryDelay: Long
+
+    /**
+     * Returns a [Transaction] instance.
+     *
+     * The returned value may be a new transaction, or it may return the [outerTransaction] if called from within
+     * an existing transaction with the database not configured to `useNestedTransactions`.
+     */
+    fun newTransaction(
+        isolation: Int = defaultIsolationLevel,
+        readOnly: Boolean = defaultReadOnly,
+        outerTransaction: Transaction? = null
+    ): Transaction
+
+    /** Returns the current [Transaction], or `null` if none exists. */
+    fun currentOrNull(): Transaction?
+
+    /** Sets the current thread's copy of the manager's thread-local variable to the specified [transaction]. */
+    fun bindTransactionToThread(transaction: Transaction?)
+}
+
+/**
+ * Represents the object responsible for storing internal data related to each registered database
+ * and its transaction manager.
+ */
+@InternalApi
+object CoreTransactionManager {
+    private val databases = ConcurrentLinkedDeque<DatabaseApi>()
+
+    private val currentDefaultDatabase = AtomicReference<DatabaseApi>()
+
+    fun getDefaultDatabase(): DatabaseApi? = currentDefaultDatabase.get()
+
+    fun getDefaultDatabaseOrFirst(): DatabaseApi? = getDefaultDatabase() ?: databases.firstOrNull()
+
+    fun setDefaultDatabase(db: DatabaseApi?) { currentDefaultDatabase.set(db) }
+
+    private val registeredDatabases = ConcurrentHashMap<DatabaseApi, TransactionManagerApi>()
+
+    fun getDatabaseManager(db: DatabaseApi): TransactionManagerApi? = registeredDatabases[db]
+
+    private val currentThreadManager = TransactionManagerThreadLocal()
+
+    fun registerDatabaseManager(db: DatabaseApi, manager: TransactionManagerApi) {
+        if (getDefaultDatabaseOrFirst() == null) {
+            currentThreadManager.remove()
+        }
+        if (!registeredDatabases.containsKey(db)) {
+            databases.push(db)
+        }
+
+        registeredDatabases[db] = manager
+    }
+
+    fun closeAndUnregisterDatabase(db: DatabaseApi) {
+        val manager = getDatabaseManager(db)
+        manager?.let {
+            registeredDatabases.remove(db)
+            databases.remove(db)
+            currentDefaultDatabase.compareAndSet(db, null)
+            if (currentThreadManager.isInitialized && getCurrentThreadManager() == it) {
+                currentThreadManager.remove()
+            }
+        }
+    }
+
+    fun getCurrentThreadManager(): TransactionManagerApi = currentThreadManager.get()
+
+    fun resetCurrentThreadManager(manager: TransactionManagerApi?) {
+        manager?.let { currentThreadManager.set(it) } ?: currentThreadManager.remove()
+    }
+
+    fun currentTransactionOrNull(): Transaction? = getCurrentThreadManager().currentOrNull()
+
+    fun currentTransaction(): Transaction = currentTransactionOrNull() ?: error("No transaction in context.")
+
+    private class TransactionManagerThreadLocal : ThreadLocal<TransactionManagerApi>() {
+        var isInitialized = false
+
+        override fun get(): TransactionManagerApi {
+            return super.get()
+        }
+
+        override fun initialValue(): TransactionManagerApi {
+            isInitialized = true
+            return getDefaultDatabaseOrFirst()?.let { registeredDatabases.getValue(it) } ?: NotInitializedTransactionManager
+        }
+
+        override fun set(value: TransactionManagerApi?) {
+            isInitialized = true
+            super.set(value)
+        }
+
+        override fun remove() {
+            isInitialized = false
+            super.remove()
+        }
+    }
+}
diff --git a/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/transactions/TransactionScope.kt b/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/transactions/TransactionScope.kt
@@ -29,7 +29,7 @@ class TransactionStore<T : Any>(val init: (Transaction.() -> T)? = null) : ReadW
     private val key = Key<T>()
 
     override fun getValue(thisRef: Any?, property: KProperty<*>): T? {
-        val currentOrNullTransaction = TransactionManager.currentOrNull()
+        val currentOrNullTransaction = CoreTransactionManager.currentTransactionOrNull()
         return currentOrNullTransaction?.getUserData(key)
             ?: init?.let {
                 val value = currentOrNullTransaction?.it() ?: error("Can't init value outside the transaction")
@@ -39,7 +39,7 @@ class TransactionStore<T : Any>(val init: (Transaction.() -> T)? = null) : ReadW
     }
 
     override fun setValue(thisRef: Any?, property: KProperty<*>, value: T?) {
-        TransactionManager.currentOrNull()?.let {
+        CoreTransactionManager.currentTransactionOrNull()?.let {
             if (value == null) {
                 it.removeUserData(key)
             } else {

diff --git a/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/vendors/DatabaseDialect.kt b/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/vendors/DatabaseDialect.kt
@@ -1,7 +1,7 @@
 package org.jetbrains.exposed.sql.vendors
 
 import org.jetbrains.exposed.sql.*
-import org.jetbrains.exposed.sql.transactions.TransactionManager
+import org.jetbrains.exposed.sql.transactions.CoreTransactionManager
 
 /**
  * Common interface for all database dialects.
@@ -142,14 +142,14 @@ internal fun <T> withDialect(dialect: DatabaseDialect, body: () -> T): T {
 }
 
 /** Returns the dialect used in the current transaction, may throw an exception if there is no current transaction. */
-val currentDialect: DatabaseDialect get() = explicitDialect.get() ?: TransactionManager.current().db.dialect
+val currentDialect: DatabaseDialect get() = explicitDialect.get() ?: CoreTransactionManager.currentTransaction().db.dialect
 
 internal val currentDialectIfAvailable: DatabaseDialect?
-    get() = if (TransactionManager.isInitialized() && TransactionManager.currentOrNull() != null) {
+    get() = if (CoreTransactionManager.getDefaultDatabaseOrFirst() != null && CoreTransactionManager.currentTransactionOrNull() != null) {
         currentDialect
     } else {
         null
     }
 
 internal fun String.inProperCase(): String =
-    TransactionManager.currentOrNull()?.db?.identifierManager?.inProperCase(this@inProperCase) ?: this
+    CoreTransactionManager.currentTransactionOrNull()?.db?.identifierManager?.inProperCase(this@inProperCase) ?: this
diff --git a/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/vendors/H2.kt b/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/vendors/H2.kt
@@ -3,7 +3,7 @@ package org.jetbrains.exposed.sql.vendors
 import org.jetbrains.exposed.exceptions.throwUnsupportedException
 import org.jetbrains.exposed.sql.*
 import org.jetbrains.exposed.sql.statements.StatementType
-import org.jetbrains.exposed.sql.transactions.TransactionManager
+import org.jetbrains.exposed.sql.transactions.CoreTransactionManager
 import java.util.*
 
 internal object H2DataTypeProvider : DataTypeProvider() {
@@ -29,7 +29,7 @@ internal object H2FunctionProvider : FunctionProvider() {
         get() = h2Mode == H2Dialect.H2CompatibilityMode.Oracle
 
     override fun nextVal(seq: Sequence, builder: QueryBuilder) =
-        when ((TransactionManager.current().db.dialect as H2Dialect).majorVersion) {
+        when ((CoreTransactionManager.currentTransaction().db.dialect as H2Dialect).majorVersion) {
             H2Dialect.H2MajorVersion.One -> super.nextVal(seq, builder)
             H2Dialect.H2MajorVersion.Two -> builder {
                 append("NEXT VALUE FOR ${seq.identifier}")
@@ -203,7 +203,7 @@ open class H2Dialect : VendorDialect(dialectName, H2DataTypeProvider, H2Function
     }
 
     internal val version by lazy {
-        exactH2Version(TransactionManager.current())
+        exactH2Version(CoreTransactionManager.currentTransaction())
     }
 
     val majorVersion: H2MajorVersion by lazy {
@@ -263,7 +263,7 @@ open class H2Dialect : VendorDialect(dialectName, H2DataTypeProvider, H2Function
 
     /** The H2 database compatibility mode retrieved from metadata. */
     val h2Mode: H2CompatibilityMode? by lazy {
-        val modeValue = TransactionManager.current().db.dialectMode
+        val modeValue = CoreTransactionManager.currentTransaction().db.dialectMode
         when {
             modeValue == null -> null
             modeValue.equals("MySQL", ignoreCase = true) -> H2CompatibilityMode.MySQL

diff --git a/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/vendors/MariaDBDialect.kt b/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/vendors/MariaDBDialect.kt
@@ -3,7 +3,7 @@ package org.jetbrains.exposed.sql.vendors
 import org.jetbrains.exposed.exceptions.UnsupportedByDialectException
 import org.jetbrains.exposed.sql.*
 import org.jetbrains.exposed.sql.Function
-import org.jetbrains.exposed.sql.transactions.TransactionManager
+import org.jetbrains.exposed.sql.transactions.CoreTransactionManager
 
 internal object MariaDBDataTypeProvider : MysqlDataTypeProvider() {
     override fun timestampType(): String = if ((currentDialect as? MariaDBDialect)?.isFractionDateTimeSupported() == true) "TIMESTAMP(6)" else "TIMESTAMP"
@@ -91,15 +91,15 @@ open class MariaDBDialect : MysqlDialect() {
     override val supportsOnlyIdentifiersInGeneratedKeys: Boolean = true
     override val supportsSetDefaultReferenceOption: Boolean = false
     override val supportsCreateSequence: Boolean by lazy {
-        TransactionManager.current().db.isVersionCovers(SEQUENCE_MIN_MAJOR_VERSION, SEQUENCE_MIN_MINOR_VERSION)
+        CoreTransactionManager.currentTransaction().db.isVersionCovers(SEQUENCE_MIN_MAJOR_VERSION, SEQUENCE_MIN_MINOR_VERSION)
     }
 
     // actually MariaDb supports it but jdbc driver prepares statement without RETURNING clause
     override val supportsSequenceAsGeneratedKeys: Boolean = false
 
     @Suppress("MagicNumber")
     override val sequenceMaxValue: Long by lazy {
-        if (TransactionManager.current().db.isVersionCovers(11, 5)) {
+        if (CoreTransactionManager.currentTransaction().db.isVersionCovers(11, 5)) {
             super.sequenceMaxValue
         } else {
             Long.MAX_VALUE - 1
@@ -108,7 +108,7 @@ open class MariaDBDialect : MysqlDialect() {
 
     /** Returns `true` if the MariaDB database version is greater than or equal to 5.3. */
     @Suppress("MagicNumber")
-    override fun isFractionDateTimeSupported(): Boolean = TransactionManager.current().db.isVersionCovers(5, 3)
+    override fun isFractionDateTimeSupported(): Boolean = CoreTransactionManager.currentTransaction().db.isVersionCovers(5, 3)
 
     override fun isTimeZoneOffsetSupported(): Boolean = false
 

diff --git a/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/vendors/MysqlDialect.kt b/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/vendors/MysqlDialect.kt
@@ -3,7 +3,7 @@ package org.jetbrains.exposed.sql.vendors
 import org.jetbrains.exposed.exceptions.UnsupportedByDialectException
 import org.jetbrains.exposed.exceptions.throwUnsupportedException
 import org.jetbrains.exposed.sql.*
-import org.jetbrains.exposed.sql.transactions.TransactionManager
+import org.jetbrains.exposed.sql.transactions.CoreTransactionManager
 import java.math.BigDecimal
 
 internal open class MysqlDataTypeProvider : DataTypeProvider() {
@@ -166,7 +166,7 @@ internal open class MysqlFunctionProvider : FunctionProvider() {
     ) {
         val oneOrAll = optional?.lowercase()
         if (oneOrAll != "one" && oneOrAll != "all") {
-            TransactionManager.current().throwUnsupportedException("MySQL requires a single optional argument: 'one' or 'all'")
+            CoreTransactionManager.currentTransaction().throwUnsupportedException("MySQL requires a single optional argument: 'one' or 'all'")
         }
         queryBuilder {
             append("JSON_CONTAINS_PATH(", expression, ", ")
@@ -320,7 +320,7 @@ internal open class MysqlFunctionProvider : FunctionProvider() {
 
     override fun queryLimitAndOffset(size: Int?, offset: Long, alreadyOrdered: Boolean): String {
         if (size == null && offset > 0) {
-            TransactionManager.current().throwUnsupportedException(
+            CoreTransactionManager.currentTransaction().throwUnsupportedException(
                 "${currentDialect.name} doesn't support OFFSET clause without LIMIT"
             )
         }
@@ -338,11 +338,11 @@ internal open class MysqlFunctionProvider : FunctionProvider() {
 open class MysqlDialect : VendorDialect(dialectName, MysqlDataTypeProvider.INSTANCE, MysqlFunctionProvider.INSTANCE) {
 
     internal val isMysql8: Boolean by lazy {
-        TransactionManager.current().db.isVersionCovers(BigDecimal("8.0"))
+        CoreTransactionManager.currentTransaction().db.isVersionCovers(BigDecimal("8.0"))
     }
 
     internal val fullVersion: String by lazy {
-        TransactionManager.current().db.fullVersion
+        CoreTransactionManager.currentTransaction().db.fullVersion
     }
 
     override val supportsCreateSequence: Boolean = false
@@ -357,7 +357,7 @@ open class MysqlDialect : VendorDialect(dialectName, MysqlDataTypeProvider.INSTA
 
     /** Returns `true` if the MySQL database version is greater than or equal to 5.6. */
     @Suppress("MagicNumber")
-    open fun isFractionDateTimeSupported(): Boolean = TransactionManager.current().db.isVersionCovers(5, 6)
+    open fun isFractionDateTimeSupported(): Boolean = CoreTransactionManager.currentTransaction().db.isVersionCovers(5, 6)
 
     /** Returns `true` if a MySQL database is being used and its version is greater than or equal to 8.0. */
     open fun isTimeZoneOffsetSupported(): Boolean = isMysql8

diff --git a/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/vendors/OracleDialect.kt b/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/vendors/OracleDialect.kt
@@ -7,7 +7,7 @@ import org.jetbrains.exposed.sql.statements.MergeStatement.ClauseAction.DELETE
 import org.jetbrains.exposed.sql.statements.MergeStatement.ClauseAction.INSERT
 import org.jetbrains.exposed.sql.statements.MergeStatement.ClauseAction.UPDATE
 import org.jetbrains.exposed.sql.statements.StatementType
-import org.jetbrains.exposed.sql.transactions.TransactionManager
+import org.jetbrains.exposed.sql.transactions.CoreTransactionManager
 import java.util.*
 
 @Suppress("TooManyFunctions")
@@ -127,7 +127,7 @@ internal object OracleFunctionProvider : FunctionProvider() {
         expr: GroupConcat<T>,
         queryBuilder: QueryBuilder
     ): Unit = queryBuilder {
-        val tr = TransactionManager.current()
+        val tr = CoreTransactionManager.currentTransaction()
         if (expr.distinct) tr.throwUnsupportedException("Oracle doesn't support DISTINCT in LISTAGG")
         if (expr.orderBy.size > 1) {
             tr.throwUnsupportedException("Oracle supports only single column in ORDER BY clause in LISTAGG")
@@ -203,7 +203,7 @@ internal object OracleFunctionProvider : FunctionProvider() {
         queryBuilder: QueryBuilder
     ) {
         if (path.size > 1) {
-            TransactionManager.current().throwUnsupportedException("Oracle does not support multiple JSON path arguments")
+            CoreTransactionManager.currentTransaction().throwUnsupportedException("Oracle does not support multiple JSON path arguments")
         }
         queryBuilder {
             append(if (toScalar) "JSON_VALUE" else "JSON_QUERY")
@@ -221,7 +221,7 @@ internal object OracleFunctionProvider : FunctionProvider() {
         queryBuilder: QueryBuilder
     ) {
         if (path.size > 1) {
-            TransactionManager.current().throwUnsupportedException("Oracle does not support multiple JSON path arguments")
+            CoreTransactionManager.currentTransaction().throwUnsupportedException("Oracle does not support multiple JSON path arguments")
         }
         queryBuilder {
             append("JSON_EXISTS(", expression, ", ")
@@ -264,7 +264,7 @@ internal object OracleFunctionProvider : FunctionProvider() {
         +"SELECT "
         columnsToSelect.values.appendTo { +it }
         +" FROM "
-        targets.describe(TransactionManager.current(), this)
+        targets.describe(CoreTransactionManager.currentTransaction(), this)
         where?.let {
             +" WHERE "
             +it
@@ -350,7 +350,7 @@ internal object OracleFunctionProvider : FunctionProvider() {
             +"DELETE (SELECT "
             tableToDelete.columns.appendTo { +it }
             +" FROM "
-            targets.describe(TransactionManager.current(), this)
+            targets.describe(CoreTransactionManager.currentTransaction(), this)
             where?.let {
                 +" WHERE "
                 +it

diff --git a/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/vendors/PostgreSQL.kt b/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/vendors/PostgreSQL.kt
@@ -3,7 +3,7 @@ package org.jetbrains.exposed.sql.vendors
 import org.jetbrains.exposed.exceptions.throwUnsupportedException
 import org.jetbrains.exposed.sql.*
 import org.jetbrains.exposed.sql.statements.StatementType
-import org.jetbrains.exposed.sql.transactions.TransactionManager
+import org.jetbrains.exposed.sql.transactions.CoreTransactionManager
 import java.util.*
 
 internal object PostgreSQLDataTypeProvider : DataTypeProvider() {
@@ -54,7 +54,7 @@ internal object PostgreSQLFunctionProvider : FunctionProvider() {
     }
 
     override fun <T : String?> groupConcat(expr: GroupConcat<T>, queryBuilder: QueryBuilder) {
-        val tr = TransactionManager.current()
+        val tr = CoreTransactionManager.currentTransaction()
         return when (expr.separator) {
             null -> tr.throwUnsupportedException("PostgreSQL requires explicit separator in STRING_AGG function.")
             else -> queryBuilder {
@@ -174,7 +174,7 @@ internal object PostgreSQLFunctionProvider : FunctionProvider() {
         queryBuilder: QueryBuilder
     ) {
         path?.let {
-            TransactionManager.current().throwUnsupportedException("PostgreSQL does not support a JSON path argument")
+            CoreTransactionManager.currentTransaction().throwUnsupportedException("PostgreSQL does not support a JSON path argument")
         }
         val isNotJsonB = !(jsonType as JsonColumnMarker).usesBinaryFormat
         queryBuilder {
@@ -193,7 +193,7 @@ internal object PostgreSQLFunctionProvider : FunctionProvider() {
         queryBuilder: QueryBuilder
     ) {
         if (path.size > 1) {
-            TransactionManager.current().throwUnsupportedException("PostgreSQL does not support multiple JSON path arguments")
+            CoreTransactionManager.currentTransaction().throwUnsupportedException("PostgreSQL does not support multiple JSON path arguments")
         }
         val isNotJsonB = !(jsonType as JsonColumnMarker).usesBinaryFormat
         queryBuilder {
@@ -369,7 +369,7 @@ open class PostgreSQLDialect(override val name: String = dialectName) : VendorDi
     override fun modifyColumn(column: Column<*>, columnDiff: ColumnDiff): List<String> {
         val list = mutableListOf(
             buildString {
-                val tr = TransactionManager.current()
+                val tr = CoreTransactionManager.currentTransaction()
                 append("ALTER TABLE ${tr.identity(column.table)} ")
                 val colName = tr.identity(column)
 

diff --git a/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/vendors/SQLServerDialect.kt b/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/vendors/SQLServerDialect.kt
@@ -7,7 +7,7 @@ import org.jetbrains.exposed.sql.statements.MergeStatement.ClauseAction.DELETE
 import org.jetbrains.exposed.sql.statements.MergeStatement.ClauseAction.INSERT
 import org.jetbrains.exposed.sql.statements.MergeStatement.ClauseAction.UPDATE
 import org.jetbrains.exposed.sql.statements.StatementType
-import org.jetbrains.exposed.sql.transactions.TransactionManager
+import org.jetbrains.exposed.sql.transactions.CoreTransactionManager
 import java.util.*
 
 internal object SQLServerDataTypeProvider : DataTypeProvider() {
@@ -93,7 +93,7 @@ internal object SQLServerFunctionProvider : FunctionProvider() {
     }
 
     override fun <T : String?> groupConcat(expr: GroupConcat<T>, queryBuilder: QueryBuilder) {
-        val tr = TransactionManager.current()
+        val tr = CoreTransactionManager.currentTransaction()
         return when {
             expr.separator == null -> tr.throwUnsupportedException("SQL Server requires explicit separator in STRING_AGG")
             expr.distinct -> tr.throwUnsupportedException("SQL Server doesn't support DISTINCT in STRING_AGG")
@@ -122,7 +122,7 @@ internal object SQLServerFunctionProvider : FunctionProvider() {
         pattern: Expression<String>,
         caseSensitive: Boolean,
         queryBuilder: QueryBuilder
-    ): Unit = TransactionManager.current().throwUnsupportedException("SQLServer doesn't provide built in REGEXP expression, use LIKE instead.")
+    ): Unit = CoreTransactionManager.currentTransaction().throwUnsupportedException("SQLServer doesn't provide built in REGEXP expression, use LIKE instead.")
 
     override fun <T> date(expr: Expression<T>, queryBuilder: QueryBuilder) = queryBuilder {
         append("CAST(", expr, " AS DATE)")
@@ -180,7 +180,7 @@ internal object SQLServerFunctionProvider : FunctionProvider() {
         queryBuilder: QueryBuilder
     ) {
         if (path.size > 1) {
-            TransactionManager.current().throwUnsupportedException("SQLServer does not support multiple JSON path arguments")
+            CoreTransactionManager.currentTransaction().throwUnsupportedException("SQLServer does not support multiple JSON path arguments")
         }
         queryBuilder {
             append(if (toScalar) "JSON_VALUE" else "JSON_QUERY")
@@ -358,7 +358,7 @@ open class SQLServerDialect : VendorDialect(dialectName, SQLServerDataTypeProvid
     }
 
     override fun modifyColumn(column: Column<*>, columnDiff: ColumnDiff): List<String> {
-        val transaction = TransactionManager.current()
+        val transaction = CoreTransactionManager.currentTransaction()
 
         val alterTablePart = "ALTER TABLE ${transaction.identity(column.table)} "
 

diff --git a/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/vendors/SQLiteDialect.kt b/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/vendors/SQLiteDialect.kt
@@ -2,7 +2,7 @@ package org.jetbrains.exposed.sql.vendors
 
 import org.jetbrains.exposed.exceptions.throwUnsupportedException
 import org.jetbrains.exposed.sql.*
-import org.jetbrains.exposed.sql.transactions.TransactionManager
+import org.jetbrains.exposed.sql.transactions.CoreTransactionManager
 
 internal object SQLiteDataTypeProvider : DataTypeProvider() {
     override fun integerAutoincType(): String = "INTEGER PRIMARY KEY AUTOINCREMENT"
@@ -42,7 +42,7 @@ internal object SQLiteFunctionProvider : FunctionProvider() {
 
     override fun <T : String?> groupConcat(expr: GroupConcat<T>, queryBuilder: QueryBuilder) {
         if (expr.distinct) {
-            TransactionManager.current().throwUnsupportedException("SQLite doesn't support DISTINCT in GROUP_CONCAT function")
+            CoreTransactionManager.currentTransaction().throwUnsupportedException("SQLite doesn't support DISTINCT in GROUP_CONCAT function")
         }
         queryBuilder {
             +"GROUP_CONCAT("
@@ -76,7 +76,7 @@ internal object SQLiteFunctionProvider : FunctionProvider() {
         pattern: Expression<String>,
         caseSensitive: Boolean,
         queryBuilder: QueryBuilder
-    ): Unit = TransactionManager.current().throwUnsupportedException("SQLite doesn't provide built in REGEXP expression, use LIKE instead.")
+    ): Unit = CoreTransactionManager.currentTransaction().throwUnsupportedException("SQLite doesn't provide built in REGEXP expression, use LIKE instead.")
 
     override fun <T> time(expr: Expression<T>, queryBuilder: QueryBuilder) = queryBuilder {
         append(
@@ -135,22 +135,22 @@ internal object SQLiteFunctionProvider : FunctionProvider() {
     override fun <T> stdDevPop(
         expression: Expression<T>,
         queryBuilder: QueryBuilder
-    ): Unit = TransactionManager.current().throwUnsupportedException("$UNSUPPORTED_AGGREGATE STDDEV_POP")
+    ): Unit = CoreTransactionManager.currentTransaction().throwUnsupportedException("$UNSUPPORTED_AGGREGATE STDDEV_POP")
 
     override fun <T> stdDevSamp(
         expression: Expression<T>,
         queryBuilder: QueryBuilder
-    ): Unit = TransactionManager.current().throwUnsupportedException("$UNSUPPORTED_AGGREGATE STDDEV_SAMP")
+    ): Unit = CoreTransactionManager.currentTransaction().throwUnsupportedException("$UNSUPPORTED_AGGREGATE STDDEV_SAMP")
 
     override fun <T> varPop(
         expression: Expression<T>,
         queryBuilder: QueryBuilder
-    ): Unit = TransactionManager.current().throwUnsupportedException("$UNSUPPORTED_AGGREGATE VAR_POP")
+    ): Unit = CoreTransactionManager.currentTransaction().throwUnsupportedException("$UNSUPPORTED_AGGREGATE VAR_POP")
 
     override fun <T> varSamp(
         expression: Expression<T>,
         queryBuilder: QueryBuilder
-    ): Unit = TransactionManager.current().throwUnsupportedException("$UNSUPPORTED_AGGREGATE VAR_SAMP")
+    ): Unit = CoreTransactionManager.currentTransaction().throwUnsupportedException("$UNSUPPORTED_AGGREGATE VAR_SAMP")
 
     override fun <T> jsonExtract(
         expression: Expression<T>,
@@ -171,7 +171,7 @@ internal object SQLiteFunctionProvider : FunctionProvider() {
         jsonType: IColumnType<*>,
         queryBuilder: QueryBuilder
     ) {
-        val transaction = TransactionManager.current()
+        val transaction = CoreTransactionManager.currentTransaction()
         if (path.size > 1) {
             transaction.throwUnsupportedException("SQLite does not support multiple JSON path arguments")
         }
@@ -242,7 +242,7 @@ internal object SQLiteFunctionProvider : FunctionProvider() {
 
     override fun queryLimitAndOffset(size: Int?, offset: Long, alreadyOrdered: Boolean): String {
         if (size == null && offset > 0) {
-            TransactionManager.current().throwUnsupportedException("SQLite doesn't support OFFSET clause without LIMIT")
+            CoreTransactionManager.currentTransaction().throwUnsupportedException("SQLite doesn't support OFFSET clause without LIMIT")
         }
         return super.queryLimitAndOffset(size, offset, alreadyOrdered)
     }

diff --git a/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/vendors/VendorDialect.kt b/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/vendors/VendorDialect.kt
@@ -2,7 +2,7 @@ package org.jetbrains.exposed.sql.vendors
 
 import org.jetbrains.exposed.sql.*
 import org.jetbrains.exposed.sql.Function
-import org.jetbrains.exposed.sql.transactions.TransactionManager
+import org.jetbrains.exposed.sql.transactions.CoreTransactionManager
 
 /**
  * Base implementation of a vendor dialect
@@ -14,7 +14,7 @@ abstract class VendorDialect(
 ) : DatabaseDialect {
 
     protected val identifierManager
-        get() = TransactionManager.current().db.identifierManager
+        get() = CoreTransactionManager.currentTransaction().db.identifierManager
 
     @Suppress("UnnecessaryAbstractClass")
     abstract class DialectNameProvider(val dialectName: String)
@@ -56,7 +56,7 @@ abstract class VendorDialect(
      * Unique indexes can be partial
      */
     override fun createIndex(index: Index): String {
-        val t = TransactionManager.current()
+        val t = CoreTransactionManager.currentTransaction()
         val quotedTableName = t.identity(index.table)
         val quotedIndexName = t.db.identifierManager.cutIfNecessaryAndQuote(index.indexName)
         val keyFields = index.columns.plus(index.functions ?: emptyList())
@@ -107,10 +107,10 @@ 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)}")
+        listOf("ALTER TABLE ${CoreTransactionManager.currentTransaction().identity(column.table)} MODIFY COLUMN ${column.descriptionDdl(true)}")
 
     override fun addPrimaryKey(table: Table, pkName: String?, vararg pkColumns: Column<*>): String {
-        val transaction = TransactionManager.current()
+        val transaction = CoreTransactionManager.currentTransaction()
         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"

diff --git a/exposed-dao/src/main/kotlin/org/jetbrains/exposed/dao/Entity.kt b/exposed-dao/src/main/kotlin/org/jetbrains/exposed/dao/Entity.kt
@@ -92,7 +92,7 @@ open class Entity<ID : Any>(val id: EntityID<ID>) {
      * @sample org.jetbrains.exposed.sql.tests.shared.entities.EntityTests.testNewWithIdAndRefresh
      */
     open fun refresh(flush: Boolean = false) {
-        val transaction = TransactionManager.current() as JdbcTransaction
+        val transaction = TransactionManager.current()
         val cache = transaction.entityCache
         val isNewEntity = isNewEntity()
         when {

diff --git a/exposed-dao/src/main/kotlin/org/jetbrains/exposed/dao/EntityBatchUpdate.kt b/exposed-dao/src/main/kotlin/org/jetbrains/exposed/dao/EntityBatchUpdate.kt
@@ -2,6 +2,7 @@ package org.jetbrains.exposed.dao
 
 import org.jetbrains.exposed.dao.id.EntityID
 import org.jetbrains.exposed.sql.Column
+import org.jetbrains.exposed.sql.JdbcTransaction
 import org.jetbrains.exposed.sql.Transaction
 import org.jetbrains.exposed.sql.statements.BatchUpdateExecutable
 import org.jetbrains.exposed.sql.statements.BatchUpdateStatement
@@ -54,7 +55,7 @@ class EntityBatchUpdate(private val klass: EntityClass<*, Entity<*>>) {
         return updateSets.values.fold(0) { acc, set ->
             acc + BatchUpdateExecutable(BatchUpdateStatement(klass.table)).let {
                 it.statement.data.addAll(set)
-                it.execute(transaction)!!
+                it.execute(transaction as JdbcTransaction)!!
             }
         }
     }

diff --git a/exposed-dao/src/main/kotlin/org/jetbrains/exposed/dao/EntityClass.kt b/exposed-dao/src/main/kotlin/org/jetbrains/exposed/dao/EntityClass.kt
@@ -344,7 +344,7 @@ abstract class EntityClass<ID : Any, out T : Entity<ID>>(
      * [EntityCache], creates a new instance using the data in [row].
      */
     fun wrap(id: EntityID<ID>, row: ResultRow?): T {
-        val transaction = TransactionManager.current() as JdbcTransaction
+        val transaction = TransactionManager.current()
         return transaction.entityCache.find(this, id) ?: createInstance(id, row).also { new ->
             new.klass = this
             new.db = transaction.db
@@ -378,7 +378,7 @@ abstract class EntityClass<ID : Any, out T : Entity<ID>>(
         val entityCache = warmCache()
         val prototype: T = createInstance(entityId, null)
         prototype.klass = this
-        prototype.db = (TransactionManager.current() as JdbcTransaction).db
+        prototype.db = TransactionManager.current().db
         prototype._readValues = ResultRow.createAndFillDefaults(dependsOnColumns)
         if (entityId._value != null) {
             prototype.writeIdColumnValue(table, entityId)
@@ -1130,7 +1130,7 @@ abstract class ImmutableCachedEntityClass<ID : Any, out T : Entity<ID>>(
     }
 
     final override fun warmCache(): EntityCache {
-        val tr = TransactionManager.current() as JdbcTransaction
+        val tr = TransactionManager.current()
         val db = tr.db
         val transactionCache = super.warmCache()
         if (_cachedValues[db] == null) {

diff --git a/exposed-dao/src/main/kotlin/org/jetbrains/exposed/dao/EntityHook.kt b/exposed-dao/src/main/kotlin/org/jetbrains/exposed/dao/EntityHook.kt
@@ -1,7 +1,6 @@
 package org.jetbrains.exposed.dao
 
 import org.jetbrains.exposed.dao.id.EntityID
-import org.jetbrains.exposed.sql.JdbcTransaction
 import org.jetbrains.exposed.sql.Transaction
 import org.jetbrains.exposed.sql.transactions.TransactionManager
 import org.jetbrains.exposed.sql.transactions.transactionScope
@@ -121,7 +120,7 @@ fun <T> withHook(action: (EntityChange) -> Unit, body: () -> T): T {
     EntityHook.subscribe(action)
     try {
         return body().apply {
-            (TransactionManager.current() as JdbcTransaction).commit()
+            TransactionManager.current().commit()
         }
     } finally {
         EntityHook.unsubscribe(action)

diff --git a/exposed-jdbc/api/exposed-jdbc.api b/exposed-jdbc/api/exposed-jdbc.api
diff --git a/exposed-jdbc/src/main/kotlin/org/jetbrains/exposed/sql/Database.kt b/exposed-jdbc/src/main/kotlin/org/jetbrains/exposed/sql/Database.kt
@@ -3,8 +3,9 @@ package org.jetbrains.exposed.sql
 import org.jetbrains.exposed.sql.statements.api.ExposedConnection
 import org.jetbrains.exposed.sql.statements.api.IdentifierManagerApi
 import org.jetbrains.exposed.sql.statements.api.JdbcExposedDatabaseMetadata
-import org.jetbrains.exposed.sql.transactions.ThreadLocalTransactionManager
+import org.jetbrains.exposed.sql.transactions.CoreTransactionManager
 import org.jetbrains.exposed.sql.transactions.TransactionManager
+import org.jetbrains.exposed.sql.transactions.TransactionManagerApi
 import org.jetbrains.exposed.sql.vendors.*
 import java.math.BigDecimal
 import java.sql.Connection
@@ -22,7 +23,7 @@ class Database private constructor(
     val connector: () -> ExposedConnection<*>
 ) : DatabaseApi(resolvedVendor, config) {
     internal fun <T> metadata(body: JdbcExposedDatabaseMetadata.() -> T): T {
-        val transaction = TransactionManager.currentOrNull() as? JdbcTransaction
+        val transaction = TransactionManager.currentOrNull()
         return if (transaction == null) {
             val connection = connector()
             try {
@@ -165,18 +166,21 @@ class Database private constructor(
             dialectMapping[prefix] = dialect
         }
 
+        @OptIn(InternalApi::class)
         private fun doConnect(
             explicitVendor: String?,
             config: DatabaseConfig?,
             connectionAutoRegistration: DatabaseConnectionAutoRegistration,
             getNewConnection: () -> Connection,
             setupConnection: (Connection) -> Unit = {},
-            manager: (Database) -> TransactionManager = { ThreadLocalTransactionManager(it) }
+            manager: (Database) -> TransactionManagerApi = { TransactionManager(it) }
         ): Database {
             return Database(explicitVendor, config ?: DatabaseConfig.invoke()) {
                 connectionAutoRegistration(getNewConnection().apply { setupConnection(this) })
             }.apply {
-                TransactionManager.registerManager(this, manager(this))
+                CoreTransactionManager.registerDatabaseManager(this, manager(this))
+                // ABOVE should be replaced with BELOW when ThreadLocalTransactionManager is fully deprecated
+                // TransactionManager.registerManager(this, manager(this))
             }
         }
 
@@ -198,7 +202,7 @@ class Database private constructor(
             setupConnection: (Connection) -> Unit = {},
             databaseConfig: DatabaseConfig? = null,
             connectionAutoRegistration: DatabaseConnectionAutoRegistration = connectionInstanceImpl,
-            manager: (Database) -> TransactionManager = { ThreadLocalTransactionManager(it) }
+            manager: (Database) -> TransactionManagerApi = { TransactionManager(it) }
         ): Database {
             return doConnect(
                 explicitVendor = null,
@@ -228,7 +232,7 @@ class Database private constructor(
             getNewConnection: () -> Connection,
             databaseConfig: DatabaseConfig? = null,
             connectionAutoRegistration: DatabaseConnectionAutoRegistration = connectionInstanceImpl,
-            manager: (Database) -> TransactionManager = { ThreadLocalTransactionManager(it) }
+            manager: (Database) -> TransactionManagerApi = { TransactionManager(it) }
         ): Database {
             return doConnect(
                 explicitVendor = null,
@@ -265,7 +269,7 @@ class Database private constructor(
             setupConnection: (Connection) -> Unit = {},
             databaseConfig: DatabaseConfig? = null,
             connectionAutoRegistration: DatabaseConnectionAutoRegistration = connectionInstanceImpl,
-            manager: (Database) -> TransactionManager = { ThreadLocalTransactionManager(it) }
+            manager: (Database) -> TransactionManagerApi = { TransactionManager(it) }
         ): Database {
             Class.forName(driver).getDeclaredConstructor().newInstance()
             val dialectName = getDialectName(url) ?: error("Can't resolve dialect for connection: $url")

diff --git a/exposed-jdbc/src/main/kotlin/org/jetbrains/exposed/sql/ExplainExecutable.kt b/exposed-jdbc/src/main/kotlin/org/jetbrains/exposed/sql/ExplainExecutable.kt
@@ -17,7 +17,7 @@ open class ExplainExecutable(
     override fun JdbcPreparedStatementApi.executeInternal(transaction: JdbcTransaction): JdbcResult = executeQuery()
 
     override fun iterator(): Iterator<ExplainResultRow> {
-        val rs = (TransactionManager.current() as JdbcTransaction).exec(this)!! as JdbcResult
+        val rs = TransactionManager.current().exec(this)!! as JdbcResult
         val resultIterator = ResultIterator(rs.result)
         return Iterable { resultIterator }.iterator()
     }

diff --git a/exposed-jdbc/src/main/kotlin/org/jetbrains/exposed/sql/JdbcTransaction.kt b/exposed-jdbc/src/main/kotlin/org/jetbrains/exposed/sql/JdbcTransaction.kt
@@ -6,6 +6,7 @@ import org.jetbrains.exposed.sql.statements.*
 import org.jetbrains.exposed.sql.statements.api.JdbcPreparedStatementApi
 import org.jetbrains.exposed.sql.statements.jdbc.JdbcResult
 import org.jetbrains.exposed.sql.transactions.JdbcTransactionInterface
+import org.jetbrains.exposed.sql.transactions.transactionManager
 import java.sql.ResultSet
 import java.util.*
 import java.util.concurrent.TimeUnit
@@ -15,6 +16,25 @@ open class JdbcTransaction(
 ) : Transaction(), JdbcTransactionInterface by transactionImpl {
     final override val db: Database = transactionImpl.db
 
+    /**
+     * The maximum amount of attempts that will be made to perform this `transaction` block.
+     *
+     * If this value is set to 1 and an SQLException happens, the exception will be thrown without performing a retry.
+     *
+     * @throws IllegalArgumentException If the amount of attempts is set to a value less than 1.
+     */
+    var maxAttempts: Int = db.transactionManager.defaultMaxAttempts
+        set(value) {
+            require(value > 0) { "maxAttempts must be set to perform at least 1 attempt." }
+            field = value
+        }
+
+    /** The minimum number of milliseconds to wait before retrying this `transaction` if an SQLException happens. */
+    var minRetryDelay: Long = db.transactionManager.defaultMinRetryDelay
+
+    /** The maximum number of milliseconds to wait before retrying this `transaction` if an SQLException happens. */
+    var maxRetryDelay: Long = db.transactionManager.defaultMaxRetryDelay
+
     /** The currently executing statement. */
     var currentStatement: JdbcPreparedStatementApi? = null
 

diff --git a/exposed-jdbc/src/main/kotlin/org/jetbrains/exposed/sql/Query.kt b/exposed-jdbc/src/main/kotlin/org/jetbrains/exposed/sql/Query.kt
@@ -28,7 +28,7 @@ open class Query(
     }
 
     protected val transaction: JdbcTransaction
-        get() = TransactionManager.current() as JdbcTransaction
+        get() = TransactionManager.current()
 
     /** Creates a new [Query] instance using all stored properties of this `SELECT` query. */
     override fun copy(): Query = Query(set, where).also { copy ->

diff --git a/exposed-jdbc/src/main/kotlin/org/jetbrains/exposed/sql/SchemaUtils.kt b/exposed-jdbc/src/main/kotlin/org/jetbrains/exposed/sql/SchemaUtils.kt
@@ -43,16 +43,15 @@ object SchemaUtils : SchemaUtilityApi() {
 
     /** Creates the provided sequences, using a batch execution if [inBatch] is set to `true`. */
     fun createSequence(vararg seq: Sequence, inBatch: Boolean = false) {
-        // NEED to potentially adjust this as the TransactionManager interface level
-        with(TransactionManager.current() as JdbcTransaction) {
+        with(TransactionManager.current()) {
             val createStatements = seq.flatMap { it.createStatement() }
             execStatements(inBatch, createStatements)
         }
     }
 
     /** Drops the provided sequences, using a batch execution if [inBatch] is set to `true`. */
     fun dropSequence(vararg seq: Sequence, inBatch: Boolean = false) {
-        with(TransactionManager.current() as JdbcTransaction) {
+        with(TransactionManager.current()) {
             val dropStatements = seq.flatMap { it.dropStatement() }
             execStatements(inBatch, dropStatements)
         }
@@ -129,7 +128,7 @@ object SchemaUtils : SchemaUtilityApi() {
 
     /** Creates all [tables] that do not already exist, using a batch execution if [inBatch] is set to `true`. */
     fun <T : Table> create(vararg tables: T, inBatch: Boolean = false) {
-        with(TransactionManager.current() as JdbcTransaction) {
+        with(TransactionManager.current()) {
             execStatements(inBatch, createStatements(*tables))
             commit()
             currentDialectMetadata.resetCaches()
@@ -147,7 +146,7 @@ object SchemaUtils : SchemaUtilityApi() {
      * @see org.jetbrains.exposed.sql.tests.shared.ddl.CreateDatabaseTest
      */
     fun createDatabase(vararg databases: String, inBatch: Boolean = false) {
-        val transaction = TransactionManager.current() as JdbcTransaction
+        val transaction = TransactionManager.current()
         try {
             with(transaction) {
                 val createStatements = databases.flatMap { listOf(currentDialect.createDatabase(it)) }
@@ -171,7 +170,7 @@ object SchemaUtils : SchemaUtilityApi() {
      * @return A list of strings representing the names of all databases.
      */
     fun listDatabases(): List<String> {
-        val transaction = TransactionManager.current() as JdbcTransaction
+        val transaction = TransactionManager.current()
         return with(transaction) {
             exec(currentDialect.listDatabases()) {
                 val result = mutableListOf<String>()
@@ -194,7 +193,7 @@ object SchemaUtils : SchemaUtilityApi() {
      * @see org.jetbrains.exposed.sql.tests.shared.ddl.CreateDatabaseTest
      */
     fun dropDatabase(vararg databases: String, inBatch: Boolean = false) {
-        val transaction = TransactionManager.current() as JdbcTransaction
+        val transaction = TransactionManager.current()
         try {
             with(transaction) {
                 val createStatements = databases.flatMap { listOf(currentDialect.dropDatabase(it)) }
@@ -244,7 +243,7 @@ object SchemaUtils : SchemaUtilityApi() {
         DeprecationLevel.WARNING
     )
     fun createMissingTablesAndColumns(vararg tables: Table, inBatch: Boolean = false, withLogs: Boolean = true) {
-        with(TransactionManager.current() as JdbcTransaction) {
+        with(TransactionManager.current()) {
             db.dialectMetadata.resetCaches()
             @OptIn(InternalApi::class)
             val createStatements = logTimeSpent(createTablesLogMessage, withLogs) {
@@ -422,7 +421,7 @@ object SchemaUtils : SchemaUtilityApi() {
     /** Drops all [tables], using a batch execution if [inBatch] is set to `true`. */
     fun drop(vararg tables: Table, inBatch: Boolean = false) {
         if (tables.isEmpty()) return
-        with(TransactionManager.current() as JdbcTransaction) {
+        with(TransactionManager.current()) {
             var tablesForDeletion = sortTablesByReferences(tables.toList()).reversed().filter { it in tables }
             if (!currentDialect.supportsIfNotExists) {
                 tablesForDeletion = tablesForDeletion.filter { it.exists() }
@@ -440,7 +439,7 @@ object SchemaUtils : SchemaUtilityApi() {
      * @sample org.jetbrains.exposed.sql.tests.shared.SchemaTests
      */
     fun setSchema(schema: Schema, inBatch: Boolean = false) {
-        with(TransactionManager.current() as JdbcTransaction) {
+        with(TransactionManager.current()) {
             val createStatements = schema.setSchemaStatement()
 
             execStatements(inBatch, createStatements)
@@ -466,7 +465,7 @@ object SchemaUtils : SchemaUtilityApi() {
      */
     fun createSchema(vararg schemas: Schema, inBatch: Boolean = false) {
         if (schemas.isEmpty()) return
-        with(TransactionManager.current() as JdbcTransaction) {
+        with(TransactionManager.current()) {
             val toCreate = schemas.distinct().filterNot { it.exists() }
             val createStatements = toCreate.flatMap { it.createStatement() }
             execStatements(inBatch, createStatements)
@@ -491,7 +490,7 @@ object SchemaUtils : SchemaUtilityApi() {
      */
     fun dropSchema(vararg schemas: Schema, cascade: Boolean = false, inBatch: Boolean = false) {
         if (schemas.isEmpty()) return
-        with(TransactionManager.current() as JdbcTransaction) {
+        with(TransactionManager.current()) {
             val schemasForDeletion = if (currentDialect.supportsIfNotExists) {
                 schemas.distinct()
             } else {

diff --git a/exposed-jdbc/src/main/kotlin/org/jetbrains/exposed/sql/SetOperations.kt b/exposed-jdbc/src/main/kotlin/org/jetbrains/exposed/sql/SetOperations.kt
@@ -31,7 +31,7 @@ sealed class SetOperation(
     override val statement: SetOperation = this
 
     protected val transaction: JdbcTransaction
-        get() = TransactionManager.current() as JdbcTransaction
+        get() = TransactionManager.current()
 
     /** The SQL statement on the left-hand side of the set operator. */
     val firstStatement: AbstractQuery<*> = when (_firstStatement) {

diff --git a/exposed-jdbc/src/main/kotlin/org/jetbrains/exposed/sql/statements/Executable.kt b/exposed-jdbc/src/main/kotlin/org/jetbrains/exposed/sql/statements/Executable.kt
@@ -3,7 +3,6 @@ package org.jetbrains.exposed.sql.statements
 import org.jetbrains.exposed.exceptions.ExposedSQLException
 import org.jetbrains.exposed.sql.InternalApi
 import org.jetbrains.exposed.sql.JdbcTransaction
-import org.jetbrains.exposed.sql.Transaction
 import org.jetbrains.exposed.sql.statements.api.JdbcPreparedStatementApi
 import org.jetbrains.exposed.sql.statements.api.PreparedStatementApi
 import java.sql.SQLException
@@ -38,8 +37,7 @@ interface Executable<out T, S : Statement<T>> {
      * Executes the SQL statement directly in the provided [transaction] and returns the generated result,
      * or `null` if either no result was retrieved or if the transaction blocked statement execution.
      */
-    fun execute(transaction: Transaction): T? {
-        transaction as JdbcTransaction
+    fun execute(transaction: JdbcTransaction): T? {
         return if (transaction.blockStatementExecution) {
             transaction.explainStatement = statement
             null

diff --git a/exposed-jdbc/src/main/kotlin/org/jetbrains/exposed/sql/statements/ReturningExecutable.kt b/exposed-jdbc/src/main/kotlin/org/jetbrains/exposed/sql/statements/ReturningExecutable.kt
@@ -15,7 +15,7 @@ open class ReturningExecutable(
     override fun JdbcPreparedStatementApi.executeInternal(transaction: JdbcTransaction): JdbcResult = executeQuery()
 
     override fun iterator(): Iterator<ResultRow> {
-        val rs = (TransactionManager.current() as JdbcTransaction).exec(this)!! as JdbcResult
+        val rs = TransactionManager.current().exec(this)!! as JdbcResult
         val resultIterator = ResultIterator(rs.result)
         return Iterable { resultIterator }.iterator()
     }

diff --git a/exposed-jdbc/src/main/kotlin/org/jetbrains/exposed/sql/statements/StatementIterator.kt b/exposed-jdbc/src/main/kotlin/org/jetbrains/exposed/sql/statements/StatementIterator.kt
@@ -1,6 +1,5 @@
 package org.jetbrains.exposed.sql.statements
 
-import org.jetbrains.exposed.sql.JdbcTransaction
 import org.jetbrains.exposed.sql.transactions.TransactionManager
 import java.sql.ResultSet
 
@@ -18,7 +17,7 @@ internal abstract class StatementIterator<T, RR>(
                 val statement = result.statement
                 result.close()
                 statement?.close()
-                (TransactionManager.current() as JdbcTransaction).openResultSetsCount--
+                TransactionManager.current().openResultSetsCount--
             }
         }
 

diff --git a/...c/src/main/kotlin/org/jetbrains/exposed/sql/statements/api/JdbcExposedDatabaseMetadata.kt b/...c/src/main/kotlin/org/jetbrains/exposed/sql/statements/api/JdbcExposedDatabaseMetadata.kt
@@ -12,6 +12,7 @@ import java.math.BigDecimal
 /**
  * Base class responsible for retrieving and storing information about the JDBC driver and underlying database.
  */
+
 abstract class JdbcExposedDatabaseMetadata(database: String) : ExposedDatabaseMetadata(database) {
     /** The connection URL for the database. */
     abstract val url: String

diff --git a/...dbc/src/main/kotlin/org/jetbrains/exposed/sql/statements/jdbc/JdbcDatabaseMetadataImpl.kt b/...dbc/src/main/kotlin/org/jetbrains/exposed/sql/statements/jdbc/JdbcDatabaseMetadataImpl.kt
@@ -63,7 +63,7 @@ class JdbcDatabaseMetadataImpl(database: String, val metadata: DatabaseMetaData)
 
         @Language("H2")
         val modeQuery = "SELECT $settingValueField FROM INFORMATION_SCHEMA.SETTINGS WHERE $settingNameField = 'MODE'"
-        (TransactionManager.current() as JdbcTransaction).exec(modeQuery) { rs ->
+        TransactionManager.current().exec(modeQuery) { rs ->
             rs.iterate { getString(settingValueField) }
         }?.firstOrNull()
     }
@@ -84,7 +84,7 @@ class JdbcDatabaseMetadataImpl(database: String, val metadata: DatabaseMetaData)
         return when (currentDialect) {
             is SQLiteDialect -> {
                 try {
-                    val transaction = TransactionManager.current() as JdbcTransaction
+                    val transaction = TransactionManager.current()
                     transaction.exec("""SELECT sqlite_compileoption_used("ENABLE_UPDATE_DELETE_LIMIT");""") { rs ->
                         rs.next()
                         rs.getBoolean(1)
@@ -295,7 +295,7 @@ class JdbcDatabaseMetadataImpl(database: String, val metadata: DatabaseMetaData)
     override fun existingSequences(vararg tables: Table): Map<Table, List<Sequence>> {
         if (currentDialect !is PostgreSQLDialect) return emptyMap()
 
-        val transaction = TransactionManager.current() as JdbcTransaction
+        val transaction = TransactionManager.current()
         return tables.associateWith { table ->
             val (_, tableSchema) = tableCatalogAndSchema(table)
             transaction.exec(
@@ -352,7 +352,7 @@ class JdbcDatabaseMetadataImpl(database: String, val metadata: DatabaseMetaData)
     @Suppress("MagicNumber")
     override fun sequences(): List<String> {
         val dialect = currentDialect
-        val transaction = TransactionManager.current() as JdbcTransaction
+        val transaction = TransactionManager.current()
         val fieldName = "SEQUENCE_NAME"
         return when (dialect) {
             is OracleDialect -> transaction.exec("SELECT $fieldName FROM USER_SEQUENCES") { rs ->
@@ -389,7 +389,7 @@ class JdbcDatabaseMetadataImpl(database: String, val metadata: DatabaseMetaData)
         val dialect = currentDialect
 
         return if (dialect is MysqlDialect) {
-            val transaction = TransactionManager.current() as JdbcTransaction
+            val transaction = TransactionManager.current()
             val inTableList = allTables.keys.joinToString("','", prefix = " ku.TABLE_NAME IN ('", postfix = "')")
             val tableSchema = "'${tables.mapNotNull { it.schemaName }.toSet().singleOrNull() ?: currentSchema}'"
             val constraintsToLoad = HashMap<String, MutableMap<String, ForeignKeyConstraint>>()

diff --git a/...d-jdbc/src/main/kotlin/org/jetbrains/exposed/sql/transactions/JdbcTransactionInterface.kt b/...d-jdbc/src/main/kotlin/org/jetbrains/exposed/sql/transactions/JdbcTransactionInterface.kt
@@ -28,8 +28,8 @@ interface JdbcTransactionInterface : TransactionInterface {
  * @throws [RuntimeException] If a manager has not been registered for the database.
  */
 @Suppress("TooGenericExceptionThrown")
-val Database?.transactionManager: ThreadLocalTransactionManager
-    get() = TransactionManager.managerFor(this) as ThreadLocalTransactionManager
+val Database?.transactionManager: TransactionManager
+    get() = TransactionManager.managerFor(this)
         ?: throw RuntimeException("Database $this does not have any transaction manager")
 
 @Suppress("TooGenericExceptionCaught")

diff --git a/...c/src/main/kotlin/org/jetbrains/exposed/sql/transactions/ThreadLocalTransactionManager.kt b/...c/src/main/kotlin/org/jetbrains/exposed/sql/transactions/ThreadLocalTransactionManager.kt
@@ -1,27 +1,21 @@
 package org.jetbrains.exposed.sql.transactions
 
-import org.jetbrains.exposed.exceptions.ExposedSQLException
+import org.jetbrains.annotations.TestOnly
 import org.jetbrains.exposed.sql.Database
 import org.jetbrains.exposed.sql.JdbcTransaction
-import org.jetbrains.exposed.sql.SchemaUtils
-import org.jetbrains.exposed.sql.SqlLogger
 import org.jetbrains.exposed.sql.Transaction
-import org.jetbrains.exposed.sql.exposedLogger
 import org.jetbrains.exposed.sql.statements.api.ExposedConnection
 import org.jetbrains.exposed.sql.statements.api.ExposedSavepoint
-import java.sql.SQLException
-import java.util.concurrent.ThreadLocalRandom
 
-/**
- * [TransactionManager] implementation registered to the provided database value [db].
- *
- * [setupTxConnection] can be provided to override the default configuration of transaction settings when a
- * connection is retrieved from the database.
- */
+@Deprecated(
+    message = "This class will be removed entirely in future releases.",
+    replaceWith = ReplaceWith("TransactionManager"),
+    level = DeprecationLevel.WARNING
+)
 class ThreadLocalTransactionManager(
     private val db: Database,
-    private val setupTxConnection: ((ExposedConnection<*>, JdbcTransactionInterface) -> Unit)? = null
-) : TransactionManager {
+    private val setupTxConnection: ((ExposedConnection<*>, TransactionInterface) -> Unit)? = null
+) : TransactionManagerApi {
     @Volatile
     override var defaultMaxAttempts: Int = db.config.defaultMaxAttempts
 
@@ -49,18 +43,15 @@ class ThreadLocalTransactionManager(
             return field
         }
 
-    /**
-     * Whether the transaction isolation level of the underlying DataSource should be retrieved from the database.
-     *
-     * This should only be set to `true` if [Database.connectsViaDataSource] has also been set to `true` and if
-     * an initial connection to the database has not already been made.
-     */
+        @Deprecated("Use DatabaseConfig to define the defaultIsolationLevel", level = DeprecationLevel.HIDDEN)
+        @TestOnly
+        set
+
     private var loadDataSourceIsolationLevel = false
 
     @Volatile
     override var defaultReadOnly: Boolean = db.config.defaultReadOnly
 
-    /** A thread local variable storing the current transaction. */
     val threadLocal = ThreadLocal<JdbcTransaction>()
 
     override fun toString(): String {
@@ -78,7 +69,7 @@ class ThreadLocalTransactionManager(
                     threadLocal = threadLocal,
                     outerTransaction = outerTransaction as? JdbcTransaction,
                     loadDataSourceIsolationLevel = loadDataSourceIsolationLevel,
-                ),
+                )
             )
 
         return transaction.apply { bindTransactionToThread(this) }
@@ -96,7 +87,7 @@ class ThreadLocalTransactionManager(
 
     private class ThreadLocalTransaction(
         override val db: Database,
-        private val setupTxConnection: ((ExposedConnection<*>, JdbcTransactionInterface) -> Unit)?,
+        private val setupTxConnection: ((ExposedConnection<*>, TransactionInterface) -> Unit)?,
         override val transactionIsolation: Int,
         override val readOnly: Boolean,
         val threadLocal: ThreadLocal<JdbcTransaction>,
@@ -193,224 +184,3 @@ class ThreadLocalTransactionManager(
             }
     }
 }
-
-/**
- * Creates a transaction then calls the [statement] block with this transaction as its receiver and returns the result.
- *
- * **Note** If the database value [db] is not set, the value used will be either the last [Database] instance created
- * or the value associated with the parent transaction (if this function is invoked in an existing transaction).
- *
- * @return The final result of the [statement] block.
- * @sample org.jetbrains.exposed.sql.tests.h2.MultiDatabaseTest.testTransactionWithDatabase
- */
-fun <T> transaction(db: Database? = null, statement: JdbcTransaction.() -> T): T =
-    transaction(
-        db.transactionManager.defaultIsolationLevel,
-        db.transactionManager.defaultReadOnly,
-        db,
-        statement
-    )
-
-/**
- * Creates a transaction with the specified [transactionIsolation] and [readOnly] settings, then calls
- * the [statement] block with this transaction as its receiver and returns the result.
- *
- * **Note** If the database value [db] is not set, the value used will be either the last [Database] instance created
- * or the value associated with the parent transaction (if this function is invoked in an existing transaction).
- *
- * @return The final result of the [statement] block.
- * @sample org.jetbrains.exposed.sql.tests.shared.ConnectionTimeoutTest.testTransactionRepetitionWithDefaults
- */
-fun <T> transaction(
-    transactionIsolation: Int,
-    readOnly: Boolean = false,
-    db: Database? = null,
-    statement: JdbcTransaction.() -> T
-): T = keepAndRestoreTransactionRefAfterRun(db) {
-    val outer = TransactionManager.currentOrNull() as? JdbcTransaction
-
-    if (outer != null && (db == null || outer.db == db)) {
-        val outerManager = outer.db.transactionManager
-
-        val transaction = outerManager.newTransaction(transactionIsolation, readOnly, outer) as JdbcTransaction
-        @Suppress("TooGenericExceptionCaught")
-        try {
-            transaction.statement().also {
-                if (outer.db.useNestedTransactions) {
-                    transaction.commit()
-                }
-            }
-        } catch (cause: SQLException) {
-            val currentStatement = transaction.currentStatement
-            transaction.rollbackLoggingException {
-                exposedLogger.warn(
-                    "Transaction rollback failed: ${it.message}. Statement: $currentStatement",
-                    it
-                )
-            }
-            throw cause
-        } catch (cause: Throwable) {
-            if (outer.db.useNestedTransactions) {
-                val currentStatement = transaction.currentStatement
-                transaction.rollbackLoggingException {
-                    exposedLogger.warn(
-                        "Transaction rollback failed: ${it.message}. Statement: $currentStatement",
-                        it
-                    )
-                }
-            }
-            throw cause
-        } finally {
-            TransactionManager.resetCurrent(outerManager)
-        }
-    } else {
-        val existingForDb = db?.transactionManager
-        existingForDb?.currentOrNull()?.let { transaction ->
-            val currentManager = outer?.db.transactionManager
-            try {
-                TransactionManager.resetCurrent(existingForDb)
-                (transaction as JdbcTransaction).statement().also {
-                    if (db.useNestedTransactions) {
-                        transaction.commit()
-                    }
-                }
-            } finally {
-                TransactionManager.resetCurrent(currentManager)
-            }
-        } ?: inTopLevelTransaction(
-            transactionIsolation,
-            readOnly,
-            db,
-            null,
-            statement
-        )
-    }
-}
-
-/**
- * Creates a transaction with the specified [transactionIsolation] and [readOnly] settings, then calls
- * the [statement] block with this transaction as its receiver and returns the result.
- *
- * **Note** All changes in this transaction will be committed at the end of the [statement] block, even if
- * it is nested and even if `DatabaseConfig.useNestedTransactions` is set to `false`.
- *
- * **Note** If the database value [db] is not set, the value used will be either the last [Database] instance created
- * or the value associated with the parent transaction (if this function is invoked in an existing transaction).
- *
- * @return The final result of the [statement] block.
- * @sample org.jetbrains.exposed.sql.tests.shared.RollbackTransactionTest.testRollbackWithoutSavepoints
- */
-fun <T> inTopLevelTransaction(
-    transactionIsolation: Int,
-    readOnly: Boolean = false,
-    db: Database? = null,
-    outerTransaction: JdbcTransaction? = null,
-    statement: JdbcTransaction.() -> T
-): T {
-    fun run(): T {
-        var attempts = 0
-
-        val outerManager = outerTransaction?.db.transactionManager.takeIf { it.currentOrNull() != null }
-
-        var intermediateDelay: Long = 0
-        var retryInterval: Long? = null
-
-        while (true) {
-            db?.let { db.transactionManager.let { m -> TransactionManager.resetCurrent(m) } }
-            val transaction = db.transactionManager.newTransaction(transactionIsolation, readOnly, outerTransaction)
-
-            @Suppress("TooGenericExceptionCaught")
-            try {
-                transaction.db.config.defaultSchema?.let { SchemaUtils.setSchema(it) }
-                val answer = transaction.statement()
-                transaction.commit()
-                return answer
-            } catch (cause: SQLException) {
-                handleSQLException(cause, transaction, attempts)
-                attempts++
-                if (attempts >= transaction.maxAttempts) {
-                    throw cause
-                }
-
-                if (retryInterval == null) {
-                    retryInterval = transaction.getRetryInterval()
-                    intermediateDelay = transaction.minRetryDelay
-                }
-                // set delay value with an exponential backoff time period.
-                val delay = when {
-                    transaction.minRetryDelay < transaction.maxRetryDelay -> {
-                        intermediateDelay += retryInterval * attempts
-                        ThreadLocalRandom.current().nextLong(intermediateDelay, intermediateDelay + retryInterval)
-                    }
-
-                    transaction.minRetryDelay == transaction.maxRetryDelay -> transaction.minRetryDelay
-                    else -> 0
-                }
-                exposedLogger.warn("Wait $delay milliseconds before retrying")
-                try {
-                    Thread.sleep(delay)
-                } catch (cause: InterruptedException) {
-                    // Do nothing
-                }
-            } catch (cause: Throwable) {
-                val currentStatement = transaction.currentStatement
-                transaction.rollbackLoggingException {
-                    exposedLogger.warn(
-                        "Transaction rollback failed: ${it.message}. Statement: $currentStatement",
-                        it
-                    )
-                }
-                throw cause
-            } finally {
-                TransactionManager.resetCurrent(outerManager)
-                closeStatementsAndConnection(transaction)
-            }
-        }
-    }
-
-    return keepAndRestoreTransactionRefAfterRun(db) {
-        run()
-    }
-}
-
-private fun <T> keepAndRestoreTransactionRefAfterRun(db: Database? = null, block: () -> T): T {
-    val manager = db.transactionManager
-    val currentTransaction = manager.currentOrNull()
-    return try {
-        block()
-    } finally {
-        manager.bindTransactionToThread(currentTransaction)
-    }
-}
-
-internal fun handleSQLException(cause: SQLException, transaction: JdbcTransaction, attempts: Int) {
-    val exposedSQLException = cause as? ExposedSQLException
-    val queriesToLog = exposedSQLException?.causedByQueries()?.joinToString(";\n") ?: "${transaction.currentStatement}"
-    val message = "Transaction attempt #$attempts failed: ${cause.message}. Statement(s): $queriesToLog"
-    exposedSQLException?.contexts?.forEach {
-        transaction.interceptors.filterIsInstance<SqlLogger>().forEach { logger ->
-            logger.log(it, transaction)
-        }
-    }
-    exposedLogger.warn(message, cause)
-    transaction.rollbackLoggingException {
-        exposedLogger.warn("Transaction rollback failed: ${it.message}. See previous log line for statement", it)
-    }
-}
-
-internal fun closeStatementsAndConnection(transaction: JdbcTransaction) {
-    val currentStatement = transaction.currentStatement
-    @Suppress("TooGenericExceptionCaught")
-    try {
-        currentStatement?.let {
-            it.closeIfPossible()
-            transaction.currentStatement = null
-        }
-        transaction.closeExecutedStatements()
-    } catch (cause: Exception) {
-        exposedLogger.warn("Statements close failed", cause)
-    }
-    transaction.closeLoggingException {
-        exposedLogger.warn("Transaction close failed: ${it.message}. Statement: $currentStatement", it)
-    }
-}
diff --git a/exposed-jdbc/src/main/kotlin/org/jetbrains/exposed/sql/transactions/TransactionManager.kt b/exposed-jdbc/src/main/kotlin/org/jetbrains/exposed/sql/transactions/TransactionManager.kt
diff --git a/...sed-jdbc/src/main/kotlin/org/jetbrains/exposed/sql/transactions/experimental/Suspended.kt b/...sed-jdbc/src/main/kotlin/org/jetbrains/exposed/sql/transactions/experimental/Suspended.kt
@@ -9,6 +9,7 @@ import org.jetbrains.exposed.sql.Database
 import org.jetbrains.exposed.sql.InternalApi
 import org.jetbrains.exposed.sql.JdbcTransaction
 import org.jetbrains.exposed.sql.exposedLogger
+import org.jetbrains.exposed.sql.transactions.CoreTransactionManager
 import org.jetbrains.exposed.sql.transactions.TransactionManager
 import org.jetbrains.exposed.sql.transactions.closeStatementsAndConnection
 import org.jetbrains.exposed.sql.transactions.handleSQLException
@@ -41,7 +42,7 @@ internal class TransactionCoroutineElement(
     override val key: CoroutineContext.Key<TransactionCoroutineElement> = Companion
 
     override fun updateThreadContext(context: CoroutineContext): TransactionContext {
-        val currentTransaction = TransactionManager.currentOrNull() as? JdbcTransaction
+        val currentTransaction = TransactionManager.currentOrNull()
         val currentManager = currentTransaction?.db?.transactionManager
         manager.bindTransactionToThread(newTransaction.value)
         TransactionManager.resetCurrent(manager)
@@ -107,14 +108,14 @@ suspend fun <T> suspendedTransactionAsync(
     readOnly: Boolean? = null,
     statement: suspend JdbcTransaction.() -> T
 ): Deferred<T> {
-    val currentTransaction = TransactionManager.currentOrNull() as? JdbcTransaction
+    val currentTransaction = TransactionManager.currentOrNull()
     return withTransactionScope(context, null, db, transactionIsolation, readOnly) {
         suspendedTransactionAsyncInternal(!holdsSameTransaction(currentTransaction), statement)
     }
 }
 
 private fun JdbcTransaction.closeAsync() {
-    val currentTransaction = TransactionManager.currentOrNull() as? JdbcTransaction
+    val currentTransaction = TransactionManager.currentOrNull()
     try {
         val temporaryManager = this.db.transactionManager
         temporaryManager.bindTransactionToThread(this)
@@ -141,14 +142,14 @@ private suspend fun <T> withTransactionScope(
     suspend fun newScope(currentTransaction: JdbcTransaction?): T {
         val currentDatabase: Database? = currentTransaction?.db
             ?: db
-            ?: TransactionManager.currentDefaultDatabase.get() as? Database
+            ?: CoreTransactionManager.getDefaultDatabase() as? Database
         val manager = currentDatabase?.transactionManager ?: TransactionManager.manager
 
         val tx = lazy(LazyThreadSafetyMode.NONE) {
             currentTransaction ?: manager.newTransaction(
                 isolation = transactionIsolation ?: manager.defaultIsolationLevel,
                 readOnly = readOnly ?: manager.defaultReadOnly
-            ) as JdbcTransaction
+            )
         }
 
         val element = TransactionCoroutineElement(tx, manager)

diff --git a/exposed-jdbc/src/main/kotlin/org/jetbrains/exposed/sql/vendors/DatabaseDialectMetadata.kt b/exposed-jdbc/src/main/kotlin/org/jetbrains/exposed/sql/vendors/DatabaseDialectMetadata.kt
@@ -18,28 +18,26 @@ abstract class DatabaseDialectMetadata {
     /** Returns a list with the names of all the defined tables in the current schema. */
     val allTablesNames: List<String>
         get() {
-            val connection = (TransactionManager.current() as JdbcTransaction).connection
+            val connection = TransactionManager.current().connection
             return connection.metadata { tableNamesByCurrentSchema(getAllTableNamesCache()).tableNames }
         }
 
     protected fun getAllTableNamesCache(): Map<String, List<String>> {
         if (_allTableNames == null) {
-            val tx = TransactionManager.current() as JdbcTransaction
-            _allTableNames = tx.connection.metadata { tableNames }
+            _allTableNames = TransactionManager.current().connection.metadata { tableNames }
         }
         return _allTableNames!!
     }
 
     private fun getAllSchemaNamesCache(): List<String> {
         if (_allSchemaNames == null) {
-            val tx = TransactionManager.current() as JdbcTransaction
-            _allSchemaNames = tx.connection.metadata { schemaNames }
+            _allSchemaNames = TransactionManager.current().connection.metadata { schemaNames }
         }
         return _allSchemaNames!!
     }
 
     /** Returns the name of the current database. */
-    fun getDatabase(): String = catalog(TransactionManager.current() as JdbcTransaction)
+    fun getDatabase(): String = catalog(TransactionManager.current())
 
     /** Returns the catalog name of the connection of the specified [transaction]. */
     fun catalog(transaction: JdbcTransaction): String = transaction.connection.catalog
@@ -49,8 +47,7 @@ abstract class DatabaseDialectMetadata {
      * The names will be returned with schema prefixes if the database supports it.
      */
     fun allTablesNames(): List<String> {
-        val tx = TransactionManager.current() as JdbcTransaction
-        return tx.connection.metadata {
+        return TransactionManager.current().connection.metadata {
             tableNamesByCurrentSchema(null).tableNames
         }
     }
@@ -73,8 +70,7 @@ abstract class DatabaseDialectMetadata {
                 it == table.nameInDatabaseCase()
             }
         } ?: run {
-            val tx = TransactionManager.current() as JdbcTransaction
-            val (schema, allTables) = tx.connection.metadata {
+            val (schema, allTables) = TransactionManager.current().connection.metadata {
                 tableNamesByCurrentSchema(getAllTableNamesCache())
             }
             allTables.any {
@@ -108,14 +104,13 @@ abstract class DatabaseDialectMetadata {
 
     /** Returns a map with the column metadata of all the defined columns in each of the specified [tables]. */
     fun tableColumns(vararg tables: Table): Map<Table, List<ColumnMetadata>> {
-        val tx = TransactionManager.current() as JdbcTransaction
-        return tx.connection.metadata { columns(*tables) }
+        return TransactionManager.current().connection.metadata { columns(*tables) }
     }
 
     protected val columnConstraintsCache: MutableMap<String, Collection<ForeignKeyConstraint>> = ConcurrentHashMap()
 
     protected open fun fillConstraintCacheForTables(tables: List<Table>) {
-        val tx = TransactionManager.current() as JdbcTransaction
+        val tx = TransactionManager.current()
         columnConstraintsCache.putAll(tx.db.metadata { tableConstraints(tables) })
     }
 
@@ -138,14 +133,12 @@ abstract class DatabaseDialectMetadata {
 
     /** Returns a map with all the defined indices in each of the specified [tables]. */
     open fun existingIndices(vararg tables: Table): Map<Table, List<Index>> {
-        val tx = TransactionManager.current() as JdbcTransaction
-        return tx.db.metadata { existingIndices(*tables) }
+        return TransactionManager.current().db.metadata { existingIndices(*tables) }
     }
 
     /** Returns a map with the primary key metadata in each of the specified [tables]. */
     fun existingPrimaryKeys(vararg tables: Table): Map<Table, PrimaryKeyMetadata?> {
-        val tx = TransactionManager.current() as JdbcTransaction
-        return tx.db.metadata { existingPrimaryKeys(*tables) }
+        return TransactionManager.current().db.metadata { existingPrimaryKeys(*tables) }
     }
 
     /**
@@ -159,21 +152,19 @@ abstract class DatabaseDialectMetadata {
      * not be returned.
      */
     fun existingSequences(vararg tables: Table): Map<Table, List<Sequence>> {
-        val tx = TransactionManager.current() as JdbcTransaction
-        return tx.db.metadata { existingSequences(*tables) }
+        return TransactionManager.current().db.metadata { existingSequences(*tables) }
     }
 
     /** Returns a list of the names of all sequences in the database. */
     fun sequences(): List<String> {
-        val tx = TransactionManager.current() as JdbcTransaction
-        return tx.db.metadata { sequences() }
+        return TransactionManager.current().db.metadata { sequences() }
     }
 
     /** Clears any cached values. */
     fun resetCaches() {
         _allTableNames = null
         columnConstraintsCache.clear()
-        (TransactionManager.current() as JdbcTransaction).db.metadata { cleanCache() }
+        TransactionManager.current().db.metadata { cleanCache() }
     }
 
     /** Clears any cached values including schema names. */
@@ -187,7 +178,7 @@ private val explicitDialect = ThreadLocal<DatabaseDialectMetadata?>()
 
 /** Returns the dialect used in the current transaction, may throw an exception if there is no current transaction. */
 val currentDialectMetadata: DatabaseDialectMetadata
-    get() = explicitDialect.get() ?: (TransactionManager.current() as JdbcTransaction).db.dialectMetadata
+    get() = explicitDialect.get() ?: TransactionManager.current().db.dialectMetadata
 
 internal fun String.inProperCase(): String =
     TransactionManager.currentOrNull()?.db?.identifierManager?.inProperCase(this@inProperCase) ?: this
diff --git a/exposed-jdbc/src/main/kotlin/org/jetbrains/exposed/sql/vendors/PostgreSQLDialectMetadata.kt b/exposed-jdbc/src/main/kotlin/org/jetbrains/exposed/sql/vendors/PostgreSQLDialectMetadata.kt
@@ -1,7 +1,7 @@
 package org.jetbrains.exposed.sql.vendors
 
 open class PostgreSQLDialectMetadata : DatabaseDialectMetadata() {
-    override fun supportsLimitWithUpdateOrDelete(): Boolean = true
+    override fun supportsLimitWithUpdateOrDelete(): Boolean = false
 }
 
 class PostgreSQLNGDialectMetadata : PostgreSQLDialectMetadata()
diff --git a/exposed-jdbc/src/main/kotlin/org/jetbrains/exposed/sql/vendors/SQLiteDialectMetadata.kt b/exposed-jdbc/src/main/kotlin/org/jetbrains/exposed/sql/vendors/SQLiteDialectMetadata.kt
@@ -1,10 +1,9 @@
 package org.jetbrains.exposed.sql.vendors
 
-import org.jetbrains.exposed.sql.JdbcTransaction
 import org.jetbrains.exposed.sql.transactions.TransactionManager
 
 open class SQLiteDialectMetadata : DatabaseDialectMetadata() {
     override fun supportsLimitWithUpdateOrDelete(): Boolean {
-        return (TransactionManager.current() as JdbcTransaction).db.metadata { supportsLimitWithUpdateOrDelete() }
+        return TransactionManager.current().db.metadata { supportsLimitWithUpdateOrDelete() }
     }
 }
diff --git a/exposed-r2dbc/api/exposed-r2dbc.api b/exposed-r2dbc/api/exposed-r2dbc.api
diff --git a/exposed-r2dbc/src/main/kotlin/org/jetbrains/exposed/r2dbc/sql/Queries.kt b/exposed-r2dbc/src/main/kotlin/org/jetbrains/exposed/r2dbc/sql/Queries.kt
@@ -22,9 +22,9 @@ import org.jetbrains.exposed.sql.SqlExpressionBuilder
 import org.jetbrains.exposed.sql.Table
 import org.jetbrains.exposed.sql.statements.*
 import org.jetbrains.exposed.sql.transactions.TransactionManager
-import org.jetbrains.exposed.sql.vendors.SQLiteDialect
 import org.jetbrains.exposed.sql.vendors.currentDialect
 import org.jetbrains.exposed.sql.vendors.currentDialectMetadata
+import kotlin.collections.ArrayList
 import kotlin.collections.Iterable
 import kotlin.collections.Iterator
 import kotlin.collections.List
@@ -79,7 +79,9 @@ suspend fun <T : Table> T.deleteWhere(
     limit: Int? = null,
     op: T.(ISqlExpressionBuilder) -> Op<Boolean>
 ): Int {
-    limit?.let { checkUpdateDeleteLimitEnabled() }
+    if (limit != null && !currentDialectMetadata.supportsLimitWithUpdateOrDelete()) {
+        throw UnsupportedByDialectException("LIMIT clause is not supported in DELETE statement.", currentDialect)
+    }
     val stmt = StatementBuilder { deleteWhere(limit, op) }
     return DeleteExecutable(stmt).execute(TransactionManager.current()) ?: 0
 }
@@ -98,7 +100,9 @@ suspend fun <T : Table> T.deleteIgnoreWhere(
     limit: Int? = null,
     op: T.(ISqlExpressionBuilder) -> Op<Boolean>
 ): Int {
-    limit?.let { checkUpdateDeleteLimitEnabled() }
+    if (limit != null && !currentDialectMetadata.supportsLimitWithUpdateOrDelete()) {
+        throw UnsupportedByDialectException("LIMIT clause is not supported in DELETE statement.", currentDialect)
+    }
     val stmt = StatementBuilder { deleteIgnoreWhere(limit, op) }
     return DeleteExecutable(stmt).execute(TransactionManager.current()) ?: 0
 }
@@ -546,7 +550,9 @@ suspend fun <T : Table> T.update(
     limit: Int? = null,
     body: T.(UpdateStatement) -> Unit
 ): Int {
-    limit?.let { checkUpdateDeleteLimitEnabled() }
+    if (limit != null && !currentDialectMetadata.supportsLimitWithUpdateOrDelete()) {
+        throw UnsupportedByDialectException("LIMIT clause is not supported in UPDATE statement.", currentDialect)
+    }
     val stmt = StatementBuilder { update(where, limit, body) }
     return UpdateExecutable(stmt).execute(TransactionManager.current()) ?: 0
 }
@@ -562,7 +568,9 @@ suspend fun <T : Table> T.update(
     limit: Int? = null,
     body: T.(UpdateStatement) -> Unit
 ): Int {
-    limit?.let { checkUpdateDeleteLimitEnabled() }
+    if (limit != null && !currentDialectMetadata.supportsLimitWithUpdateOrDelete()) {
+        throw UnsupportedByDialectException("LIMIT clause is not supported in UPDATE statement.", currentDialect)
+    }
     val stmt = StatementBuilder { update(null, limit, body) }
     return UpdateExecutable(stmt).execute(TransactionManager.current()) ?: 0
 }
@@ -592,7 +600,6 @@ suspend fun Join.update(
     limit: Int? = null,
     body: (UpdateStatement) -> Unit
 ): Int {
-    limit?.let { checkUpdateDeleteLimitEnabled() }
     val stmt = StatementBuilder { update(where, limit, body) }
     return UpdateExecutable(stmt).execute(TransactionManager.current()) ?: 0
 }
@@ -608,17 +615,10 @@ suspend fun Join.update(
     limit: Int? = null,
     body: (UpdateStatement) -> Unit
 ): Int {
-    limit?.let { checkUpdateDeleteLimitEnabled() }
     val stmt = StatementBuilder { update(null, limit, body) }
     return UpdateExecutable(stmt).execute(TransactionManager.current()) ?: 0
 }
 
-private suspend fun checkUpdateDeleteLimitEnabled() {
-    if (currentDialect is SQLiteDialect && !currentDialectMetadata.updateDeleteLimitEnabled()) {
-        throw UnsupportedByDialectException("SQLite doesn't support LIMIT in UPDATE or DELETE operations.", currentDialect)
-    }
-}
-
 @Deprecated(
     "This `updateReturning()` with a nullable `where` parameter will be removed in future releases. Please leave a comment on " +
         "[YouTrack](https://youtrack.jetbrains.com/issue/EXPOSED-494/Inline-DSL-statement-and-query-functions) " +

diff --git a/exposed-r2dbc/src/main/kotlin/org/jetbrains/exposed/sql/ExplainExecutable.kt b/exposed-r2dbc/src/main/kotlin/org/jetbrains/exposed/sql/ExplainExecutable.kt
@@ -17,7 +17,7 @@ open class ExplainExecutable(
     override suspend fun R2dbcPreparedStatementApi.executeInternal(transaction: R2dbcTransaction): R2dbcResult = executeQuery()
 
     override suspend fun collect(collector: FlowCollector<ExplainResultRow>) {
-        val rs = ((TransactionManager.current() as R2dbcTransaction).exec(this)!! as R2dbcResult)
+        val rs = TransactionManager.current().exec(this)!! as R2dbcResult
         // how to iterate over fields in a Row/RowMetadata?
 //        val fieldIndex: Map<String, Int> = List(rs.result.columnCount) { i ->
 //            rs.metaData.getColumnName(i + 1) to i

diff --git a/exposed-r2dbc/src/main/kotlin/org/jetbrains/exposed/sql/Query.kt b/exposed-r2dbc/src/main/kotlin/org/jetbrains/exposed/sql/Query.kt
@@ -26,7 +26,7 @@ open class Query(
     }
 
     protected val transaction: R2dbcTransaction
-        get() = TransactionManager.current() as R2dbcTransaction
+        get() = TransactionManager.current()
 
     /** Creates a new [Query] instance using all stored properties of this `SELECT` query. */
     override fun copy(): Query = Query(set, where).also { copy ->
@@ -297,7 +297,7 @@ open class Query(
         val fieldIndex = set.realFields.toSet()
             .mapIndexed { index, expression -> expression to index }
             .toMap()
-        val tx = (TransactionManager.current() as R2dbcTransaction)
+        val tx = TransactionManager.current()
         val rs = tx.exec(queryToExecute)!! as R2dbcResult
 
         collector.emit(ResultRow.create(rs, fieldIndex).also { trackResultSet(tx) })

diff --git a/exposed-r2dbc/src/main/kotlin/org/jetbrains/exposed/sql/R2dbcDatabase.kt b/exposed-r2dbc/src/main/kotlin/org/jetbrains/exposed/sql/R2dbcDatabase.kt
@@ -12,8 +12,9 @@ import org.jetbrains.exposed.sql.statements.api.R2dbcExposedDatabaseMetadata
 import org.jetbrains.exposed.sql.statements.r2dbc.R2dbcConnectionImpl
 import org.jetbrains.exposed.sql.statements.r2dbc.R2dbcScope
 import org.jetbrains.exposed.sql.statements.r2dbc.asInt
-import org.jetbrains.exposed.sql.transactions.R2dbcTransactionManager
+import org.jetbrains.exposed.sql.transactions.CoreTransactionManager
 import org.jetbrains.exposed.sql.transactions.TransactionManager
+import org.jetbrains.exposed.sql.transactions.TransactionManagerApi
 import org.jetbrains.exposed.sql.vendors.*
 import org.reactivestreams.Publisher
 import java.math.BigDecimal
@@ -28,7 +29,7 @@ class R2dbcDatabase private constructor(
     val connector: () -> R2dbcExposedConnection<*>
 ) : DatabaseApi(resolvedVendor, config) {
     internal suspend fun <T> metadata(body: suspend R2dbcExposedDatabaseMetadata.() -> T): T {
-        val transaction = TransactionManager.currentOrNull() as? R2dbcTransaction
+        val transaction = TransactionManager.currentOrNull()
         return if (transaction == null) {
             val connection = connector()
             try {
@@ -128,18 +129,21 @@ class R2dbcDatabase private constructor(
             r2dbcDialectMapping[prefix] = dialect
         }
 
+        @OptIn(InternalApi::class)
         private fun doConnect(
             url: String,
             explicitVendor: String,
             config: DatabaseConfig?,
             getNewConnection: () -> Publisher<out Connection>,
             dispatcher: CoroutineDispatcher?,
-            manager: (R2dbcDatabase) -> TransactionManager = { R2dbcTransactionManager(it) }
+            manager: (R2dbcDatabase) -> TransactionManagerApi = { TransactionManager(it) }
         ): R2dbcDatabase {
             return R2dbcDatabase(url, config ?: DatabaseConfig.invoke()) {
                 R2dbcConnectionImpl(explicitVendor, getNewConnection(), R2dbcScope(dispatcher))
             }.apply {
-                TransactionManager.registerManager(this, manager(this))
+                CoreTransactionManager.registerDatabaseManager(this, manager(this))
+                // ABOVE should be replaced with BELOW when ThreadLocalTransactionManager is fully deprecated
+                // TransactionManager.registerManager(this, manager(this))
             }
         }
 
@@ -161,7 +165,7 @@ class R2dbcDatabase private constructor(
             connectionOptions: ConnectionFactoryOptions,
             databaseConfig: DatabaseConfig? = null,
             dispatcher: CoroutineDispatcher? = null,
-            manager: (R2dbcDatabase) -> TransactionManager = { R2dbcTransactionManager(it) }
+            manager: (R2dbcDatabase) -> TransactionManagerApi = { TransactionManager(it) }
         ): R2dbcDatabase {
             val url = "r2dbc:${connectionOptions.getValue(ConnectionFactoryOptions.DRIVER)}"
             val dialectName = getR2dbcDialectName(url) ?: error("Can't resolve dialect for connection: $url")
@@ -194,7 +198,7 @@ class R2dbcDatabase private constructor(
             databaseConfig: DatabaseConfig? = null,
             databaseDialect: DatabaseDialect = databaseConfig?.explicitDialect ?: error("Can't resolve dialect for connection"),
             dispatcher: CoroutineDispatcher? = null,
-            manager: (R2dbcDatabase) -> TransactionManager = { R2dbcTransactionManager(it) }
+            manager: (R2dbcDatabase) -> TransactionManagerApi = { TransactionManager(it) }
         ): R2dbcDatabase {
             return doConnect(
                 url = "",
@@ -222,7 +226,7 @@ class R2dbcDatabase private constructor(
             url: String,
             databaseConfig: DatabaseConfig? = null,
             dispatcher: CoroutineDispatcher? = null,
-            manager: (R2dbcDatabase) -> TransactionManager = { R2dbcTransactionManager(it) }
+            manager: (R2dbcDatabase) -> TransactionManagerApi = { TransactionManager(it) }
         ): R2dbcDatabase {
             val dialectName = getR2dbcDialectName(url) ?: error("Can't resolve dialect for connection: $url")
             return doConnect(

diff --git a/exposed-r2dbc/src/main/kotlin/org/jetbrains/exposed/sql/R2dbcTransaction.kt b/exposed-r2dbc/src/main/kotlin/org/jetbrains/exposed/sql/R2dbcTransaction.kt
@@ -17,6 +17,7 @@ import org.jetbrains.exposed.sql.statements.api.R2dbcPreparedStatementApi
 import org.jetbrains.exposed.sql.statements.executeIn
 import org.jetbrains.exposed.sql.statements.r2dbc.R2dbcResult
 import org.jetbrains.exposed.sql.transactions.R2dbcTransactionInterface
+import org.jetbrains.exposed.sql.transactions.transactionManager
 import java.util.*
 import java.util.concurrent.TimeUnit
 
@@ -25,6 +26,25 @@ open class R2dbcTransaction(
 ) : Transaction(), R2dbcTransactionInterface by transactionImpl {
     final override val db: R2dbcDatabase = transactionImpl.db
 
+    /**
+     * The maximum amount of attempts that will be made to perform this `transaction` block.
+     *
+     * If this value is set to 1 and an SQLException happens, the exception will be thrown without performing a retry.
+     *
+     * @throws IllegalArgumentException If the amount of attempts is set to a value less than 1.
+     */
+    var maxAttempts: Int = db.transactionManager.defaultMaxAttempts
+        set(value) {
+            require(value > 0) { "maxAttempts must be set to perform at least 1 attempt." }
+            field = value
+        }
+
+    /** The minimum number of milliseconds to wait before retrying this `transaction` if an SQLException happens. */
+    var minRetryDelay: Long = db.transactionManager.defaultMinRetryDelay
+
+    /** The maximum number of milliseconds to wait before retrying this `transaction` if an SQLException happens. */
+    var maxRetryDelay: Long = db.transactionManager.defaultMaxRetryDelay
+
     /** The currently executing statement. */
     var currentStatement: R2dbcPreparedStatementApi? = null
 

diff --git a/exposed-r2dbc/src/main/kotlin/org/jetbrains/exposed/sql/SchemaUtils.kt b/exposed-r2dbc/src/main/kotlin/org/jetbrains/exposed/sql/SchemaUtils.kt
@@ -50,15 +50,15 @@ object SchemaUtils : SchemaUtilityApi() {
 
     /** Creates the provided sequences, using a batch execution if [inBatch] is set to `true`. */
     suspend fun createSequence(vararg seq: Sequence, inBatch: Boolean = false) {
-        with(TransactionManager.current() as R2dbcTransaction) {
+        with(TransactionManager.current()) {
             val createStatements = seq.flatMap { it.createStatement() }
             execStatements(inBatch, createStatements)
         }
     }
 
     /** Drops the provided sequences, using a batch execution if [inBatch] is set to `true`. */
     suspend fun dropSequence(vararg seq: Sequence, inBatch: Boolean = false) {
-        with(TransactionManager.current() as R2dbcTransaction) {
+        with(TransactionManager.current()) {
             val dropStatements = seq.flatMap { it.dropStatement() }
             execStatements(inBatch, dropStatements)
         }
@@ -135,7 +135,7 @@ object SchemaUtils : SchemaUtilityApi() {
 
     /** Creates all [tables] that do not already exist, using a batch execution if [inBatch] is set to `true`. */
     suspend fun <T : Table> create(vararg tables: T, inBatch: Boolean = false) {
-        with(TransactionManager.current() as R2dbcTransaction) {
+        with(TransactionManager.current()) {
             execStatements(inBatch, createStatements(*tables))
             commit()
             currentDialectMetadata.resetCaches()
@@ -153,7 +153,7 @@ object SchemaUtils : SchemaUtilityApi() {
      * @see org.jetbrains.exposed.sql.tests.shared.ddl.CreateDatabaseTest
      */
     suspend fun createDatabase(vararg databases: String, inBatch: Boolean = false) {
-        val transaction = TransactionManager.current() as R2dbcTransaction
+        val transaction = TransactionManager.current()
         try {
             with(transaction) {
                 val createStatements = databases.flatMap { listOf(currentDialect.createDatabase(it)) }
@@ -177,7 +177,7 @@ object SchemaUtils : SchemaUtilityApi() {
      * @return A list of strings representing the names of all databases.
      */
     suspend fun listDatabases(): List<String> {
-        val transaction = TransactionManager.current() as R2dbcTransaction
+        val transaction = TransactionManager.current()
         return with(transaction) {
             exec(currentDialect.listDatabases()) { row ->
                 row.get(1, String::class.java)
@@ -196,7 +196,7 @@ object SchemaUtils : SchemaUtilityApi() {
      * @see org.jetbrains.exposed.sql.tests.shared.ddl.CreateDatabaseTest
      */
     suspend fun dropDatabase(vararg databases: String, inBatch: Boolean = false) {
-        val transaction = TransactionManager.current() as R2dbcTransaction
+        val transaction = TransactionManager.current()
         try {
             with(transaction) {
                 val createStatements = databases.flatMap { listOf(currentDialect.dropDatabase(it)) }
@@ -246,7 +246,7 @@ object SchemaUtils : SchemaUtilityApi() {
         DeprecationLevel.WARNING
     )
     suspend fun createMissingTablesAndColumns(vararg tables: Table, inBatch: Boolean = false, withLogs: Boolean = true) {
-        with(TransactionManager.current() as R2dbcTransaction) {
+        with(TransactionManager.current()) {
             db.dialectMetadata.resetCaches()
             @OptIn(InternalApi::class)
             val createStatements = logTimeSpent(createTablesLogMessage, withLogs) {
@@ -424,7 +424,7 @@ object SchemaUtils : SchemaUtilityApi() {
     /** Drops all [tables], using a batch execution if [inBatch] is set to `true`. */
     suspend fun drop(vararg tables: Table, inBatch: Boolean = false) {
         if (tables.isEmpty()) return
-        with(TransactionManager.current() as R2dbcTransaction) {
+        with(TransactionManager.current()) {
             var tablesForDeletion = sortTablesByReferences(tables.toList()).reversed().filter { it in tables }
             if (!currentDialect.supportsIfNotExists) {
                 tablesForDeletion = tablesForDeletion.filter { it.exists() }
@@ -442,7 +442,7 @@ object SchemaUtils : SchemaUtilityApi() {
      * @sample org.jetbrains.exposed.sql.tests.shared.SchemaTests
      */
     suspend fun setSchema(schema: Schema, inBatch: Boolean = false) {
-        with(TransactionManager.current() as R2dbcTransaction) {
+        with(TransactionManager.current()) {
             val createStatements = schema.setSchemaStatement()
 
             execStatements(inBatch, createStatements)
@@ -468,7 +468,7 @@ object SchemaUtils : SchemaUtilityApi() {
      */
     suspend fun createSchema(vararg schemas: Schema, inBatch: Boolean = false) {
         if (schemas.isEmpty()) return
-        with(TransactionManager.current() as R2dbcTransaction) {
+        with(TransactionManager.current()) {
             val toCreate = schemas.distinct().filterNot { it.exists() }
             val createStatements = toCreate.flatMap { it.createStatement() }
             execStatements(inBatch, createStatements)
@@ -493,7 +493,7 @@ object SchemaUtils : SchemaUtilityApi() {
      */
     suspend fun dropSchema(vararg schemas: Schema, cascade: Boolean = false, inBatch: Boolean = false) {
         if (schemas.isEmpty()) return
-        with(TransactionManager.current() as R2dbcTransaction) {
+        with(TransactionManager.current()) {
             val schemasForDeletion = if (currentDialect.supportsIfNotExists) {
                 schemas.distinct()
             } else {

diff --git a/exposed-r2dbc/src/main/kotlin/org/jetbrains/exposed/sql/SetOperations.kt b/exposed-r2dbc/src/main/kotlin/org/jetbrains/exposed/sql/SetOperations.kt
@@ -31,7 +31,7 @@ sealed class SetOperation(
     override val statement: SetOperation = this
 
     protected val transaction: R2dbcTransaction
-        get() = TransactionManager.current() as R2dbcTransaction
+        get() = TransactionManager.current()
 
     /** The SQL statement on the left-hand side of the set operator. */
     val firstStatement: AbstractQuery<*> = when (_firstStatement) {
@@ -156,7 +156,7 @@ sealed class SetOperation(
         val fieldIndex = set.realFields.toSet()
             .mapIndexed { index, expression -> expression to index }
             .toMap()
-        val tx = (TransactionManager.current() as R2dbcTransaction)
+        val tx = TransactionManager.current()
         val rs = tx.exec(queryToExecute)!! as R2dbcResult
 
         collector.emit(ResultRow.create(rs, fieldIndex).also { trackResultSet(tx) })

diff --git a/exposed-r2dbc/src/main/kotlin/org/jetbrains/exposed/sql/statements/Executable.kt b/exposed-r2dbc/src/main/kotlin/org/jetbrains/exposed/sql/statements/Executable.kt
@@ -3,7 +3,6 @@ package org.jetbrains.exposed.sql.statements
 import org.jetbrains.exposed.exceptions.ExposedSQLException
 import org.jetbrains.exposed.sql.InternalApi
 import org.jetbrains.exposed.sql.R2dbcTransaction
-import org.jetbrains.exposed.sql.Transaction
 import org.jetbrains.exposed.sql.statements.api.R2dbcPreparedStatementApi
 import java.sql.SQLException
 
@@ -37,8 +36,7 @@ interface Executable<out T, S : Statement<T>> {
      * Executes the SQL statement directly in the provided [transaction] and returns the generated result,
      * or `null` if either no result was retrieved or if the transaction blocked statement execution.
      */
-    suspend fun execute(transaction: Transaction): T? {
-        transaction as R2dbcTransaction
+    suspend fun execute(transaction: R2dbcTransaction): T? {
         return if (transaction.blockStatementExecution) {
             transaction.explainStatement = statement
             null

diff --git a/exposed-r2dbc/src/main/kotlin/org/jetbrains/exposed/sql/statements/ReturningExecutable.kt b/exposed-r2dbc/src/main/kotlin/org/jetbrains/exposed/sql/statements/ReturningExecutable.kt
@@ -17,7 +17,7 @@ open class ReturningExecutable(
     override suspend fun collect(collector: FlowCollector<ResultRow>) {
         val fieldIndex = statement.returningExpressions.withIndex()
             .associateBy({ it.value }, { it.index })
-        val rs = (TransactionManager.current() as R2dbcTransaction).exec(this)!! as R2dbcResult
+        val rs = TransactionManager.current().exec(this)!! as R2dbcResult
 
         collector.emit(ResultRow.create(rs, fieldIndex))
     }

diff --git a/.../src/main/kotlin/org/jetbrains/exposed/sql/statements/api/R2dbcExposedDatabaseMetadata.kt b/.../src/main/kotlin/org/jetbrains/exposed/sql/statements/api/R2dbcExposedDatabaseMetadata.kt
@@ -22,8 +22,8 @@ abstract class R2dbcExposedDatabaseMetadata(database: String) : ExposedDatabaseM
     /** Whether the database supports `SELECT FOR UPDATE` statements. */
     abstract val supportsSelectForUpdate: Boolean
 
-    /** Whether the database has been set up to allow `LIMIT` clause with update and delete statements. */
-    abstract suspend fun updateDeleteLimitEnabled(): Boolean
+    /** Whether the database supports the `LIMIT` clause with update and delete statements. */
+    abstract suspend fun supportsLimitWithUpdateOrDelete(): Boolean
 
     /** The connection URL for the database. */
     abstract suspend fun getUrl(): String