ptrdom commented on code in PR #517: URL: https://github.com/apache/pekko-persistence-r2dbc/pull/517#discussion_r4371548613
########## core/src/main/scala/org/apache/pekko/persistence/r2dbc/internal/h2/H2Connections.scala: ########## @@ -0,0 +1,165 @@ +/* + * 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.persistence.r2dbc.internal.h2 + +import java.util.Locale + +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 com.typesafe.config.Config +import io.r2dbc.h2.H2ConnectionConfiguration +import io.r2dbc.h2.H2ConnectionFactory +import io.r2dbc.h2.H2ConnectionOption +import io.r2dbc.spi.ConnectionFactory + +/** + * INTERNAL API + * + * Creates the H2 connection factory from the `h2` section of the connection factory config. Only used when the + * connection factory `driver` is `h2`, so that the optional H2 dependencies are not loaded otherwise. + */ +@InternalApi +private[r2dbc] object H2Connections { + + def create(config: Config): ConnectionFactory = + new H2ConnectionFactory(configuration(config)) + + def configuration(config: Config): H2ConnectionConfiguration = { + // starting point for both url and regular configs, + // to allow url to override anything but provide sane defaults + val builder = H2ConnectionConfiguration.builder() + val url = config.getString("url") + if (url.nonEmpty) { + builder.url(url) + } else { + val db = config.getString("database") + config.getString("protocol").toLowerCase(Locale.ROOT) match { + case "file" => builder.file(db) + case "mem" => builder.inMemory(db) + case unknown => + throw new IllegalArgumentException(s"Unknown H2 protocol [$unknown], supported protocols are [file, mem].") + } + + val createSliceIndexes = config.getBoolean("create-slice-indexes") + builder + // create schema on first connect + .property(H2ConnectionOption.INIT, dbSchema(config, createSliceIndexes, config.getString("additional-init"))) + // don't auto close connections + .property(H2ConnectionOption.DB_CLOSE_DELAY, "-1") + + // workaround for https://github.com/akka/akka-projection/issues/992 + builder.option("OPTIMIZE_REUSE_RESULTS=FALSE") + + if (config.getBoolean("trace-logging")) + // log to SLF4J instead of print to stdout, logger name will be 'h2database' + builder.property(H2ConnectionOption.TRACE_LEVEL_FILE, "4") + + // Arbitrary config pass through/override for non-url setup + config + .getConfig("additional-options") + .entrySet() + .iterator() + .asScala + .foreach(entry => builder.option(s"${entry.getKey}=${entry.getValue.unwrapped()}")) + } + + builder.build() + } + + /** + * The H2 R2DBC driver blocks in surprising places (Mono.toFuture in stmt.execute().asFuture()), so the DAOs run on + * the configured dispatcher. + */ + def executionContext(system: ActorSystem[?], connectionFactoryConfig: Config): ExecutionContext = + system.dispatchers.lookup( + DispatcherSelector.fromConfig(connectionFactoryConfig.getString("h2.use-dispatcher"))) + + private def dbSchema(config: Config, createSliceIndexes: Boolean, additionalInit: String): String = { Review Comment: These table names come from `connection-factory.h2` rather than from the plugin using the connection, so `journal.table = "j2"` starts fine and then fails with `Table "J2" not found`. See point 2; [80fe963](https://github.com/ptrdom/pekko-persistence-r2dbc/commit/80fe963) moves the DDL here into per-plugin functions that the DAOs run on first use. ########## core/src/main/resources/reference.conf: ########## @@ -334,6 +334,51 @@ pekko.persistence.r2dbc { # The class must have a constructor with a single parameter of type ActorSystem[_]. # If this setting is empty, the default no-op customizer will be used. connection-factory-options-customizer = "" + + # Settings that are only used when driver = "h2". The H2 database runs in the same JVM process, + # 'url', 'host', 'port', 'database', 'user', 'password', 'ssl' and the postgres specific settings + # above are not used. + // #connection-settings-h2 + h2 { + # the connection can be configured with a url, eg: 'r2dbc:h2:file:////some/path/db-file;DB_CLOSE_DELAY=-1' + # When a url is used the tables are not created automatically. + url = "" + # 'file' or 'mem', separate process over 'tcp' is not supported + protocol = "mem" + # For mem protocol, symbolic database name, all interactions with the same name in the same process see the same data. + # For file protocol, relative or absolute file path to where database is kept on disk. + # Note: for file based the file cannot be shared by multiple JVMs, it must be one unique path per process + database = "pekko-r2dbc" + # Enable trace logging via slf4j, logger used is named 'h2database' + # http://www.h2database.com/html/features.html#trace_options + # (Enabling may come with a performance overhead) + trace-logging = off + # DDL or DML included in init together with plugin table creation + # http://www.h2database.com/html/features.html#execute_sql_on_connection + additional-init = "" Review Comment: H2 runs `INIT` on every new connection, so this runs again whenever the pool opens one. Fine for `... IF NOT EXISTS` DDL, but DML here would repeat. On `h2-dao-tables` ([1401530](https://github.com/ptrdom/pekko-persistence-r2dbc/commit/14015308879ed067165762c592d5018ecf98ebfb)) it becomes a per-plugin `h2.additional-init`, which runs once after that plugin has created its tables. ########## .github/workflows/build-test.yml: ########## @@ -265,6 +265,45 @@ jobs: - name: test run: sbt -Dpekko.persistence.r2dbc.dialect=mysql ++${{ matrix.SCALA_VERSION }} ${{ matrix.COMPILE_ONLY && 'Test/compile' || 'test' }} + test-h2: + name: Run tests with H2 + runs-on: ubuntu-latest Review Comment: Could we also run this job on `windows-latest` (an OS axis, or one extra entry)? H2 needs no DB container, and Windows' coarse clock is where the timestamp issue in point 1 shows up. ########## core/src/main/scala/org/apache/pekko/persistence/r2dbc/query/scaladsl/h2/H2QueryDao.scala: ########## @@ -0,0 +1,90 @@ +/* + * 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.persistence.r2dbc.query.scaladsl.h2 + +import java.time.Instant + +import scala.concurrent.ExecutionContext +import scala.concurrent.Future +import scala.concurrent.duration.Duration +import scala.concurrent.duration.FiniteDuration +import org.apache.pekko +import pekko.actor.typed.ActorSystem +import pekko.annotation.InternalApi +import pekko.persistence.r2dbc.QuerySettings +import pekko.persistence.r2dbc.internal.InstantFactory +import pekko.persistence.r2dbc.internal.Sql.DialectInterpolation +import pekko.persistence.r2dbc.internal.h2.H2Utils +import pekko.persistence.r2dbc.query.scaladsl.QueryDao +import io.r2dbc.spi.ConnectionFactory +import io.r2dbc.spi.Row + +/** + * INTERNAL API + */ +@InternalApi +private[r2dbc] class H2QueryDao( + querySettings: QuerySettings, + connectionFactory: ConnectionFactory, + closeCallsExceeding: Option[FiniteDuration] = None +)(implicit ec: ExecutionContext, system: ActorSystem[?]) + extends QueryDao(querySettings, connectionFactory, closeCallsExceeding) { + + override lazy val statementTimestampSql: String = "CURRENT_TIMESTAMP" + + override protected def tagsFromRow(row: Row): Set[String] = + H2Utils.tagsFromDb(row, "tags") + + override protected def selectColumns(backtracking: Boolean): String = + if (backtracking) + s"SELECT slice, persistence_id, seq_nr, db_timestamp, $statementTimestampSql AS read_db_timestamp, tags, event_ser_id " + else + s"SELECT slice, persistence_id, seq_nr, db_timestamp, $statementTimestampSql AS read_db_timestamp, tags, event_ser_id, event_ser_manifest, event_payload, meta_ser_id, meta_ser_manifest, meta_payload " + + override protected def behindCurrentTimeIntervalCondition(behindCurrentTime: FiniteDuration): String = + if (behindCurrentTime > Duration.Zero) + s"AND db_timestamp < CURRENT_TIMESTAMP - interval '${behindCurrentTime.toMillis.toDouble / 1000}' second" + else "" + + override def eventsBySlicesRangeSql( + toDbTimestampParam: Boolean, + behindCurrentTime: FiniteDuration, + backtracking: Boolean, + minSlice: Int, + maxSlice: Int): String = + sql""" + ${selectColumns(backtracking)} + FROM $journalTable + WHERE entity_type = ? + AND slice BETWEEN $minSlice AND $maxSlice + AND db_timestamp >= ? ${toDbTimestampParamCondition(toDbTimestampParam)} ${behindCurrentTimeIntervalCondition( + behindCurrentTime)} + AND deleted = false + ORDER BY db_timestamp, seq_nr + LIMIT ?""" + + override def selectBucketsSql(minSlice: Int, maxSlice: Int): String = + sql""" + SELECT CAST(EXTRACT(EPOCH FROM db_timestamp) AS BIGINT) / 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 ? + """ + + // the database runs in the same process, app timestamp is db timestamp + override def currentDbTimestamp(): Future[Instant] = Future.successful(InstantFactory.now()) Review Comment: This bounds the by-slice queries, so it needs to be later than every completed write. `InstantFactory.now()` can return the same value as the last event's timestamp for about 0.5 ms on Windows. Suggested: `InstantFactory.nowStrictlyIncreasing()`, in the same commit as above. ########## core/src/main/scala/org/apache/pekko/persistence/r2dbc/state/scaladsl/h2/H2DurableStateDao.scala: ########## @@ -0,0 +1,90 @@ +/* + * 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.persistence.r2dbc.state.scaladsl.h2 + +import java.time.Instant + +import scala.concurrent.ExecutionContext +import scala.concurrent.Future +import scala.concurrent.duration.Duration +import scala.concurrent.duration.FiniteDuration +import org.apache.pekko +import pekko.actor.typed.ActorSystem +import pekko.annotation.InternalApi +import pekko.persistence.r2dbc.StateSettings +import pekko.persistence.r2dbc.internal.InstantFactory +import pekko.persistence.r2dbc.internal.Sql.DialectInterpolation +import pekko.persistence.r2dbc.internal.h2.H2Utils +import pekko.persistence.r2dbc.state.scaladsl.DurableStateDao +import io.r2dbc.spi.ConnectionFactory + +/** + * INTERNAL API + */ +@InternalApi +private[r2dbc] class H2DurableStateDao( + settings: StateSettings, + connectionFactory: ConnectionFactory, + closeCallsExceeding: Option[FiniteDuration] = None +)(implicit ec: ExecutionContext, system: ActorSystem[?]) + extends DurableStateDao(settings, connectionFactory, closeCallsExceeding) { + // always app timestamp (db is same process) monotonic increasing + H2Utils.settingRequirements(settings) + + override lazy val transactionTimestampSql: String = "CURRENT_TIMESTAMP" Review Comment: On Windows, `CURRENT_TIMESTAMP` here can be equal to or later than the `currentDbTimestamp()` taken after the write completes. H2 rounds the same coarse JVM clock to microseconds, while `InstantFactory` truncates it. The latest state then falls outside `db_timestamp < until`. See point 1 in the review; [889b196](https://github.com/ptrdom/pekko-persistence-r2dbc/commit/889b19639e5e3751c05ee82ed5da3e192b7e4aeb) writes it from a strictly increasing app clock instead. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
