ptrdom commented on code in PR #532: URL: https://github.com/apache/pekko-persistence-r2dbc/pull/532#discussion_r4491094466
########## 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: Added `OracleUtils.requireByteArrayPayload`. The Oracle journal, query, snapshot and durable state DAOs call it with their own codec, so a JSON `payload-column-type` fails when the plugin starts. Unit test in `OracleUtilsSpec`; the docs say so too. ########## .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: Updated to `sbt/setup-sbt` v1.5.13 and `coursier/cache-action` v9.0.1, matching the other jobs. ########## 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: Done. `QueryDao.sliceCondition` defaults to `slice BETWEEN`, like `SnapshotDao`, and `OracleQueryDao` no longer copies `eventsBySlicesRangeSql`. I left `OracleDurableStateDao.stateBySlicesRangeSql` as it is, because of its timestamp expressions. ########## project/Dependencies.scala: ########## @@ -83,6 +87,7 @@ object Dependencies { r2dbcMariadb % "provided,test", r2dbcH2 % "provided,test", h2 % "provided,test", + r2dbcOracle % Test, Review Comment: Changed to `provided,test`, like the other drivers. -- 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]
