diff --git a/.github/workflows/integration-tests-r2dbc.yml b/.github/workflows/integration-tests-r2dbc.yml
index 8b6fd807..6d70e39b 100644
--- a/.github/workflows/integration-tests-r2dbc.yml
+++ b/.github/workflows/integration-tests-r2dbc.yml
@@ -207,3 +207,112 @@ jobs:
       - name: Print logs on failure
         if: ${{ failure() }}
         run: find . -name "*.log" -exec ./scripts/cat-log.sh {} \;
+
+  mariadb-test:
+    name: Build and Test Integration for R2DBC with MariaDB
+    runs-on: ubuntu-22.04
+    if: github.repository == 'apache/pekko-projection'
+    strategy:
+      fail-fast: false
+      matrix:
+        include:
+          - { java-version: 17, scala-version: 2.13,  sbt-opts: '' }
+          - { java-version: 17, scala-version: next,   sbt-opts: '' }
+
+    steps:
+      - name: Checkout
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+        with:
+          fetch-depth: 0
+          fetch-tags: true
+
+      - name: Checkout GitHub merge
+        if: github.event.pull_request
+        run: |-
+          git fetch origin pull/${{ github.event.pull_request.number }}/merge:scratch
+          git checkout scratch
+
+      - name: Cache Coursier cache
+        uses: coursier/cache-action@057d08c2f9a2c4e7a4d33df3aae657b7db88b8cc # v8.2.0
+
+      - name: Setup JDK ${{ matrix.java-version }}
+        uses: actions/setup-java@b843ebf9b913146b38da2eaa7aeddf76074ae1ad # v6.1.0
+        with:
+          java-version: ${{ matrix.java-version }}
+          distribution: temurin
+
+      - name: Install sbt
+        uses: sbt/setup-sbt@ecad189dd245444b9ae0a0cd5922fd938b88906f # v1.5.12
+
+      - name: Start MariaDB
+        run: |-
+          docker compose -f docker-files/docker-compose-mariadb.yml up -d --wait
+          # loop to wait for the DB to be ready
+          for i in {1..8}; do
+            if docker exec -i docker-mariadb-db-1 mariadb \
+                -h 127.0.0.1 --user=root --password=root --database=mysql -e 'SELECT 1;' 2>/dev/null; then
+              break
+            fi
+            echo "Retry $i: SQL check failed, waiting before retry..."
+            sleep $((i * 2))
+          done
+          docker exec -i docker-mariadb-db-1 mariadb -h 127.0.0.1 --user=root --password=root --database=mysql < r2dbc-int-test/ddl-scripts/create_tables_mysql.sql
+          docker exec -i docker-mariadb-db-1 mariadb -h 127.0.0.1 --user=root --password=root -e 'CREATE SCHEMA database1;'
+          docker exec -i docker-mariadb-db-1 mariadb -h 127.0.0.1 --user=root --password=root --database=database1 < r2dbc-int-test/ddl-scripts/create_tables_mysql.sql
+          docker exec -i docker-mariadb-db-1 mariadb -h 127.0.0.1 --user=root --password=root -e 'CREATE SCHEMA database2;'
+          docker exec -i docker-mariadb-db-1 mariadb -h 127.0.0.1 --user=root --password=root --database=database2 < r2dbc-int-test/ddl-scripts/create_tables_mysql.sql
+
+      - name: test
+        run: |-
+          scala_version="$(bash .github/scripts/resolve-scala-version.sh "${{ matrix.scala-version }}")"
+          sbt -Dpekko.persistence.r2dbc.dialect=mariadb -Dpekko.projection.r2dbc.dialect=mariadb "++${scala_version}!" "r2dbc-int-test/test" ${{ matrix.sbt-opts }}
+
+      - name: Print logs on failure
+        if: ${{ failure() }}
+        run: find . -name "*.log" -exec ./scripts/cat-log.sh {} \;
+
+  h2-test:
+    name: Build and Test Integration for R2DBC with H2
+    runs-on: ubuntu-22.04
+    if: github.repository == 'apache/pekko-projection'
+    strategy:
+      fail-fast: false
+      matrix:
+        include:
+          - { java-version: 17, scala-version: 2.13,  sbt-opts: '' }
+          - { java-version: 17, scala-version: next,   sbt-opts: '' }
+
+    steps:
+      - name: Checkout
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+        with:
+          fetch-depth: 0
+          fetch-tags: true
+
+      - name: Checkout GitHub merge
+        if: github.event.pull_request
+        run: |-
+          git fetch origin pull/${{ github.event.pull_request.number }}/merge:scratch
+          git checkout scratch
+
+      - name: Cache Coursier cache
+        uses: coursier/cache-action@057d08c2f9a2c4e7a4d33df3aae657b7db88b8cc # v8.2.0
+
+      - name: Setup JDK ${{ matrix.java-version }}
+        uses: actions/setup-java@b843ebf9b913146b38da2eaa7aeddf76074ae1ad # v6.1.0
+        with:
+          java-version: ${{ matrix.java-version }}
+          distribution: temurin
+
+      - name: Install sbt
+        uses: sbt/setup-sbt@ecad189dd245444b9ae0a0cd5922fd938b88906f # v1.5.12
+
+      # H2 runs in the test JVM and the projection creates its tables, no database container or schema script
+      - name: test
+        run: |-
+          scala_version="$(bash .github/scripts/resolve-scala-version.sh "${{ matrix.scala-version }}")"
+          sbt -Dpekko.persistence.r2dbc.dialect=h2 -Dpekko.projection.r2dbc.dialect=h2 "++${scala_version}!" "r2dbc-int-test/test" ${{ matrix.sbt-opts }}
+
+      - name: Print logs on failure
+        if: ${{ failure() }}
+        run: find . -name "*.log" -exec ./scripts/cat-log.sh {} \;
diff --git a/docker-files/docker-compose-mariadb.yml b/docker-files/docker-compose-mariadb.yml
new file mode 100644
index 00000000..244bde5e
--- /dev/null
+++ b/docker-files/docker-compose-mariadb.yml
@@ -0,0 +1,32 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+services:
+  mariadb-db:
+    image: mariadb:12.3.3
+    container_name: docker-mariadb-db-1
+    ports:
+      - 3306:3306
+    environment:
+      MARIADB_ROOT_PASSWORD: root
+      # a server time zone with daylight saving time, so that the tests verify that timestamps are stored in UTC
+      TZ: America/New_York
+    healthcheck:
+      test: [ "CMD", "healthcheck.sh", "--connect", "--innodb_initialized" ]
+      interval: 1s
+      timeout: 1s
+      retries: 60
diff --git a/docs/src/main/paradox/r2dbc.md b/docs/src/main/paradox/r2dbc.md
index 304e1c2a..ea3f08b8 100644
--- a/docs/src/main/paradox/r2dbc.md
+++ b/docs/src/main/paradox/r2dbc.md
@@ -54,6 +54,36 @@ YugaByte
 MySQL
 :  @@snip [MySQL Schema](/r2dbc-int-test/ddl-scripts/create_tables_mysql.sql)
 
+MariaDB
+:  @@snip [MySQL Schema](/r2dbc-int-test/ddl-scripts/create_tables_mysql.sql)
+
+H2
+:  The tables are created automatically, see @ref:[H2](#h2).
+
+### MariaDB
+
+MariaDB 10.11 or later is supported with the `mariadb` dialect and the `org.mariadb:r2dbc-mariadb` driver, see
+@extref:[MariaDB in Apache Pekko Persistence R2DBC](pekko-persistence-r2dbc:getting-started.html#mariadb) for how to
+configure the connection and the driver dependency.
+
+The offset store uses the same tables as MySQL. The MariaDB driver doesn't support `java.time.Instant`, so the
+connection factory of Apache Pekko Persistence R2DBC binds and reads timestamps as `LocalDateTime` in UTC, with a UTC
+database session. The same applies to the statements of an `R2dbcHandler` that run on the `R2dbcSession`.
+
+### H2
+
+With the `h2` dialect the database runs in the same JVM process, see
+@extref:[H2 in Apache Pekko Persistence R2DBC](pekko-persistence-r2dbc:getting-started.html#h2) for how to configure the
+connection and the H2 dependencies.
+
+Like the persistence plugins, a projection creates the offset store tables of its own `offset-store` settings when it is
+first used, so no schema script is needed. That includes the `schema` and table names of the projection settings and a
+separate `use-connection-factory`. The `projection_offset_store` table is not created when `offset-table` is set to
+`""`. The tables are not created when the connection is configured with an H2 `url`.
+
+The database calls of the projection run on the `h2.use-dispatcher` of the connection factory, since the H2 driver
+blocks.
+
 ## Configuration
 
 By default, `pekko-projection-r2dbc` uses the same connection pool and `dialect` as `pekko-persistence-r2dbc`, see
diff --git a/grpc-int-test/src/test/scala/org/apache/pekko/projection/grpc/TestDbLifecycle.scala b/grpc-int-test/src/test/scala/org/apache/pekko/projection/grpc/TestDbLifecycle.scala
index 316c334b..6dcfee7d 100644
--- a/grpc-int-test/src/test/scala/org/apache/pekko/projection/grpc/TestDbLifecycle.scala
+++ b/grpc-int-test/src/test/scala/org/apache/pekko/projection/grpc/TestDbLifecycle.scala
@@ -17,6 +17,7 @@ import org.apache.pekko
 import pekko.actor.typed.ActorSystem
 import pekko.persistence.Persistence
 import pekko.persistence.r2dbc.ConnectionFactoryProvider
+import pekko.persistence.r2dbc.ConnectionFactorySettings
 import pekko.persistence.r2dbc.JournalSettings
 import pekko.persistence.r2dbc.internal.R2dbcExecutor
 import pekko.projection.r2dbc.R2dbcProjectionSettings
@@ -40,7 +41,11 @@ trait TestDbLifecycle extends BeforeAndAfterAll { this: Suite =>
     new R2dbcExecutor(
       ConnectionFactoryProvider(typedSystem).connectionFactoryFor(r2dbcProjectionSettings.useConnectionFactory),
       LoggerFactory.getLogger(getClass),
-      r2dbcProjectionSettings.logDbCallsExceeding)(typedSystem.executionContext, typedSystem)
+      r2dbcProjectionSettings.logDbCallsExceeding,
+      ConnectionFactorySettings.closeCallsExceeding(
+        typedSystem.settings.config.getConfig(r2dbcProjectionSettings.useConnectionFactory)))(
+      typedSystem.executionContext,
+      typedSystem)
   }
 
   lazy val persistenceExt: Persistence = Persistence(typedSystem)
diff --git a/project/Dependencies.scala b/project/Dependencies.scala
index 4c2c84ab..4c24627b 100644
--- a/project/Dependencies.scala
+++ b/project/Dependencies.scala
@@ -69,6 +69,11 @@ object Dependencies {
     val r2dbcPool = "io.r2dbc" % "r2dbc-pool" % "1.0.2.RELEASE"
     val r2dbcPostgres = "org.postgresql" % "r2dbc-postgresql" % "1.1.3.RELEASE"
     val r2dbcMysql = "io.asyncer" % "r2dbc-mysql" % "1.4.3"
+    // the version that pekko-persistence-r2dbc is built against
+    val r2dbcMariadb = "org.mariadb" % "r2dbc-mariadb" % "1.3.2" // ApacheV2
+    val r2dbcH2 = "io.r2dbc" % "r2dbc-h2" % "1.1.0.RELEASE" // ApacheV2
+    // the H2 version that pekko-persistence-r2dbc is built against
+    val h2R2dbc = "com.h2database" % "h2" % "2.5.252" // EPL 1.0 or MPL 2.0
   }
 
   object Test {
@@ -114,6 +119,9 @@ object Dependencies {
 
     val r2dbcPostgres = Compile.r2dbcPostgres % "test"
     val r2dbcMysql = Compile.r2dbcMysql % "test"
+    val r2dbcMariadb = Compile.r2dbcMariadb % "test"
+    val r2dbcH2 = Compile.r2dbcH2 % "test"
+    val h2R2dbc = Compile.h2R2dbc % "test"
   }
 
   object Examples {
@@ -276,6 +284,9 @@ object Dependencies {
       Compile.r2dbcPool,
       Test.r2dbcPostgres,
       Test.r2dbcMysql,
+      Test.r2dbcMariadb,
+      Test.r2dbcH2,
+      Test.h2R2dbc,
       Test.pekkoSerializationJackson,
       Test.pekkoDiscovery,
       Test.pekkoDistributedData,
diff --git a/project/PekkoPersistenceR2DBCDependency.scala b/project/PekkoPersistenceR2DBCDependency.scala
index 3a3e2835..41619ff5 100644
--- a/project/PekkoPersistenceR2DBCDependency.scala
+++ b/project/PekkoPersistenceR2DBCDependency.scala
@@ -21,5 +21,5 @@ import com.github.pjfanning.pekkobuild.PekkoDependency
 object PekkoPersistenceR2DBCDependency extends PekkoDependency {
   override val checkProject: String = "pekko-persistence-r2dbc"
   override val module: Option[String] = None
-  override val currentVersion: String = "2.0.0-M1"
+  override val currentVersion: String = "2.0.0-M1+120-5a212bab-SNAPSHOT"
 }
diff --git a/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/H2OffsetStoreTablesSpec.scala b/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/H2OffsetStoreTablesSpec.scala
new file mode 100644
index 00000000..e20bb84a
--- /dev/null
+++ b/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/H2OffsetStoreTablesSpec.scala
@@ -0,0 +1,161 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.projection.r2dbc
+
+import java.util.UUID
+
+import org.apache.pekko
+import pekko.actor.testkit.typed.scaladsl.LogCapturing
+import pekko.actor.testkit.typed.scaladsl.ScalaTestWithActorTestKit
+import pekko.persistence.r2dbc.ConnectionFactoryProvider
+import pekko.persistence.r2dbc.Dialect
+import pekko.projection.ProjectionId
+import pekko.projection.internal.ManagementState
+import pekko.projection.r2dbc.internal.OffsetPidSeqNr
+import pekko.projection.r2dbc.internal.R2dbcOffsetStore
+import pekko.projection.r2dbc.internal.R2dbcProjectionImpl
+import pekko.projection.r2dbc.internal.h2.H2R2dbcOffsetStore
+import com.typesafe.config.Config
+import com.typesafe.config.ConfigFactory
+import io.r2dbc.spi.R2dbcException
+import org.scalatest.wordspec.AnyWordSpecLike
+
+object H2OffsetStoreTablesSpec {
+
+  // independent of the dialect of the other tests, the H2 database runs in the same process
+  val config: Config = ConfigFactory
+    .parseString(s"""
+      h2-spec.connection-factory = $${pekko.persistence.r2dbc.connection-factory}
+      h2-spec.connection-factory {
+        driver = "h2"
+        h2.database = "h2-offset-store-tables-spec-${UUID.randomUUID()}"
+      }
+      h2-url-spec.connection-factory = $${pekko.persistence.r2dbc.connection-factory}
+      h2-url-spec.connection-factory {
+        driver = "h2"
+        h2.url = "mem:h2-offset-store-tables-url-spec-${UUID.randomUUID()};DB_CLOSE_DELAY=-1"
+      }
+      """)
+    .withFallback(ConfigFactory.load())
+    .resolve()
+}
+
+class H2OffsetStoreTablesSpec
+    extends ScalaTestWithActorTestKit(H2OffsetStoreTablesSpec.config)
+    with AnyWordSpecLike
+    with LogCapturing {
+
+  private def settings(overrides: String): R2dbcProjectionSettings =
+    R2dbcProjectionSettings(
+      ConfigFactory
+        .parseString(overrides)
+        .withFallback(ConfigFactory.parseString("""
+          dialect = "h2"
+          use-connection-factory = "h2-spec.connection-factory"
+          """))
+        .withFallback(system.settings.config.getConfig(R2dbcProjectionSettings.DefaultConfigPath)))
+
+  private def createOffsetStore(projectionId: ProjectionId, settings: R2dbcProjectionSettings): R2dbcOffsetStore = {
+    val connectionFactory = R2dbcProjectionImpl.connectionFactory(settings, ConfigFactory.empty())
+    val r2dbcExecutor = R2dbcProjectionImpl.createR2dbcExecutor(connectionFactory, settings, ConfigFactory.empty())
+    R2dbcProjectionImpl.createOffsetStore(projectionId, None, settings, r2dbcExecutor)
+  }
+
+  private def tablesInSchema(schema: String): Set[String] = {
+    // the plain connection factory, which doesn't create tables
+    val connectionFactory = ConnectionFactoryProvider(system).connectionFactoryFor("h2-spec.connection-factory")
+    val executor = R2dbcProjectionImpl.createR2dbcExecutor(connectionFactory, settings(""), ConfigFactory.empty())
+    executor
+      .select("tables in schema")(
+        _.createStatement("SELECT table_name FROM information_schema.tables WHERE table_schema = ?")
+          .bind(0, schema.toUpperCase),
+        row => row.get("table_name", classOf[String]).toLowerCase)
+      .futureValue
+      .toSet
+  }
+
+  private def genProjectionId() = ProjectionId(UUID.randomUUID().toString, "00")
+
+  "The H2 offset store" must {
+
+    "be created for the h2 dialect" in {
+      createOffsetStore(genProjectionId(), settings("")) shouldBe a[H2R2dbcOffsetStore]
+      settings("").dialect shouldBe Dialect.H2
+    }
+
+    "create the tables of its own schema and table names" in {
+      val projectionSettings = settings("""
+        offset-store {
+          schema = "h2_spec_custom"
+          offset-table = "custom_offset"
+          timestamp-offset-table = "custom_timestamp_offset"
+          management-table = "custom_management"
+        }
+        """)
+      val projectionId = genProjectionId()
+      val offsetStore = createOffsetStore(projectionId, projectionSettings)
+
+      offsetStore.saveOffset(OffsetPidSeqNr(1L)).futureValue
+      offsetStore.readOffset[Long]().futureValue shouldBe Some(1L)
+
+      tablesInSchema("h2_spec_custom") shouldBe Set("custom_offset", "custom_timestamp_offset", "custom_management")
+    }
+
+    "update offsets and management state with MERGE" in {
+      val offsetStore = createOffsetStore(genProjectionId(), settings(""))
+
+      offsetStore.saveOffset(OffsetPidSeqNr(1L)).futureValue
+      offsetStore.saveOffset(OffsetPidSeqNr(2L)).futureValue
+      offsetStore.readOffset[Long]().futureValue shouldBe Some(2L)
+
+      offsetStore.savePaused(paused = true).futureValue
+      offsetStore.readManagementState().futureValue shouldBe Some(ManagementState(paused = true))
+      offsetStore.savePaused(paused = false).futureValue
+      offsetStore.readManagementState().futureValue shouldBe Some(ManagementState(paused = false))
+    }
+
+    "not create the offset table when it is not used" in {
+      val projectionSettings = settings("""
+        offset-store {
+          schema = "h2_spec_no_offset_table"
+          offset-table = ""
+        }
+        """)
+      val offsetStore = createOffsetStore(genProjectionId(), projectionSettings)
+
+      offsetStore.savePaused(paused = true).futureValue
+
+      tablesInSchema("h2_spec_no_offset_table") shouldBe Set(
+        projectionSettings.timestampOffsetTable,
+        projectionSettings.managementTable)
+    }
+
+    "not create the tables when the connection is configured with a url" in {
+      val offsetStore = createOffsetStore(genProjectionId(),
+        settings("""
+        use-connection-factory = "h2-url-spec.connection-factory"
+        """))
+
+      val exc = offsetStore.readManagementState().failed.futureValue
+      exc shouldBe a[R2dbcException]
+      withClue(exc.toString) {
+        TestDbLifecycle.isH2TableNotFound(exc.asInstanceOf[R2dbcException]) shouldBe true
+      }
+    }
+  }
+}
diff --git a/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/MariaDBR2dbcOffsetStoreSpec.scala b/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/MariaDBR2dbcOffsetStoreSpec.scala
new file mode 100644
index 00000000..bbfb19e1
--- /dev/null
+++ b/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/MariaDBR2dbcOffsetStoreSpec.scala
@@ -0,0 +1,128 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.projection.r2dbc
+
+import java.time.Instant
+import java.util.UUID
+
+import org.apache.pekko
+import pekko.actor.testkit.typed.scaladsl.LogCapturing
+import pekko.actor.testkit.typed.scaladsl.ScalaTestWithActorTestKit
+import pekko.persistence.Persistence
+import pekko.persistence.query.TimestampOffset
+import pekko.persistence.r2dbc.Dialect
+import pekko.persistence.r2dbc.internal.R2dbcExecutor
+import pekko.projection.ProjectionId
+import pekko.projection.internal.ManagementState
+import pekko.projection.r2dbc.R2dbcTimestampOffsetStoreSpec.TestTimestampSourceProvider
+import pekko.projection.r2dbc.internal.OffsetPidSeqNr
+import pekko.projection.r2dbc.internal.R2dbcOffsetStore
+import pekko.projection.r2dbc.internal.R2dbcProjectionImpl
+import pekko.projection.r2dbc.internal.mariadb.MariaDBR2dbcOffsetStore
+import com.typesafe.config.ConfigFactory
+import org.scalatest.wordspec.AnyWordSpecLike
+
+class MariaDBR2dbcOffsetStoreSpec
+    extends ScalaTestWithActorTestKit(TestConfig.config)
+    with AnyWordSpecLike
+    with LogCapturing {
+
+  private val settings = R2dbcProjectionSettings(system)
+
+  private def assumeMariaDB(): Unit =
+    assume(settings.dialect == Dialect.MariaDB, "requires a MariaDB database, run with the mariadb dialect")
+
+  private def createR2dbcExecutor(settings: R2dbcProjectionSettings): R2dbcExecutor =
+    R2dbcProjectionImpl.createR2dbcExecutor(
+      R2dbcProjectionImpl.connectionFactory(settings, ConfigFactory.empty()),
+      settings,
+      ConfigFactory.empty())
+
+  private def createOffsetStore(projectionId: ProjectionId, settings: R2dbcProjectionSettings): R2dbcOffsetStore = {
+    val r2dbcExecutor = createR2dbcExecutor(settings)
+    val sourceProvider =
+      new TestTimestampSourceProvider(0, Persistence(system).numberOfSlices - 1, TestClock.nowMicros())
+    R2dbcOffsetStore.fromConfig(projectionId, Some(sourceProvider), system, settings, r2dbcExecutor)
+  }
+
+  private def genProjectionId() = ProjectionId(UUID.randomUUID().toString, "00")
+
+  "The MariaDB offset store" must {
+
+    "be created for the mariadb dialect, with upserts that don't use the MySQL row alias" in {
+      // no database calls, so this runs with every dialect
+      val offsetStore = createOffsetStore(genProjectionId(), settings.withDialect(Dialect.MariaDB)) match {
+        case mariaDB: MariaDBR2dbcOffsetStore => mariaDB
+        case other                            => fail(s"Expected MariaDBR2dbcOffsetStore, but was [${other.getClass.getName}]")
+      }
+
+      (offsetStore.upsertOffsetSql should not).include("AS excluded")
+      offsetStore.upsertOffsetSql should include("current_offset = VALUES(current_offset)")
+      (offsetStore.updateManagementStateSql should not).include("AS excluded")
+      offsetStore.updateManagementStateSql should include("paused = VALUES(paused)")
+    }
+
+    "update offsets and management state" in {
+      assumeMariaDB()
+      val offsetStore = createOffsetStore(genProjectionId(), settings)
+
+      offsetStore.saveOffset(OffsetPidSeqNr(1L)).futureValue
+      offsetStore.saveOffset(OffsetPidSeqNr(2L)).futureValue
+      offsetStore.readOffset[Long]().futureValue shouldBe Some(2L)
+
+      offsetStore.savePaused(paused = true).futureValue
+      offsetStore.readManagementState().futureValue shouldBe Some(ManagementState(paused = true))
+      offsetStore.savePaused(paused = false).futureValue
+      offsetStore.readManagementState().futureValue shouldBe Some(ManagementState(paused = false))
+    }
+
+    "store timestamp offsets as the same instant, independent of the time zone of the server" in {
+      assumeMariaDB()
+      // The MariaDB container of the tests runs in America/New_York. Instants are bound as their UTC local time,
+      // which the server interprets in the session time zone. With a New York session, 01:30 on 2026-11-01 would be
+      // stored 5 hours later and still read back unchanged, so the stored instant is checked with UNIX_TIMESTAMP,
+      // which doesn't depend on the session time zone. 02:30 on 2027-03-14 doesn't exist in New York, the server
+      // would reject it.
+      val r2dbcExecutor = createR2dbcExecutor(settings)
+      val timestamps = Seq(Instant.parse("2026-11-01T01:30:00.123456Z"), Instant.parse("2027-03-14T02:30:00.654321Z"))
+
+      timestamps.zipWithIndex.foreach { case (timestamp, i) =>
+        val projectionId = genProjectionId()
+        val offsetStore = createOffsetStore(projectionId, settings)
+        val offset = TimestampOffset(timestamp, Map("p1" -> (i + 1L)))
+        offsetStore.saveOffset(OffsetPidSeqNr(offset, "p1", i + 1L)).futureValue
+        withClue(s"timestamp [$timestamp]: ") {
+          val storedEpochSeconds = r2dbcExecutor
+            .selectOne("select stored timestamp offset")(
+              _.createStatement(
+                s"SELECT UNIX_TIMESTAMP(timestamp_offset) AS epoch_seconds FROM ${settings.timestampOffsetTableWithSchema} " +
+                "WHERE projection_name = ?")
+                .bind(0, projectionId.name),
+              row => row.get("epoch_seconds", classOf[java.math.BigDecimal]))
+            .futureValue
+          storedEpochSeconds.map(BigDecimal(_)) shouldBe Some(
+            BigDecimal(timestamp.getEpochSecond) + BigDecimal(timestamp.getNano) / 1000000000)
+
+          // a new offset store reads the offset from the database
+          createOffsetStore(projectionId, settings).readOffset[TimestampOffset]().futureValue shouldBe Some(
+            offset)
+        }
+      }
+    }
+  }
+}
diff --git a/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/R2dbcProjectionSpec.scala b/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/R2dbcProjectionSpec.scala
index 376bb3f7..ba0338f2 100644
--- a/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/R2dbcProjectionSpec.scala
+++ b/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/R2dbcProjectionSpec.scala
@@ -98,8 +98,9 @@ object R2dbcProjectionSpec {
 
     def createTableSql(dialect: Dialect): String = {
       val quotedTable = dialect match {
-        case Dialect.MySQL => table
-        case _             => s""""$table""""
+        // H2 would make an unquoted name upper case, so it must not be quoted in one place only
+        case Dialect.MySQL | Dialect.MariaDB | Dialect.H2 => table
+        case _                                            => s""""$table""""
       }
       s"""|CREATE table IF NOT EXISTS $quotedTable (
           |  id VARCHAR(255) NOT NULL,
@@ -157,6 +158,17 @@ object R2dbcProjectionSpec {
               id = excluded.id,
               concatenated = excluded.concatenated
           """
+        case Dialect.MariaDB =>
+          sql"""
+            INSERT INTO $table (id, concatenated)  VALUES (?, ?)
+            ON DUPLICATE KEY UPDATE
+              id = VALUES(id),
+              concatenated = VALUES(concatenated)
+          """
+        case Dialect.H2 =>
+          sql"""
+            MERGE INTO $table (id, concatenated) KEY(id) VALUES (?, ?)
+          """
       }
       val stmt = session
         .createStatement(stmtSql)
diff --git a/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/R2dbcTimestampOffsetProjectionSpec.scala b/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/R2dbcTimestampOffsetProjectionSpec.scala
index 92ba4c94..3721f495 100644
--- a/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/R2dbcTimestampOffsetProjectionSpec.scala
+++ b/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/R2dbcTimestampOffsetProjectionSpec.scala
@@ -53,6 +53,8 @@ import pekko.projection.internal.NoopStatusObserver
 import pekko.projection.internal.SingleHandlerStrategy
 import pekko.projection.r2dbc.internal.R2dbcOffsetStore
 import pekko.projection.r2dbc.internal.R2dbcProjectionImpl
+import pekko.projection.r2dbc.internal.h2.H2R2dbcOffsetStore
+import pekko.projection.r2dbc.internal.mariadb.MariaDBR2dbcOffsetStore
 import pekko.projection.r2dbc.internal.mysql.MySQLR2dbcOffsetStore
 import pekko.projection.r2dbc.scaladsl.R2dbcHandler
 import pekko.projection.r2dbc.scaladsl.R2dbcProjection
@@ -1479,6 +1481,20 @@ class R2dbcTimestampOffsetProjectionSpec
               super.deleteOldTimestampOffsets()
             }
           }
+        case Dialect.MariaDB =>
+          new MariaDBR2dbcOffsetStore(projectionId, Some(sourceProvider), system, spySettings, r2dbcExecutor) {
+            override def deleteOldTimestampOffsets(): Future[Long] = {
+              deleteCount.incrementAndGet()
+              super.deleteOldTimestampOffsets()
+            }
+          }
+        case Dialect.H2 =>
+          new H2R2dbcOffsetStore(projectionId, Some(sourceProvider), system, spySettings, r2dbcExecutor) {
+            override def deleteOldTimestampOffsets(): Future[Long] = {
+              deleteCount.incrementAndGet()
+              super.deleteOldTimestampOffsets()
+            }
+          }
       }
     }
 
diff --git a/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/RuntimePluginConfigSpec.scala b/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/RuntimePluginConfigSpec.scala
index 3fe20557..fcc92acc 100644
--- a/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/RuntimePluginConfigSpec.scala
+++ b/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/RuntimePluginConfigSpec.scala
@@ -22,9 +22,7 @@ package org.apache.pekko.projection.r2dbc
 import java.util.UUID
 
 import scala.collection.immutable.ListSet
-import scala.concurrent.Await
 import scala.concurrent.Future
-import scala.concurrent.duration.DurationInt
 
 import org.apache.pekko
 import pekko.Done
@@ -36,6 +34,7 @@ import pekko.actor.typed.Behavior
 import pekko.persistence.Persistence
 import pekko.persistence.query.typed.EventEnvelope
 import pekko.persistence.r2dbc.ConnectionFactoryProvider
+import pekko.persistence.r2dbc.ConnectionFactorySettings
 import pekko.persistence.r2dbc.JournalSettings
 import pekko.persistence.r2dbc.SnapshotSettings
 import pekko.persistence.r2dbc.internal.R2dbcExecutor
@@ -75,6 +74,8 @@ object RuntimePluginConfigSpec {
               $configKey = {
                 connection-factory {
                   database = "$database"
+                  # one in-memory database per plugin, like the separate databases of the other dialects
+                  h2.database = "$database"
                 }
 
                 journal.$configKey.connection-factory = $${$configKey.connection-factory}
@@ -157,32 +158,20 @@ class RuntimePluginConfigSpec extends ScalaTestWithActorTestKit(TestConfig.confi
         new R2dbcExecutor(
           connectionFactoryProvider,
           LoggerFactory.getLogger(getClass),
-          journalSettings.logDbCallsExceeding)(system.executionContext, system)
+          journalSettings.logDbCallsExceeding,
+          ConnectionFactorySettings.closeCallsExceeding(journalConfig.getConfig(journalSettings.useConnectionFactory)))(
+          system.executionContext,
+          system)
 
-      Await.result(
-        r2dbcExecutor.updateOne("beforeAll delete")(
-          _.createStatement(s"delete from ${journalSettings.journalTableWithSchema}")),
-        10.seconds)
-      Await.result(
-        r2dbcExecutor.updateOne("beforeAll delete")(
-          _.createStatement(s"delete from ${snapshotSettings.snapshotsTableWithSchema}")),
-        10.seconds)
+      val dialect = journalSettings.dialect
+      TestDbLifecycle.deleteAll(r2dbcExecutor, dialect, journalSettings.journalTableWithSchema)
+      TestDbLifecycle.deleteAll(r2dbcExecutor, dialect, snapshotSettings.snapshotsTableWithSchema)
 
       val r2dbcProjectionSettings = R2dbcProjectionSettings(eventSourced.projectionConfig, system)
-      if (r2dbcProjectionSettings.isOffsetTableDefined) {
-        Await.result(
-          r2dbcExecutor.updateOne("beforeAll delete")(
-            _.createStatement(s"delete from ${r2dbcProjectionSettings.offsetTableWithSchema}")),
-          10.seconds)
-      }
-      Await.result(
-        r2dbcExecutor.updateOne("beforeAll delete")(
-          _.createStatement(s"delete from ${r2dbcProjectionSettings.timestampOffsetTableWithSchema}")),
-        10.seconds)
-      Await.result(
-        r2dbcExecutor.updateOne("beforeAll delete")(
-          _.createStatement(s"delete from ${r2dbcProjectionSettings.managementTableWithSchema}")),
-        10.seconds)
+      if (r2dbcProjectionSettings.isOffsetTableDefined)
+        TestDbLifecycle.deleteAll(r2dbcExecutor, dialect, r2dbcProjectionSettings.offsetTableWithSchema)
+      TestDbLifecycle.deleteAll(r2dbcExecutor, dialect, r2dbcProjectionSettings.timestampOffsetTableWithSchema)
+      TestDbLifecycle.deleteAll(r2dbcExecutor, dialect, r2dbcProjectionSettings.managementTableWithSchema)
     }
   }
 
diff --git a/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/TestConfig.scala b/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/TestConfig.scala
index aadf9a42..68414afb 100644
--- a/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/TestConfig.scala
+++ b/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/TestConfig.scala
@@ -59,6 +59,32 @@ object TestConfig {
             use-app-timestamp = on
           }
           """)
+      case "mariadb" =>
+        ConfigFactory.parseString("""
+          pekko.persistence.r2dbc {
+            connection-factory {
+              driver = "mariadb"
+              host = "localhost"
+              port = 3306
+              user = "root"
+              password = "root"
+              database = "mysql"
+            }
+            db-timestamp-monotonic-increasing = on
+            use-app-timestamp = on
+          }
+          """)
+      case "h2" =>
+        ConfigFactory.parseString("""
+          pekko.persistence.r2dbc {
+            connection-factory {
+              driver = "h2"
+              h2.protocol = "mem"
+            }
+            db-timestamp-monotonic-increasing = on
+            use-app-timestamp = on
+          }
+          """)
     }
 
     // using load here so that connection-factory can be overridden
diff --git a/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/TestDbLifecycle.scala b/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/TestDbLifecycle.scala
index fd6cf94b..675f2b11 100644
--- a/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/TestDbLifecycle.scala
+++ b/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/TestDbLifecycle.scala
@@ -14,18 +14,43 @@
 package org.apache.pekko.projection.r2dbc
 
 import scala.concurrent.Await
+import scala.concurrent.ExecutionContext
 import scala.concurrent.duration._
 
 import org.apache.pekko
 import pekko.actor.typed.ActorSystem
 import pekko.persistence.Persistence
-import pekko.persistence.r2dbc.ConnectionFactoryProvider
+import pekko.persistence.r2dbc.Dialect
 import pekko.persistence.r2dbc.JournalSettings
 import pekko.persistence.r2dbc.StateSettings
 import pekko.persistence.r2dbc.internal.R2dbcExecutor
+import pekko.projection.r2dbc.internal.R2dbcProjectionImpl
+import com.typesafe.config.ConfigFactory
+import io.r2dbc.spi.R2dbcException
 import org.scalatest.BeforeAndAfterAll
 import org.scalatest.Suite
-import org.slf4j.LoggerFactory
+
+object TestDbLifecycle {
+
+  // H2 error codes of a table that doesn't exist: TABLE_OR_VIEW_NOT_FOUND_1, _WITH_CANDIDATES_2 and _DATABASE_EMPTY_1
+  def isH2TableNotFound(e: R2dbcException): Boolean =
+    Set(42102, 42103, 42104).contains(e.getErrorCode)
+
+  /**
+   * Deletes all rows of `table`. The H2 persistence plugins create their tables when they are first used, which can
+   * be after a test harness deletes from them, so with H2 a table that doesn't exist yet counts as empty.
+   */
+  def deleteAll(r2dbcExecutor: R2dbcExecutor, dialect: Dialect, table: String): Unit = {
+    val deleted = r2dbcExecutor.updateOne("beforeAll delete")(_.createStatement(s"delete from $table"))
+    val result =
+      if (dialect == Dialect.H2)
+        deleted.recover {
+          case e: R2dbcException if isH2TableNotFound(e) => 0L
+        }(ExecutionContext.parasitic)
+      else deleted
+    Await.result(result, 10.seconds)
+  }
+}
 
 trait TestDbLifecycle extends BeforeAndAfterAll { this: Suite =>
 
@@ -36,12 +61,12 @@ trait TestDbLifecycle extends BeforeAndAfterAll { this: Suite =>
   lazy val r2dbcProjectionSettings: R2dbcProjectionSettings =
     R2dbcProjectionSettings(typedSystem.settings.config.getConfig(testConfigPath))
 
-  lazy val r2dbcExecutor: R2dbcExecutor = {
-    new R2dbcExecutor(
-      ConnectionFactoryProvider(typedSystem).connectionFactoryFor(r2dbcProjectionSettings.useConnectionFactory),
-      LoggerFactory.getLogger(getClass),
-      r2dbcProjectionSettings.logDbCallsExceeding)(typedSystem.executionContext, typedSystem)
-  }
+  // same connection factory and executor as the projections, which creates the offset store tables with H2
+  lazy val r2dbcExecutor: R2dbcExecutor =
+    R2dbcProjectionImpl.createR2dbcExecutor(
+      R2dbcProjectionImpl.connectionFactory(r2dbcProjectionSettings, ConfigFactory.empty())(typedSystem),
+      r2dbcProjectionSettings,
+      ConfigFactory.empty())(typedSystem)
 
   lazy val persistenceExt: Persistence = Persistence(typedSystem)
 
@@ -50,28 +75,13 @@ trait TestDbLifecycle extends BeforeAndAfterAll { this: Suite =>
       new JournalSettings(typedSystem.settings.config.getConfig("pekko.persistence.r2dbc.journal"))
     val stateSettings: StateSettings =
       new StateSettings(typedSystem.settings.config.getConfig("pekko.persistence.r2dbc.state"))
-    Await.result(
-      r2dbcExecutor.updateOne("beforeAll delete")(
-        _.createStatement(s"delete from ${journalSettings.journalTableWithSchema}")),
-      10.seconds)
-    Await.result(
-      r2dbcExecutor.updateOne("beforeAll delete")(
-        _.createStatement(s"delete from ${stateSettings.durableStateTableWithSchema}")),
-      10.seconds)
-    if (r2dbcProjectionSettings.isOffsetTableDefined) {
-      Await.result(
-        r2dbcExecutor.updateOne("beforeAll delete")(
-          _.createStatement(s"delete from ${r2dbcProjectionSettings.offsetTableWithSchema}")),
-        10.seconds)
-    }
-    Await.result(
-      r2dbcExecutor.updateOne("beforeAll delete")(
-        _.createStatement(s"delete from ${r2dbcProjectionSettings.timestampOffsetTableWithSchema}")),
-      10.seconds)
-    Await.result(
-      r2dbcExecutor.updateOne("beforeAll delete")(
-        _.createStatement(s"delete from ${r2dbcProjectionSettings.managementTableWithSchema}")),
-      10.seconds)
+    val dialect = r2dbcProjectionSettings.dialect
+    TestDbLifecycle.deleteAll(r2dbcExecutor, dialect, journalSettings.journalTableWithSchema)
+    TestDbLifecycle.deleteAll(r2dbcExecutor, dialect, stateSettings.durableStateTableWithSchema)
+    if (r2dbcProjectionSettings.isOffsetTableDefined)
+      TestDbLifecycle.deleteAll(r2dbcExecutor, dialect, r2dbcProjectionSettings.offsetTableWithSchema)
+    TestDbLifecycle.deleteAll(r2dbcExecutor, dialect, r2dbcProjectionSettings.timestampOffsetTableWithSchema)
+    TestDbLifecycle.deleteAll(r2dbcExecutor, dialect, r2dbcProjectionSettings.managementTableWithSchema)
     super.beforeAll()
   }
 
diff --git a/r2dbc/src/main/resources/reference.conf b/r2dbc/src/main/resources/reference.conf
index 7ca39ced..d506709d 100644
--- a/r2dbc/src/main/resources/reference.conf
+++ b/r2dbc/src/main/resources/reference.conf
@@ -5,7 +5,7 @@
 
 //#projection-config
 pekko.projection.r2dbc {
-  # postgres, yugabyte or mysql
+  # postgres, yugabyte, mysql, mariadb or h2
   dialect = ${pekko.persistence.r2dbc.dialect}
 
   offset-store {
@@ -40,7 +40,7 @@ pekko.projection.r2dbc {
     # Batch size for inserts of timestamp offsets. Can be used to improve throughput for projections
     # with many events per persistence id and that is using `groupedWithinAsync`.
     # For Postgres (and Yugabyte) this will use multi-row inserts for this number of records. For MySQL
-    # this will use multiple single row inserts.
+    # and MariaDB this will use multiple single row inserts.
     # Use 0 to disable batching and always use single row inserts.
     offset-batch-size = 20
   }
diff --git a/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/internal/R2dbcOffsetStore.scala b/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/internal/R2dbcOffsetStore.scala
index d152645b..63d2525c 100644
--- a/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/internal/R2dbcOffsetStore.scala
+++ b/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/internal/R2dbcOffsetStore.scala
@@ -51,6 +51,8 @@ import pekko.projection.internal.OffsetSerialization
 import pekko.projection.internal.OffsetSerialization.MultipleOffsets
 import pekko.projection.internal.OffsetSerialization.SingleOffset
 import pekko.projection.r2dbc.R2dbcProjectionSettings
+import pekko.projection.r2dbc.internal.h2.H2R2dbcOffsetStore
+import pekko.projection.r2dbc.internal.mariadb.MariaDBR2dbcOffsetStore
 import pekko.projection.r2dbc.internal.mysql.MySQLR2dbcOffsetStore
 import io.r2dbc.spi.Connection
 import io.r2dbc.spi.Statement
@@ -193,6 +195,10 @@ private[projection] object R2dbcOffsetStore {
         new R2dbcOffsetStore(projectionId, sourceProvider, system, settings, r2dbcExecutor, clock)
       case Dialect.MySQL =>
         new MySQLR2dbcOffsetStore(projectionId, sourceProvider, system, settings, r2dbcExecutor, clock)
+      case Dialect.MariaDB =>
+        new MariaDBR2dbcOffsetStore(projectionId, sourceProvider, system, settings, r2dbcExecutor, clock)
+      case Dialect.H2 =>
+        new H2R2dbcOffsetStore(projectionId, sourceProvider, system, settings, r2dbcExecutor, clock)
     }
 
   final class RejectedEnvelope(message: String) extends IllegalStateException(message)
diff --git a/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/internal/R2dbcProjectionImpl.scala b/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/internal/R2dbcProjectionImpl.scala
index 7eb48d87..e02d6535 100644
--- a/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/internal/R2dbcProjectionImpl.scala
+++ b/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/internal/R2dbcProjectionImpl.scala
@@ -32,6 +32,9 @@ import pekko.persistence.query.DeletedDurableState
 import pekko.persistence.query.UpdatedDurableState
 import pekko.persistence.query.typed.EventEnvelope
 import pekko.persistence.query.typed.scaladsl.LoadEventQuery
+import pekko.persistence.r2dbc.ConnectionFactoryProvider
+import pekko.persistence.r2dbc.ConnectionFactorySettings
+import pekko.persistence.r2dbc.Dialect
 import pekko.persistence.r2dbc.internal.R2dbcExecutor
 import pekko.persistence.state.scaladsl.DurableStateStore
 import pekko.persistence.state.scaladsl.GetObjectResult
@@ -64,6 +67,7 @@ import pekko.projection.javadsl
 import pekko.projection.r2dbc.R2dbcProjectionSettings
 import pekko.projection.r2dbc.internal.R2dbcOffsetStore.RejectedEnvelope
 import pekko.projection.r2dbc.internal.R2dbcProjectionImpl.extractOffsetPidSeqNr
+import pekko.projection.r2dbc.internal.h2.H2OffsetStoreTables
 import pekko.projection.r2dbc.scaladsl.R2dbcHandler
 import pekko.projection.r2dbc.scaladsl.R2dbcSession
 import pekko.projection.scaladsl
@@ -72,6 +76,7 @@ import pekko.projection.scaladsl.SourceProvider
 import pekko.stream.RestartSettings
 import pekko.stream.scaladsl.FlowWithContext
 import pekko.stream.scaladsl.Source
+import com.typesafe.config.Config
 import io.r2dbc.spi.ConnectionFactory
 import org.slf4j.Logger
 import org.slf4j.LoggerFactory
@@ -92,12 +97,49 @@ private[projection] object R2dbcProjectionImpl {
       projectionId: ProjectionId,
       sourceProvider: Option[BySlicesSourceProvider],
       settings: R2dbcProjectionSettings,
-      connectionFactory: ConnectionFactory)(implicit system: ActorSystem[?]) = {
-    val r2dbcExecutor =
-      new R2dbcExecutor(connectionFactory, log, settings.logDbCallsExceeding)(system.executionContext, system)
+      r2dbcExecutor: R2dbcExecutor)(implicit system: ActorSystem[?]): R2dbcOffsetStore =
     R2dbcOffsetStore.fromConfig(projectionId, sourceProvider, system, settings, r2dbcExecutor)
+
+  /**
+   * The connection factory of `settings.useConnectionFactory`. With the H2 dialect it creates the offset store
+   * tables before the first connection is handed out, since there is no schema script for an in-process database.
+   */
+  private[projection] def connectionFactory(settings: R2dbcProjectionSettings, config: Config)(
+      implicit system: ActorSystem[?]): ConnectionFactory = {
+    val connectionFactory =
+      ConnectionFactoryProvider(system).connectionFactoryFor(settings.useConnectionFactory, config)
+    settings.dialect match {
+      case Dialect.H2 =>
+        H2OffsetStoreTables.creatingTables(connectionFactory, connectionFactoryConfig(settings, config), settings)
+      case _ => connectionFactory
+    }
   }
 
+  /**
+   * With the H2 dialect the database calls run on the `h2.use-dispatcher` of the connection factory, because the
+   * H2 driver blocks.
+   */
+  private[projection] def createR2dbcExecutor(
+      connectionFactory: ConnectionFactory,
+      settings: R2dbcProjectionSettings,
+      config: Config,
+      logger: Logger = log)(implicit system: ActorSystem[?]): R2dbcExecutor = {
+    val cfConfig = connectionFactoryConfig(settings, config)
+    val executionContext = settings.dialect match {
+      case Dialect.H2 => H2OffsetStoreTables.executionContext(system, cfConfig)
+      case _          => system.executionContext
+    }
+    new R2dbcExecutor(
+      connectionFactory,
+      logger,
+      settings.logDbCallsExceeding,
+      ConnectionFactorySettings.closeCallsExceeding(cfConfig))(executionContext, system)
+  }
+
+  private def connectionFactoryConfig(settings: R2dbcProjectionSettings, config: Config)(
+      implicit system: ActorSystem[?]): Config =
+    config.withFallback(system.settings.config).getConfig(settings.useConnectionFactory)
+
   private val loadEnvelopeCounter = new AtomicLong
 
   def loadEnvelope[Envelope](env: Envelope, sourceProvider: SourceProvider[?, Envelope])(
diff --git a/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/internal/h2/H2OffsetStoreTables.scala b/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/internal/h2/H2OffsetStoreTables.scala
new file mode 100644
index 00000000..69dab89b
--- /dev/null
+++ b/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/internal/h2/H2OffsetStoreTables.scala
@@ -0,0 +1,133 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+/*
+ * Copyright (C) 2022 - 2023 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package org.apache.pekko.projection.r2dbc.internal.h2
+
+import java.time.Duration
+
+import scala.concurrent.ExecutionContext
+import scala.jdk.CollectionConverters._
+
+import org.apache.pekko
+import pekko.actor.typed.ActorSystem
+import pekko.actor.typed.DispatcherSelector
+import pekko.annotation.InternalApi
+import pekko.projection.r2dbc.R2dbcProjectionSettings
+import com.typesafe.config.Config
+import io.r2dbc.spi.Connection
+import io.r2dbc.spi.ConnectionFactory
+import io.r2dbc.spi.ConnectionFactoryMetadata
+import io.r2dbc.spi.Result
+import org.reactivestreams.Publisher
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
+
+/**
+ * INTERNAL API
+ *
+ * The H2 database runs in the same process, so its tables can't be created up front with a schema script. Like the
+ * pekko-persistence-r2dbc plugins with the h2 dialect, the offset store creates the tables of its own settings before
+ * the first connection is handed out.
+ */
+@InternalApi
+private[projection] object H2OffsetStoreTables {
+
+  /**
+   * Wraps the connection factory so that the offset store tables of `settings` are created before the first
+   * connection is handed out. Not when the connection is configured with a `url`, same as the persistence plugins.
+   */
+  def creatingTables(
+      connectionFactory: ConnectionFactory,
+      connectionFactoryConfig: Config,
+      settings: R2dbcProjectionSettings): ConnectionFactory =
+    if (connectionFactoryConfig.getString("h2.url").nonEmpty) connectionFactory
+    else new TableCreatingConnectionFactory(connectionFactory, ddl(settings))
+
+  /**
+   * The H2 R2DBC driver blocks in surprising places, so the database calls run on the `h2.use-dispatcher` of the
+   * connection factory, same as the persistence plugins.
+   */
+  def executionContext(system: ActorSystem[?], connectionFactoryConfig: Config): ExecutionContext =
+    system.dispatchers.lookup(DispatcherSelector.fromConfig(connectionFactoryConfig.getString("h2.use-dispatcher")))
+
+  def ddl(settings: R2dbcProjectionSettings): Seq[String] = {
+    val schema = settings.schema.map(s => s"CREATE SCHEMA IF NOT EXISTS $s").toSeq
+
+    val offsetTable =
+      if (settings.isOffsetTableDefined)
+        Seq(s"""CREATE TABLE IF NOT EXISTS ${settings.offsetTableWithSchema} (
+          projection_name VARCHAR(255) NOT NULL,
+          projection_key VARCHAR(255) NOT NULL,
+          current_offset VARCHAR(255) NOT NULL,
+          manifest VARCHAR(32) NOT NULL,
+          mergeable BOOLEAN NOT NULL,
+          last_updated BIGINT NOT NULL,
+          PRIMARY KEY(projection_name, projection_key)
+        )""")
+      else Seq.empty
+
+    val timestampOffsetTable =
+      s"""CREATE TABLE IF NOT EXISTS ${settings.timestampOffsetTableWithSchema} (
+          projection_name VARCHAR(255) NOT NULL,
+          projection_key VARCHAR(255) NOT NULL,
+          slice INT NOT NULL,
+          persistence_id VARCHAR(255) NOT NULL,
+          seq_nr BIGINT NOT NULL,
+          timestamp_offset timestamp with time zone NOT NULL,
+          timestamp_consumed timestamp with time zone NOT NULL,
+          PRIMARY KEY(slice, projection_name, timestamp_offset, persistence_id, seq_nr)
+        )"""
+
+    val managementTable =
+      s"""CREATE TABLE IF NOT EXISTS ${settings.managementTableWithSchema} (
+          projection_name VARCHAR(255) NOT NULL,
+          projection_key VARCHAR(255) NOT NULL,
+          paused BOOLEAN NOT NULL,
+          last_updated BIGINT NOT NULL,
+          PRIMARY KEY(projection_name, projection_key)
+        )"""
+
+    schema ++ offsetTable ++ Seq(timestampOffsetTable, managementTable)
+  }
+
+  /**
+   * Runs `ddl` once, on a connection of its own, before handing out the first connection. A failure is not cached,
+   * so the next `create()` runs `ddl` again, for example after a transient lock on a file database.
+   */
+  private final class TableCreatingConnectionFactory(underlying: ConnectionFactory, ddl: Seq[String])
+      extends ConnectionFactory {
+
+    private val tablesCreated: Mono[Void] =
+      Mono
+        .usingWhen[Void, Connection](
+          Mono.from[Connection](underlying.create()),
+          (connection: Connection) =>
+            Flux
+              .fromIterable(ddl.asJava)
+              .concatMap((statement: String) =>
+                Flux
+                  .from[Result](connection.createStatement(statement).execute())
+                  .concatMap((result: Result) => result.getRowsUpdated))
+              .`then`(),
+          (connection: Connection) => connection.close())
+        .cache(
+          (_: Void) => Duration.ofMillis(Long.MaxValue),
+          (_: Throwable) => Duration.ZERO,
+          () => Duration.ofMillis(Long.MaxValue))
+
+    override def create(): Publisher[? <: Connection] =
+      tablesCreated.`then`(Mono.from[Connection](underlying.create()))
+
+    override def getMetadata: ConnectionFactoryMetadata = underlying.getMetadata
+  }
+}
diff --git a/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/internal/h2/H2R2dbcOffsetStore.scala b/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/internal/h2/H2R2dbcOffsetStore.scala
new file mode 100644
index 00000000..442c250c
--- /dev/null
+++ b/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/internal/h2/H2R2dbcOffsetStore.scala
@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+/*
+ * Copyright (C) 2023 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package org.apache.pekko.projection.r2dbc.internal.h2
+
+import java.time.Clock
+
+import org.apache.pekko
+import pekko.actor.typed.ActorSystem
+import pekko.annotation.InternalApi
+import pekko.persistence.r2dbc.internal.R2dbcExecutor
+import pekko.persistence.r2dbc.internal.Sql.DialectInterpolation
+import pekko.projection.BySlicesSourceProvider
+import pekko.projection.ProjectionId
+import pekko.projection.r2dbc.R2dbcProjectionSettings
+import pekko.projection.r2dbc.internal.R2dbcOffsetStore
+
+/**
+ * INTERNAL API
+ */
+@InternalApi
+private[projection] class H2R2dbcOffsetStore(
+    projectionId: ProjectionId,
+    sourceProvider: Option[BySlicesSourceProvider],
+    system: ActorSystem[?],
+    settings: R2dbcProjectionSettings,
+    r2dbcExecutor: R2dbcExecutor,
+    clock: Clock = Clock.systemUTC())
+    extends R2dbcOffsetStore(projectionId, sourceProvider, system, settings, r2dbcExecutor, clock) {
+
+  // H2 has no transaction_timestamp(), CURRENT_TIMESTAMP is the start of the transaction
+  override lazy val timestampSql: String = "CURRENT_TIMESTAMP"
+
+  override val upsertOffsetSql: String = sql"""
+    MERGE INTO $offsetTable
+    (projection_name, projection_key, current_offset, manifest, mergeable, last_updated)
+    KEY(projection_name, projection_key)
+    VALUES (?,?,?,?,?,?)"""
+
+  override val updateManagementStateSql: String = sql"""
+    MERGE INTO $managementTable
+    (projection_name, projection_key, paused, last_updated)
+    KEY(projection_name, projection_key)
+    VALUES (?,?,?,?)"""
+}
diff --git a/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/internal/mariadb/MariaDBR2dbcOffsetStore.scala b/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/internal/mariadb/MariaDBR2dbcOffsetStore.scala
new file mode 100644
index 00000000..794b2b82
--- /dev/null
+++ b/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/internal/mariadb/MariaDBR2dbcOffsetStore.scala
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.projection.r2dbc.internal.mariadb
+
+import java.time.Clock
+
+import org.apache.pekko
+import pekko.actor.typed.ActorSystem
+import pekko.annotation.InternalApi
+import pekko.persistence.r2dbc.internal.R2dbcExecutor
+import pekko.persistence.r2dbc.internal.Sql.DialectInterpolation
+import pekko.projection.BySlicesSourceProvider
+import pekko.projection.ProjectionId
+import pekko.projection.r2dbc.R2dbcProjectionSettings
+import pekko.projection.r2dbc.internal.mysql.MySQLR2dbcOffsetStore
+
+/**
+ * INTERNAL API
+ *
+ * MariaDB uses the MySQL offset store, except for the upserts: MariaDB doesn't support the `AS excluded` row alias of
+ * MySQL, so the inserted values are referenced with `VALUES()`. `Instant` values are converted by the connection
+ * factory of pekko-persistence-r2dbc, since the MariaDB driver doesn't support them.
+ */
+@InternalApi
+private[projection] class MariaDBR2dbcOffsetStore(
+    projectionId: ProjectionId,
+    sourceProvider: Option[BySlicesSourceProvider],
+    system: ActorSystem[?],
+    settings: R2dbcProjectionSettings,
+    r2dbcExecutor: R2dbcExecutor,
+    clock: Clock = Clock.systemUTC())
+    extends MySQLR2dbcOffsetStore(projectionId, sourceProvider, system, settings, r2dbcExecutor, clock) {
+
+  override val upsertOffsetSql: String = sql"""
+    INSERT INTO $offsetTable
+    (projection_name, projection_key, current_offset, manifest, mergeable, last_updated)
+    VALUES (?,?,?,?,?,?)
+    ON DUPLICATE KEY UPDATE
+    current_offset = VALUES(current_offset),
+    manifest = VALUES(manifest),
+    mergeable = VALUES(mergeable),
+    last_updated = VALUES(last_updated)"""
+
+  override val updateManagementStateSql: String = sql"""
+    INSERT INTO $managementTable
+    (projection_name, projection_key, paused, last_updated)
+    VALUES (?,?,?,?)
+    ON DUPLICATE KEY UPDATE
+    paused = VALUES(paused),
+    last_updated = VALUES(last_updated)"""
+}
diff --git a/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/scaladsl/R2dbcProjection.scala b/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/scaladsl/R2dbcProjection.scala
index 9872a938..6441f921 100644
--- a/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/scaladsl/R2dbcProjection.scala
+++ b/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/scaladsl/R2dbcProjection.scala
@@ -17,8 +17,6 @@ import org.apache.pekko
 import pekko.Done
 import pekko.actor.typed.ActorSystem
 import pekko.annotation.ApiMayChange
-import pekko.persistence.r2dbc.ConnectionFactoryProvider
-import pekko.persistence.r2dbc.internal.R2dbcExecutor
 import pekko.projection.BySlicesSourceProvider
 import pekko.projection.ProjectionContext
 import pekko.projection.ProjectionId
@@ -40,7 +38,6 @@ import pekko.projection.scaladsl.SourceProvider
 import pekko.stream.scaladsl.FlowWithContext
 import com.typesafe.config.Config
 import com.typesafe.config.ConfigFactory
-import io.r2dbc.spi.ConnectionFactory
 
 @ApiMayChange
 object R2dbcProjection {
@@ -68,16 +65,14 @@ object R2dbcProjection {
       handler: () => R2dbcHandler[Envelope])(implicit
       system: ActorSystem[?]): ExactlyOnceProjection[Offset, Envelope] = {
     val r2dbcSettings = settings.getOrElse(R2dbcProjectionSettings(config, system))
-    val connFactory = connectionFactory(system, config, r2dbcSettings)
+    val connFactory = R2dbcProjectionImpl.connectionFactory(r2dbcSettings, config)
+    val r2dbcExecutor = R2dbcProjectionImpl.createR2dbcExecutor(connFactory, r2dbcSettings, config)
     val offsetStore =
       R2dbcProjectionImpl.createOffsetStore(
         projectionId,
         timestampOffsetBySlicesSourceProvider(sourceProvider),
         r2dbcSettings,
-        connFactory)
-    val r2dbcExecutor = new R2dbcExecutor(connFactory, R2dbcProjectionImpl.log, r2dbcSettings.logDbCallsExceeding)(
-      system.executionContext,
-      system)
+        r2dbcExecutor)
 
     val adaptedHandler =
       R2dbcProjectionImpl.adaptedHandlerForExactlyOnce(sourceProvider, handler, offsetStore, r2dbcExecutor)(
@@ -128,16 +123,14 @@ object R2dbcProjection {
       system: ActorSystem[?]): AtLeastOnceProjection[Offset, Envelope] = {
 
     val r2dbcSettings = settings.getOrElse(R2dbcProjectionSettings(config, system))
-    val connFactory = connectionFactory(system, config, r2dbcSettings)
+    val connFactory = R2dbcProjectionImpl.connectionFactory(r2dbcSettings, config)
+    val r2dbcExecutor = R2dbcProjectionImpl.createR2dbcExecutor(connFactory, r2dbcSettings, config)
     val offsetStore =
       R2dbcProjectionImpl.createOffsetStore(
         projectionId,
         timestampOffsetBySlicesSourceProvider(sourceProvider),
         r2dbcSettings,
-        connFactory)
-    val r2dbcExecutor = new R2dbcExecutor(connFactory, R2dbcProjectionImpl.log, r2dbcSettings.logDbCallsExceeding)(
-      system.executionContext,
-      system)
+        r2dbcExecutor)
 
     val adaptedHandler =
       R2dbcProjectionImpl.adaptedHandlerForAtLeastOnce(sourceProvider, handler, offsetStore, r2dbcExecutor)(
@@ -186,13 +179,13 @@ object R2dbcProjection {
       handler: () => Handler[Envelope])(implicit system: ActorSystem[?]): AtLeastOnceProjection[Offset, Envelope] = {
 
     val r2dbcSettings = settings.getOrElse(R2dbcProjectionSettings(config, system))
-    val connFactory = connectionFactory(system, config, r2dbcSettings)
+    val connFactory = R2dbcProjectionImpl.connectionFactory(r2dbcSettings, config)
     val offsetStore =
       R2dbcProjectionImpl.createOffsetStore(
         projectionId,
         timestampOffsetBySlicesSourceProvider(sourceProvider),
         r2dbcSettings,
-        connFactory)
+        R2dbcProjectionImpl.createR2dbcExecutor(connFactory, r2dbcSettings, config))
 
     val adaptedHandler =
       R2dbcProjectionImpl.adaptedHandlerForAtLeastOnceAsync(sourceProvider, handler, offsetStore)(
@@ -238,16 +231,14 @@ object R2dbcProjection {
       system: ActorSystem[?]): GroupedProjection[Offset, Envelope] = {
 
     val r2dbcSettings = settings.getOrElse(R2dbcProjectionSettings(config, system))
-    val connFactory = connectionFactory(system, config, r2dbcSettings)
+    val connFactory = R2dbcProjectionImpl.connectionFactory(r2dbcSettings, config)
+    val r2dbcExecutor = R2dbcProjectionImpl.createR2dbcExecutor(connFactory, r2dbcSettings, config)
     val offsetStore =
       R2dbcProjectionImpl.createOffsetStore(
         projectionId,
         timestampOffsetBySlicesSourceProvider(sourceProvider),
         r2dbcSettings,
-        connFactory)
-    val r2dbcExecutor = new R2dbcExecutor(connFactory, R2dbcProjectionImpl.log, r2dbcSettings.logDbCallsExceeding)(
-      system.executionContext,
-      system)
+        r2dbcExecutor)
 
     val adaptedHandler =
       R2dbcProjectionImpl.adaptedHandlerForGrouped(sourceProvider, handler, offsetStore, r2dbcExecutor)(
@@ -297,13 +288,13 @@ object R2dbcProjection {
       system: ActorSystem[?]): GroupedProjection[Offset, Envelope] = {
 
     val r2dbcSettings = settings.getOrElse(R2dbcProjectionSettings(config, system))
-    val connFactory = connectionFactory(system, config, r2dbcSettings)
+    val connFactory = R2dbcProjectionImpl.connectionFactory(r2dbcSettings, config)
     val offsetStore =
       R2dbcProjectionImpl.createOffsetStore(
         projectionId,
         timestampOffsetBySlicesSourceProvider(sourceProvider),
         r2dbcSettings,
-        connFactory)
+        R2dbcProjectionImpl.createR2dbcExecutor(connFactory, r2dbcSettings, config))
 
     val adaptedHandler =
       R2dbcProjectionImpl.adaptedHandlerForGroupedAsync(sourceProvider, handler, offsetStore)(
@@ -361,13 +352,13 @@ object R2dbcProjection {
       system: ActorSystem[?]): AtLeastOnceFlowProjection[Offset, Envelope] = {
 
     val r2dbcSettings = settings.getOrElse(R2dbcProjectionSettings(config, system))
-    val connFactory = connectionFactory(system, config, r2dbcSettings)
+    val connFactory = R2dbcProjectionImpl.connectionFactory(r2dbcSettings, config)
     val offsetStore =
       R2dbcProjectionImpl.createOffsetStore(
         projectionId,
         timestampOffsetBySlicesSourceProvider(sourceProvider),
         r2dbcSettings,
-        connFactory)
+        R2dbcProjectionImpl.createR2dbcExecutor(connFactory, r2dbcSettings, config))
 
     val adaptedHandler =
       R2dbcProjectionImpl.adaptedHandlerForFlow(sourceProvider, handler, offsetStore, r2dbcSettings)(system)
@@ -384,11 +375,6 @@ object R2dbcProjection {
       offsetStore)
   }
 
-  private def connectionFactory(system: ActorSystem[?], config: Config, r2dbcSettings: R2dbcProjectionSettings)
-      : ConnectionFactory = {
-    ConnectionFactoryProvider(system).connectionFactoryFor(r2dbcSettings.useConnectionFactory, config)
-  }
-
   private def timestampOffsetBySlicesSourceProvider(
       sourceProvider: SourceProvider[?, ?]): Option[BySlicesSourceProvider] = {
     sourceProvider match {
diff --git a/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/scaladsl/R2dbcSession.scala b/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/scaladsl/R2dbcSession.scala
index 398baa92..5bbd4547 100644
--- a/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/scaladsl/R2dbcSession.scala
+++ b/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/scaladsl/R2dbcSession.scala
@@ -24,6 +24,8 @@ import pekko.annotation.ApiMayChange
 import pekko.persistence.r2dbc.ConnectionFactoryProvider
 import pekko.persistence.r2dbc.internal.R2dbcExecutor
 import pekko.projection.r2dbc.R2dbcProjectionSettings
+import pekko.projection.r2dbc.internal.R2dbcProjectionImpl
+import com.typesafe.config.ConfigFactory
 import io.r2dbc.spi.Connection
 import io.r2dbc.spi.Row
 import io.r2dbc.spi.Statement
@@ -61,11 +63,12 @@ object R2dbcSession {
       connectionFactoryConfigPath: Option[String],
       fun: R2dbcSession => Future[A]): Future[A] =
     try {
-      val settings = R2dbcProjectionSettings(system)
-      val connectionFactory = ConnectionFactoryProvider(system).connectionFactoryFor(
-        connectionFactoryConfigPath.getOrElse(settings.useConnectionFactory))
+      val defaultSettings = R2dbcProjectionSettings(system)
+      val settings =
+        connectionFactoryConfigPath.fold(defaultSettings)(path => defaultSettings.withUseConnectionFactory(path))
+      val connectionFactory = ConnectionFactoryProvider(system).connectionFactoryFor(settings.useConnectionFactory)
       val r2dbcExecutor =
-        new R2dbcExecutor(connectionFactory, log, settings.logDbCallsExceeding)(system.executionContext, system)
+        R2dbcProjectionImpl.createR2dbcExecutor(connectionFactory, settings, ConfigFactory.empty(), log)(system)
       r2dbcExecutor.withConnection("R2dbcSession") { connection =>
         fun(new R2dbcSession(connection)(system.executionContext, system))
       }