pjfanning commented on code in PR #532: URL: https://github.com/apache/pekko-persistence-r2dbc/pull/532#discussion_r4480571577
########## core/src/main/scala/org/apache/pekko/persistence/r2dbc/query/scaladsl/oracle/OracleQueryDao.scala: ########## @@ -0,0 +1,92 @@ +/* + * 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.persistence.r2dbc.query.scaladsl.oracle + +import java.time.Instant + +import scala.concurrent.ExecutionContext +import scala.concurrent.Future +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.oracle.OracleUtils +import pekko.persistence.r2dbc.journal.mysql.MySQLJournalDao +import pekko.persistence.r2dbc.query.scaladsl.QueryDao +import io.r2dbc.spi.ConnectionFactory +import io.r2dbc.spi.Row + +/** + * INTERNAL API + */ +@InternalApi +private[r2dbc] class OracleQueryDao( + querySettings: QuerySettings, + connectionFactory: ConnectionFactory, + closeCallsExceeding: Option[FiniteDuration] = None +)(implicit ec: ExecutionContext, system: ActorSystem[?]) + extends QueryDao(querySettings, connectionFactory, closeCallsExceeding) { + + override lazy val statementTimestampSql: String = OracleUtils.currentTimestampSql + + override protected def tagsFromRow(row: Row): Set[String] = + MySQLJournalDao.tagsFromJson(row.get("tags", classOf[String])) + + 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, event_ser_manifest " + 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 = + OracleUtils.behindCurrentTimeIntervalCondition(behindCurrentTime) + + override def eventsBySlicesRangeSql( + toDbTimestampParam: Boolean, + behindCurrentTime: FiniteDuration, + backtracking: Boolean, + minSlice: Int, + maxSlice: Int): String = + sql""" + ${selectColumns(backtracking)} + FROM $journalTable + WHERE entity_type = ? Review Comment: **Empty entity types return no rows.** Oracle stores `''` as `NULL`, and the DDL makes `entity_type` nullable for that reason, but `entity_type = ?` bound to `''` never matches. A persistence id with no separator, such as `pid-1`, has an empty entity type, so `eventsBySlices("", ...)` returns no rows and no error, whereas other dialects return the events. The same applies to the bucket query here, the `OracleDurableStateDao` queries and the snapshot slice queries. These need to handle the empty case, for example with `entity_type IS NULL`, or reject it clearly if it isn't meant to be supported. ########## core/src/main/scala/org/apache/pekko/persistence/r2dbc/internal/oracle/OracleConnections.scala: ########## @@ -0,0 +1,77 @@ +/* + * 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.persistence.r2dbc.internal.oracle + +import java.nio.ByteBuffer +import java.util.Locale + +import org.apache.pekko +import pekko.annotation.InternalApi +import pekko.persistence.r2dbc.internal.ConnectionAdapter +import io.r2dbc.spi.Blob +import io.r2dbc.spi.ConnectionFactoryOptions +import io.r2dbc.spi.Row +import reactor.core.publisher.Mono + +/** + * INTERNAL API + * + * Adapts the Oracle R2DBC driver to the types used by the plugin: + * + * - `java.time.Instant` is not supported by the driver, so it is bound and read as a UTC `LocalDateTime`. The + * `TIMESTAMP` columns of the Oracle DDL have no time zone, so the session time zone doesn't affect them. + * - `Array[Byte]` is bound as a `Blob`, because the driver binds byte arrays as size limited `RAW` values in the + * `SELECT` of a `MERGE`. + * - Oracle stores empty strings and empty byte arrays as `NULL`, so the entity type, serializer manifest, writer and + * payload columns of the plugin tables are read back as empty values instead of `null`. + */ +@InternalApi +private[r2dbc] object OracleConnections extends ConnectionAdapter.InstantAsUtcLocalDateTime { + + def isOracle(options: ConnectionFactoryOptions): Boolean = { + def is(option: io.r2dbc.spi.Option[?]): Boolean = + Option(options.getValue(option)).exists(_.toString.toLowerCase(Locale.ROOT).startsWith("oracle")) + is(ConnectionFactoryOptions.DRIVER) || is(ConnectionFactoryOptions.PROTOCOL) + } + + private val EmptyStringColumns = + Set("entity_type", "event_ser_manifest", "ser_manifest", "state_ser_manifest", "meta_ser_manifest", "writer") + + private val EmptyPayloadColumns = Set("event_payload", "snapshot", "state_payload") + + override def bindValue(value: Any): Any = + value match { + case bytes: Array[Byte] => Blob.from(Mono.just(ByteBuffer.wrap(bytes))) + case other => super.bindValue(other) + } + + override def bindNullType(tpe: Class[?]): Class[?] = + if (tpe == classOf[Array[Byte]]) classOf[Blob] + else super.bindNullType(tpe) + + override def get[T](row: Row, name: String, tpe: Class[T]): T = { Review Comment: **This also changes results for users' own tables.** The mapping goes by column name only, across every result set on the connection, including queries through `R2dbcSession`. `SELECT writer FROM my_table` returns `""` by name where the column is really `NULL`, but `row.get(0, classOf[String])` still returns `null`, so the result depends on how the column is read. Applying the defaults in the plugin's DAO row mappers would avoid both problems. ########## .github/workflows/build-test.yml: ########## @@ -375,6 +375,61 @@ jobs: - name: test run: sbt -Dpekko.persistence.r2dbc.dialect=mariadb ${{ matrix.SCALA_VERSION }} ${{ matrix.COMPILE_ONLY && 'Test/compile' || 'test' }} + test-oracle: + name: Run tests with Oracle + runs-on: ubuntu-latest + strategy: + matrix: + SCALA_VERSION: [ scala213, scala3 ] + JAVA_VERSION: [ 17, 21 ] + # integration tests only run for the published Scala versions + if: github.repository == 'apache/pekko-persistence-r2dbc' && github.event_name != 'schedule' + 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: Setup Java ${{ matrix.JAVA_VERSION }} + uses: actions/setup-java@b843ebf9b913146b38da2eaa7aeddf76074ae1ad # v6.1.0 + with: + distribution: temurin + java-version: ${{ matrix.JAVA_VERSION }} + + - name: Install sbt + uses: sbt/setup-sbt@ecad189dd245444b9ae0a0cd5922fd938b88906f # v1.5.12 Review Comment: This job pins `sbt/setup-sbt` v1.5.12 and, a few lines below, `coursier/cache-action` v8.2.0. The other jobs use v1.5.13 and v9.0.1 after #540 and #541, so these should be updated to match. ########## core/src/main/scala/org/apache/pekko/persistence/r2dbc/internal/oracle/OracleUtils.scala: ########## @@ -0,0 +1,59 @@ +/* + * 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.persistence.r2dbc.internal.oracle + +import scala.concurrent.duration.Duration +import scala.concurrent.duration.FiniteDuration + +import org.apache.pekko +import pekko.annotation.InternalApi +import pekko.persistence.r2dbc.internal.SettingsTypes + +/** + * INTERNAL API + */ +@InternalApi +private[r2dbc] object OracleUtils { + + /** + * The current time in UTC, matching the UTC `TIMESTAMP` columns of the Oracle DDL. + */ + val currentTimestampSql: String = "SYS_EXTRACT_UTC(SYSTIMESTAMP)" + + /** + * The 10 second bucket of the `db_timestamp` column, as seconds since epoch divided by 10. The whole minutes are + * counted with date arithmetic, which is rounded because a minute isn't exact in fractions of a day, and the seconds + * are taken from the timestamp, so that fractional seconds don't move a timestamp into the next bucket. + */ + val bucketSql: String = + "ROUND((TRUNC(db_timestamp, 'MI') - DATE '1970-01-01') * 8640) + FLOOR(EXTRACT(SECOND FROM db_timestamp) / 10)" + + def behindCurrentTimeIntervalCondition(behindCurrentTime: FiniteDuration): String = + if (behindCurrentTime > Duration.Zero) + s"AND db_timestamp < $currentTimestampSql - NUMTODSINTERVAL(${behindCurrentTime.toMicros} / 1000000, 'SECOND')" + else "" + + def settingRequirements(settings: SettingsTypes.AppTimestampSettings): Unit = { Review Comment: **JSON payload types aren't rejected.** The docs say `JSONB` isn't supported on Oracle, but with `payload-column-type = JSONB` startup succeeds and the first write then binds an `io.r2dbc.spi.Json` into a BLOB column and fails with a driver error. A `require` could fail fast with a clear message, though the journal, state and snapshot DAOs would each need the check for their own codec. ########## ddl-scripts/create_tables_oracle.sql: ########## @@ -0,0 +1,76 @@ +-- 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. + +-- Requires Oracle Database 23ai or later. +-- Oracle stores empty strings and empty byte arrays as NULL, so the entity type, serializer manifest, writer and +-- payload columns are nullable. + +CREATE TABLE IF NOT EXISTS event_journal( + slice NUMBER(10) NOT NULL, + entity_type VARCHAR2(255), Review Comment: **`VARCHAR2(255)` is 255 bytes** on a database with the default `NLS_LENGTH_SEMANTICS=BYTE`, so these columns hold fewer multibyte characters than the 255 that Postgres and MySQL allow. A 100-character persistence id of 3-byte UTF-8 characters needs 300 bytes and fails with ORA-12899. `VARCHAR2(255 CHAR)` fixes it, across the identifier and manifest columns in all three tables. It's easier to change before this DDL ships. ########## core/src/main/scala/org/apache/pekko/persistence/r2dbc/snapshot/oracle/OracleSnapshotDao.scala: ########## @@ -0,0 +1,113 @@ +/* + * 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.persistence.r2dbc.snapshot.oracle + +import java.time.Instant + +import scala.concurrent.ExecutionContext +import scala.concurrent.Future +import scala.concurrent.duration.FiniteDuration + +import org.apache.pekko +import pekko.actor.typed.ActorSystem +import pekko.annotation.InternalApi +import pekko.persistence.r2dbc.SnapshotSettings +import pekko.persistence.r2dbc.internal.InstantFactory +import pekko.persistence.r2dbc.internal.Sql.DialectInterpolation +import pekko.persistence.r2dbc.internal.oracle.OracleUtils +import pekko.persistence.r2dbc.journal.mysql.MySQLJournalDao +import pekko.persistence.r2dbc.snapshot.SnapshotDao +import io.r2dbc.spi.ConnectionFactory +import io.r2dbc.spi.Row +import io.r2dbc.spi.Statement + +/** + * INTERNAL API + */ +@InternalApi +private[r2dbc] class OracleSnapshotDao( + settings: SnapshotSettings, connectionFactory: ConnectionFactory, + closeCallsExceeding: Option[FiniteDuration] = None +)(implicit ec: ExecutionContext, system: ActorSystem[?]) + extends SnapshotDao(settings, connectionFactory, closeCallsExceeding) { + + override protected def createUpsertSql: String = + if (settings.startFromSnapshotEnabled) + sql""" + MERGE INTO $snapshotTable t Review Comment: **Concurrent first saves can fail.** `MERGE` isn't atomic against concurrent inserts, unlike `ON CONFLICT` and `ON DUPLICATE KEY UPDATE` in the other dialects. Two saves for the same persistence id with no existing row, such as rapid `saveSnapshot` calls or an old and new incarnation during a rebalance, can both take `WHEN NOT MATCHED`. The loser fails with ORA-00001 and surfaces as `SaveSnapshotFailure`, whereas other dialects let the last write win. Consider retrying the `MERGE` on a unique constraint violation. ########## core/src/main/scala/org/apache/pekko/persistence/r2dbc/internal/oracle/OracleConnections.scala: ########## @@ -0,0 +1,77 @@ +/* + * 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.persistence.r2dbc.internal.oracle + +import java.nio.ByteBuffer +import java.util.Locale + +import org.apache.pekko +import pekko.annotation.InternalApi +import pekko.persistence.r2dbc.internal.ConnectionAdapter +import io.r2dbc.spi.Blob +import io.r2dbc.spi.ConnectionFactoryOptions +import io.r2dbc.spi.Row +import reactor.core.publisher.Mono + +/** + * INTERNAL API + * + * Adapts the Oracle R2DBC driver to the types used by the plugin: + * + * - `java.time.Instant` is not supported by the driver, so it is bound and read as a UTC `LocalDateTime`. The + * `TIMESTAMP` columns of the Oracle DDL have no time zone, so the session time zone doesn't affect them. + * - `Array[Byte]` is bound as a `Blob`, because the driver binds byte arrays as size limited `RAW` values in the + * `SELECT` of a `MERGE`. + * - Oracle stores empty strings and empty byte arrays as `NULL`, so the entity type, serializer manifest, writer and + * payload columns of the plugin tables are read back as empty values instead of `null`. + */ +@InternalApi +private[r2dbc] object OracleConnections extends ConnectionAdapter.InstantAsUtcLocalDateTime { + + def isOracle(options: ConnectionFactoryOptions): Boolean = { + def is(option: io.r2dbc.spi.Option[?]): Boolean = + Option(options.getValue(option)).exists(_.toString.toLowerCase(Locale.ROOT).startsWith("oracle")) + is(ConnectionFactoryOptions.DRIVER) || is(ConnectionFactoryOptions.PROTOCOL) + } + + private val EmptyStringColumns = + Set("entity_type", "event_ser_manifest", "ser_manifest", "state_ser_manifest", "meta_ser_manifest", "writer") + + private val EmptyPayloadColumns = Set("event_payload", "snapshot", "state_payload") + + override def bindValue(value: Any): Any = + value match { + case bytes: Array[Byte] => Blob.from(Mono.just(ByteBuffer.wrap(bytes))) Review Comment: **This converts every byte array bind, but ORA-01461 only affects the snapshot `MERGE`.** A user `R2dbcSession` statement, change handler or `AdditionalColumn` that binds bytes to a `RAW` column, or compares `raw_col = ?`, now sends a BLOB, which Oracle should reject with ORA-00932. Journal and durable state writes also pay for temporary LOB handling, even for small payloads. Consider limiting this to the snapshot upsert, for example by binding the `Blob` from `OracleSnapshotDao`. ########## core/src/main/scala/org/apache/pekko/persistence/r2dbc/internal/oracle/OracleConnections.scala: ########## @@ -0,0 +1,77 @@ +/* + * 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.persistence.r2dbc.internal.oracle + +import java.nio.ByteBuffer +import java.util.Locale + +import org.apache.pekko +import pekko.annotation.InternalApi +import pekko.persistence.r2dbc.internal.ConnectionAdapter +import io.r2dbc.spi.Blob +import io.r2dbc.spi.ConnectionFactoryOptions +import io.r2dbc.spi.Row +import reactor.core.publisher.Mono + +/** + * INTERNAL API + * + * Adapts the Oracle R2DBC driver to the types used by the plugin: + * + * - `java.time.Instant` is not supported by the driver, so it is bound and read as a UTC `LocalDateTime`. The + * `TIMESTAMP` columns of the Oracle DDL have no time zone, so the session time zone doesn't affect them. + * - `Array[Byte]` is bound as a `Blob`, because the driver binds byte arrays as size limited `RAW` values in the + * `SELECT` of a `MERGE`. + * - Oracle stores empty strings and empty byte arrays as `NULL`, so the entity type, serializer manifest, writer and + * payload columns of the plugin tables are read back as empty values instead of `null`. + */ +@InternalApi +private[r2dbc] object OracleConnections extends ConnectionAdapter.InstantAsUtcLocalDateTime { + + def isOracle(options: ConnectionFactoryOptions): Boolean = { + def is(option: io.r2dbc.spi.Option[?]): Boolean = + Option(options.getValue(option)).exists(_.toString.toLowerCase(Locale.ROOT).startsWith("oracle")) + is(ConnectionFactoryOptions.DRIVER) || is(ConnectionFactoryOptions.PROTOCOL) + } + + private val EmptyStringColumns = + Set("entity_type", "event_ser_manifest", "ser_manifest", "state_ser_manifest", "meta_ser_manifest", "writer") + + private val EmptyPayloadColumns = Set("event_payload", "snapshot", "state_payload") Review Comment: **`meta_payload` is missing here.** Metadata whose serializer produces zero bytes is written with `meta_ser_id` set, and the payload then comes back as `NULL`: - `JournalDao.readMetadata` treats `null` as no metadata, so the event's metadata is silently dropped. - `SnapshotDao` builds `SerializedSnapshotMetadata(null, ...)` because `meta_ser_id` is non-null, which then fails on deserialization. `OracleDialectSpec` only covers an empty `meta_ser_manifest` with a non-empty payload, so this needs a round-trip test too. ########## core/src/main/scala/org/apache/pekko/persistence/r2dbc/query/scaladsl/oracle/OracleQueryDao.scala: ########## @@ -0,0 +1,92 @@ +/* + * 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.persistence.r2dbc.query.scaladsl.oracle + +import java.time.Instant + +import scala.concurrent.ExecutionContext +import scala.concurrent.Future +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.oracle.OracleUtils +import pekko.persistence.r2dbc.journal.mysql.MySQLJournalDao +import pekko.persistence.r2dbc.query.scaladsl.QueryDao +import io.r2dbc.spi.ConnectionFactory +import io.r2dbc.spi.Row + +/** + * INTERNAL API + */ +@InternalApi +private[r2dbc] class OracleQueryDao( + querySettings: QuerySettings, + connectionFactory: ConnectionFactory, + closeCallsExceeding: Option[FiniteDuration] = None +)(implicit ec: ExecutionContext, system: ActorSystem[?]) + extends QueryDao(querySettings, connectionFactory, closeCallsExceeding) { + + override lazy val statementTimestampSql: String = OracleUtils.currentTimestampSql + + override protected def tagsFromRow(row: Row): Set[String] = + MySQLJournalDao.tagsFromJson(row.get("tags", classOf[String])) + + 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, event_ser_manifest " + 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 = + OracleUtils.behindCurrentTimeIntervalCondition(behindCurrentTime) + + override def eventsBySlicesRangeSql( Review Comment: Nit: this copy matches the base query apart from the slice predicate, and only exists because `sliceCondition` in `QueryDao` throws for other dialects. Defaulting it to `slice BETWEEN`, as `SnapshotDao.sliceCondition` already does, would remove it. `OracleDurableStateDao.stateBySlicesRangeSql` has the same shape, but it would also need its timestamp expressions made overridable. -- 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]
