Support MySQL

.github/workflows/build-test.yml

@@ -84,7 +84,7 @@ jobs:
           docker exec -i docker-postgres-db-1 psql -U postgres -t < ddl-scripts/create_tables_postgres.sql
 
       - name: test
-        run: sbt ++${{ matrix.SCALA_VERSION }} test
+        run: sbt ++${{ matrix.SCALA_VERSION }} postgres/test
 
   test-yugabyte:
     name: Run tests with Yugabyte
@@ -130,7 +130,62 @@ jobs:
           docker exec -i yb-tserver-n1 /home/yugabyte/bin/ysqlsh -h yb-tserver-n1 -t < ddl-scripts/create_tables_yugabyte.sql
 
       - name: test
-        run: sbt -Dpekko.persistence.r2dbc.dialect=yugabyte -Dpekko.projection.r2dbc.dialect=yugabyte ++${{ matrix.SCALA_VERSION }} test
+        run: sbt -Dpekko.persistence.r2dbc.dialect=yugabyte -Dpekko.projection.r2dbc.dialect=yugabyte ++${{ matrix.SCALA_VERSION }} yugabyte/test
+
+  test-mysql:
+    name: Run test with MySQL
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        SCALA_VERSION: [ 2.12, 2.13, 3.3 ]
+        JAVA_VERSION: [ 11, 17, 21 ]
+        # only compiling on JDK 8 because of lower timestamp precision, which causes flakiness in certain tests
+        include:
+          - JAVA_VERSION: 8
+            SCALA_VERSION: 2.12
+            COMPILE_ONLY: true
+          - JAVA_VERSION: 8
+            SCALA_VERSION: 2.13
+            COMPILE_ONLY: true
+          - JAVA_VERSION: 8
+            SCALA_VERSION: 3.3
+            COMPILE_ONLY: true
+    if: github.repository == 'apache/pekko-persistence-r2dbc'
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+        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: Setup Java ${{ matrix.JAVA_VERSION }}
+        uses: actions/setup-java@v4
+        with:
+          distribution: temurin
+          java-version: ${{ matrix.JAVA_VERSION }}
+
+      - name: Install sbt
+        uses: sbt/setup-sbt@v1
+
+      - name: Cache Coursier cache
+        uses: coursier/cache-action@v6
+
+      - name: Enable jvm-opts
+        run: cp .jvmopts-ci .jvmopts
+
+      - name: Start DB
+        run: |-
+          docker compose -f docker/docker-compose-mysql.yml up -d --wait
+          docker exec -i docker-mysql-db-1 mysql -h 127.0.0.1 --user=root --password=root --database=mysql < ddl-scripts/create_tables_mysql.sql
+
+      - name: test
+        run: sbt ++${{ matrix.SCALA_VERSION }} mysql/${{ matrix.COMPILE_ONLY && 'Test/compile' || 'test' }}
 
   test-docs:
     name: Docs

build.sbt

@@ -45,16 +45,33 @@ lazy val root = (project in file("."))
   .settings(dontPublish)
   .settings(
     name := "pekko-persistence-r2dbc-root")
-  .aggregate(core, projection, migration, docs)
+  .aggregate(core, `core-mysql`, projection, migration, docs)
 
 def suffixFileFilter(suffix: String): FileFilter = new SimpleFileFilter(f => f.getAbsolutePath.endsWith(suffix))
 
+lazy val postgres = (project in file("postgres"))
+  .aggregate(core, projection, migration)
+
+lazy val yugabyte = (project in file("yugabyte"))
+  .aggregate(core, projection)
+
+lazy val mysql = (project in file("mysql"))
+  .aggregate(`core-mysql`, `projection-mysql`)
+
 lazy val core = (project in file("core"))
   .enablePlugins(ReproducibleBuildsPlugin)
   .settings(
     name := "pekko-persistence-r2dbc",
     libraryDependencies ++= Dependencies.core)
 
+lazy val `core-mysql` = (project in file("core-mysql"))
+  .dependsOn(core % "compile->compile;test->test") // TODO core-testkit would be ideal, though some classes are problematic to move
+  .enablePlugins(ReproducibleBuildsPlugin)
+  .disablePlugins(MimaPlugin) // TODO enable after future release
+  .settings(
+    name := "pekko-persistence-r2dbc-mysql",
+    libraryDependencies ++= Dependencies.`core-mysql`)
+
 lazy val projection = (project in file("projection"))
   .dependsOn(core)
   .enablePlugins(ReproducibleBuildsPlugin)
@@ -63,6 +80,15 @@ lazy val projection = (project in file("projection"))
     libraryDependencies ++= Dependencies.projection,
     dependencyOverrides ++= Dependencies.pekkoTestDependencyOverrides)
 
+lazy val `projection-mysql` = (project in file("projection-mysql"))
+  .dependsOn(`core-mysql` % "compile->compile;test->test", projection % "compile->compile;test->test") // TODO projection-testkit would be ideal, though some classes are problematic to move
+  .enablePlugins(ReproducibleBuildsPlugin)
+  .disablePlugins(MimaPlugin) // TODO enable after future release
+  .settings(
+    name := "pekko-projection-r2dbc-mysql",
+    libraryDependencies ++= Dependencies.projection,
+    dependencyOverrides ++= Dependencies.pekkoTestDependencyOverrides)
+
 lazy val migration = (project in file("migration"))
   .enablePlugins(ReproducibleBuildsPlugin)
   .settings(

core-mysql/src/main/resources/reference.conf

@@ -0,0 +1,31 @@
+# SPDX-License-Identifier: Apache-2.0
+
+// #connection-settings
+pekko.persistence.r2dbc {
+
+  # setting dialect to empty toggles use of the following class config
+  dialect = ""
+  journalDaoClass = "org.apache.pekko.persistence.r2dbc.mysql.journal.MySQLJournalDao"
+  queryDaoClass = "org.apache.pekko.persistence.r2dbc.mysql.query.MySQLQueryDao"
+  snapshotDaoClass = "org.apache.pekko.persistence.r2dbc.mysql.snapshot.MySQLSnapshotDao"
+  durableStateDaoClass = "org.apache.pekko.persistence.r2dbc.mysql.state.MySQLDurableStateDao"
+
+  schema = ""
+
+  connection-factory {
+    driver = "mysql"
+
+    host = "localhost"
+    port = 3306
+    database = "mysql"
+    user = "root"
+    password = "root"
+
+    connection-factory-options-customizer = "org.apache.pekko.persistence.r2dbc.mysql.MySQLConnectionFactoryOptionsCustomizer"
+  }
+
+  db-timestamp-monotonic-increasing = on
+
+  use-app-timestamp = on
+}
+// #connection-settings

core-mysql/src/main/scala/org/apache/pekko/persistence/r2dbc/mysql/MySQLConnectionFactoryOptionsCustomizer.scala

@@ -0,0 +1,25 @@
+/*
+ * 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.
+ */
+
+package org.apache.pekko.persistence.r2dbc.mysql
+
+import com.typesafe.config.Config
+import io.r2dbc.spi.ConnectionFactoryOptions
+import io.r2dbc.spi.Option
+import org.apache.pekko
+import pekko.actor.typed.ActorSystem
+import pekko.persistence.r2dbc.ConnectionFactoryProvider.ConnectionFactoryOptionsCustomizer
+
+class MySQLConnectionFactoryOptionsCustomizer(system: ActorSystem[_]) extends ConnectionFactoryOptionsCustomizer {
+  override def apply(builder: ConnectionFactoryOptions.Builder, config: Config): ConnectionFactoryOptions.Builder = {
+    // Either `connectionTimeZone = SERVER` or `forceConnectionTimeZoneToSession = true` need to be set for timezones to work correctly,
+    // likely caused by bug in https://github.com/asyncer-io/r2dbc-mysql/pull/240.
+    builder.option(Option.valueOf("connectionTimeZone"), "SERVER")
+  }
+}

core-mysql/src/main/scala/org/apache/pekko/persistence/r2dbc/mysql/journal/MySQLJournalDao.scala

@@ -0,0 +1,57 @@
+/*
+ * 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.
+ */
+
+package org.apache.pekko.persistence.r2dbc.mysql.journal
+
+import scala.concurrent.ExecutionContext
+import io.r2dbc.spi.ConnectionFactory
+import org.apache.pekko
+import pekko.actor.typed.ActorSystem
+import pekko.persistence.r2dbc.R2dbcSettings
+import pekko.persistence.r2dbc.internal.Sql
+import pekko.persistence.r2dbc.internal.Sql.ConfigurableInterpolation
+import pekko.persistence.r2dbc.journal.JournalDao
+import org.slf4j.Logger
+import org.slf4j.LoggerFactory
+
+object MySQLJournalDao {
+  val log: Logger = LoggerFactory.getLogger(classOf[MySQLJournalDao])
+
+  def settingRequirements(journalSettings: R2dbcSettings): Unit = {
+    // Application timestamps are used because MySQL does not have transaction_timestamp like Postgres. In future releases
+    // they could be tried to be emulated, but the benefits are questionable - no matter where the timestamps are generated,
+    // risk of clock skews remains.
+    require(journalSettings.useAppTimestamp,
+      "use-app-timestamp config must be on for MySQL support")
+    // Supporting the non-monotonic increasing timestamps by incrementing the timestamp within the insert queries based on
+    // latest row in the database seems to cause deadlocks when running tests like PersistTimestampSpec. Possibly this could
+    // be fixed.
+    require(journalSettings.dbTimestampMonotonicIncreasing,
+      "db-timestamp-monotonic-increasing config must be on for MySQL support")
+    // Also, missing RETURNING implementation makes grabbing the timestamp generated by the database less efficient - this
+    // applies for both of the requirements above.
+  }
+}
+
+class MySQLJournalDao(
+    journalSettings: R2dbcSettings,
+    connectionFactory: ConnectionFactory
+)(implicit ec: ExecutionContext, system: ActorSystem[_])
+    extends JournalDao(journalSettings, connectionFactory) {
+  MySQLJournalDao.settingRequirements(journalSettings)
+
+  override implicit lazy val sqlReplacements: Sql.Replacements = Sql.Replacements.None
+  override lazy val timestampSql: String = "NOW(6)"
+
+  override val insertEventWithParameterTimestampSql: String =
+    sql"INSERT INTO $journalTable " +
+    "(slice, entity_type, persistence_id, seq_nr, writer, adapter_manifest, event_ser_id, event_ser_manifest, " +
+    "event_payload, tags, meta_ser_id, meta_ser_manifest, meta_payload, db_timestamp) " +
+    s"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
+}

core-mysql/src/main/scala/org/apache/pekko/persistence/r2dbc/mysql/query/MySQLQueryDao.scala

@@ -0,0 +1,80 @@
+/*
+ * 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.
+ */
+
+package org.apache.pekko.persistence.r2dbc.mysql.query
+
+import java.time.Instant
+
+import scala.concurrent.ExecutionContext
+import scala.concurrent.Future
+import scala.concurrent.duration.Duration
+import scala.concurrent.duration.FiniteDuration
+import io.r2dbc.spi.ConnectionFactory
+import org.apache.pekko
+import pekko.actor.typed.ActorSystem
+import pekko.persistence.r2dbc.R2dbcSettings
+import pekko.persistence.r2dbc.internal.Sql
+import pekko.persistence.r2dbc.internal.Sql.ConfigurableInterpolation
+import pekko.persistence.r2dbc.query.scaladsl.QueryDao
+
+class MySQLQueryDao(
+    journalSettings: R2dbcSettings,
+    connectionFactory: ConnectionFactory
+)(implicit ec: ExecutionContext, system: ActorSystem[_]) extends QueryDao(journalSettings, connectionFactory) {
+
+  override implicit lazy val sqlReplacements: Sql.Replacements = Sql.Replacements.None
+  override lazy val statementTimestampSql: String = "NOW(6)"
+
+  override def eventsBySlicesRangeSql(
+      toDbTimestampParam: Boolean,
+      behindCurrentTime: FiniteDuration,
+      backtracking: Boolean,
+      minSlice: Int,
+      maxSlice: Int): String = {
+
+    def toDbTimestampParamCondition =
+      if (toDbTimestampParam) "AND db_timestamp <= ?" else ""
+
+    def behindCurrentTimeIntervalCondition =
+      if (behindCurrentTime > Duration.Zero)
+        s"AND db_timestamp < DATE_SUB($statementTimestampSql, INTERVAL '${behindCurrentTime.toMicros}' MICROSECOND)"
+      else ""
+
+    val selectColumns = {
+      if (backtracking)
+        s"SELECT slice, persistence_id, seq_nr, db_timestamp, $statementTimestampSql AS read_db_timestamp "
+      else
+        s"SELECT slice, persistence_id, seq_nr, db_timestamp, $statementTimestampSql AS read_db_timestamp, event_ser_id, event_ser_manifest, event_payload, meta_ser_id, meta_ser_manifest, meta_payload "
+    }
+
+    sql"""
+      $selectColumns
+      FROM $journalTable
+      WHERE entity_type = ?
+      AND slice BETWEEN $minSlice AND $maxSlice
+      AND db_timestamp >= ? $toDbTimestampParamCondition $behindCurrentTimeIntervalCondition
+      AND deleted = false
+      ORDER BY db_timestamp, seq_nr
+      LIMIT ?"""
+  }
+
+  override def selectBucketsSql(minSlice: Int, maxSlice: Int): String = {
+    sql"""
+      SELECT CAST(UNIX_TIMESTAMP(db_timestamp) AS SIGNED) / 10 AS bucket, count(*) AS count
+      FROM $journalTable
+      WHERE entity_type = ?
+      AND slice BETWEEN $minSlice AND $maxSlice
+      AND db_timestamp >= ? AND db_timestamp <= ?
+      AND deleted = false
+      GROUP BY bucket ORDER BY bucket LIMIT ?
+      """
+  }
+
+  override def currentDbTimestamp(): Future[Instant] = Future.successful(Instant.now())
+}

core-mysql/src/main/scala/org/apache/pekko/persistence/r2dbc/mysql/snapshot/MySQLSnapshotDao.scala

@@ -0,0 +1,39 @@
+/*
+ * 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.
+ */
+
+package org.apache.pekko.persistence.r2dbc.mysql.snapshot
+
+import scala.concurrent.ExecutionContext
+import io.r2dbc.spi.ConnectionFactory
+import org.apache.pekko
+import pekko.actor.typed.ActorSystem
+import pekko.persistence.r2dbc.R2dbcSettings
+import pekko.persistence.r2dbc.internal.Sql
+import pekko.persistence.r2dbc.snapshot.SnapshotDao
+
+class MySQLSnapshotDao(
+    settings: R2dbcSettings, connectionFactory: ConnectionFactory
+)(implicit ec: ExecutionContext, system: ActorSystem[_]) extends SnapshotDao(settings, connectionFactory) {
+
+  override implicit lazy val sqlReplacements: Sql.Replacements = Sql.Replacements.None
+
+  override val upsertSql = s"""
+    INSERT INTO $snapshotTable
+    (slice, entity_type, persistence_id, seq_nr, write_timestamp, snapshot, ser_id, ser_manifest, meta_payload, meta_ser_id, meta_ser_manifest)
+    VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) AS excluded
+    ON DUPLICATE KEY UPDATE
+      seq_nr = excluded.seq_nr,
+      write_timestamp = excluded.write_timestamp,
+      snapshot = excluded.snapshot,
+      ser_id = excluded.ser_id,
+      ser_manifest = excluded.ser_manifest,
+      meta_payload = excluded.meta_payload,
+      meta_ser_id = excluded.meta_ser_id,
+      meta_ser_manifest = excluded.meta_ser_manifest"""
+}