apache / pekko-persistence-jdbc Goto Github PK

Asynchronously writes journal and snapshot entries to configured JDBC databases so that Apache Pekko Actors can recover state

Home Page: https://pekko.apache.org/

License: Apache License 2.0

Scala 95.92% PLSQL 1.02% Java 1.75% Shell 1.30%

journal pekko pekko-persistence persistence-query scala slick

pekko-persistence-jdbc's People

Stargazers

Watchers

Forkers

pjfanning mdedetrich seetaramayya zackman0010 lefou glngn jxnu-liguobin laglangyue wjohnsonbynder magnho samueleresca roiocam rohankumardubey he-pin

pekko-persistence-jdbc's Issues

Upgrade Slick to 3.4.1

Please release a version of pekko-persistence-jdbc with the Slick dependency updated to v3.4.1.
Rationale: the last few Play versions are on this version, we are using it and get a version conflict with the version used in pekko.

I am aware of the discussion regarding Slick 3.5, but the last messages I saw were from August 2023. Now it is December 2023, Slick 3.5 is still not released (AFAIK waiting for 2 packages in the ecosystem). So while waiting for Slick 3.5 w. Scala 3, which has an unknown timeline, a release with 3.4.1 would enable everyone to use the last Slick version with the last Play version without any version conflicts.

I tried updating it locally and it seems to work without issue. I'll gladly open a PR if this proposal is acceptable to you.

Update Slick to `3.4.x`

While updating dependencies on a project of mine I got an error from sbt about issues with mixing 3.3.x and 3.4.x versions of Slick. I assume it is because of potential binary incompatibility. Would it be beneficial to update Slick to 3.4.x for this project?

investigate oracle jdbc jar upgrade

Upgrading to using oracle jdbc8 instead of jdbc6 runs into trouble.

https://github.com/apache/incubator-pekko-persistence-jdbc/actions/runs/7862784116/job/21452649793

We might need to upgrade the oracle db image.
We might need to change our code
We are testing with an old jar and we should try to use a newer one

[info]   Cause: java.sql.SQLException: ORA-00604: error occurred at recursive SQL level 1
[info] ORA-01882: timezone region not found
[info]   at oracle.jdbc.driver.T4CTTIoer11.processError(T4CTTIoer11.java:494)
[info]   at oracle.jdbc.driver.T4CTTIoer11.processError(T4CTTIoer11.java:441)
[info]   at oracle.jdbc.driver.T4CTTIoer11.processError(T4CTTIoer11.java:436)
[info]   at oracle.jdbc.driver.T4CTTIfun.processError(T4CTTIfun.java:1110)
[info]   at oracle.jdbc.driver.T4CTTIoauthenticate.processError(T4CTTIoauthenticate.java:552)
[info]   at oracle.jdbc.driver.T4CTTIfun.receive(T4CTTIfun.java:540)
[info]   at oracle.jdbc.driver.T4CTTIfun.doRPC(T4CTTIfun.java:256)
[info]   at oracle.jdbc.driver.T4CTTIoauthenticate.doOAUTH(T4CTTIoauthenticate.java:501)
[info]   at oracle.jdbc.driver.T4CTTIoauthenticate.doOAUTH(T4CTTIoauthenticate.java:1297)
[info]   at oracle.jdbc.driver.T4CConnection.logon(T4CConnection.java:694)
[info]   at oracle.jdbc.driver.PhysicalConnection.connect(PhysicalConnection.java:782)
[info]   at oracle.jdbc.driver.T4CDriverExtension.getConnection(T4CDriverExtension.java:39)
[info]   at oracle.jdbc.driver.OracleDriver.connect(OracleDriver.java:704)
[info]   at com.zaxxer.hikari.util.DriverDataSource.getConnection(DriverDataSource.java:136)
[info]   at com.zaxxer.hikari.pool.PoolBase.newConnection(PoolBase.java:369)
[info]   at com.zaxxer.hikari.pool.PoolBase.newPoolEntry(PoolBase.java:198)
[info]   at com.zaxxer.hikari.pool.HikariPool.createPoolEntry(HikariPool.java:467)
[info]   at com.zaxxer.hikari.pool.HikariPool.access$100(HikariPool.java:71)
[info]   at com.zaxxer.hikari.pool.HikariPool$PoolEntryCreator.call(HikariPool.java:706)
[info]   at com.zaxxer.hikari.pool.HikariPool$PoolEntryCreator.call(HikariPool.java:692)
[info]   at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)

Query performance improvement

Motivation

A well-known problem was that the very large page query was inefficient in RDBMS.

Usually, we divide the query into multiple small batches of queries to avoid it, as like as we do below:

pekko-persistence-jdbc/core/src/main/scala/org/apache/pekko/persistence/jdbc/journal/dao/BaseJournalDaoWithReadMessages.scala

Lines 47 to 68 in d04a145

    
             xs <- messages(persistenceId, from, toSequenceNr, batchSize).runWith(Sink.seq) 
        
           } yield { 
        
             val hasMoreEvents = xs.size == batchSize 
        
             // Events are ordered by sequence number, therefore the last one is the largest) 
        
             val lastSeqNrInBatch: Option[Long] = xs.lastOption match { 
        
               case Some(Success((repr, _))) => Some(repr.sequenceNr) 
        
               case Some(Failure(e))         => throw e // fail the returned Future 
        
               case None                     => None 
        
             } 
        
             val hasLastEvent = lastSeqNrInBatch.exists(_ >= toSequenceNr) 
        
             val nextControl: FlowControl = 
        
               if (hasLastEvent || from > toSequenceNr) Stop 
        
               else if (hasMoreEvents) Continue 
        
               else if (refreshInterval.isEmpty) Stop 
        
               else ContinueDelayed 
        
             val nextFrom: Long = lastSeqNrInBatch match { 
        
               // Continue querying from the last sequence number (the events are ordered) 
        
               case Some(lastSeqNr) => lastSeqNr + 1 
        
               case None            => from 
        
             } 
        
             Some(((nextFrom, nextControl), xs))

However, it is the same as inefficient querying small batches with a large offset in RDBMS. see:

There may be multiple reasons, One of the theories is that the RDBMS must first fetch all the indexes, and then sort them to find the actual location of the data.

Fortunately, we can shorten the index range by using query conditions instead of implementing pagination based on offset. The former is usually several times faster than the latter in terms of efficiency (depending on the size of the table; generally, a very large offset may cause the query to degrade into a full table scan).

pekko-persistence-jdbc/core/src/main/scala/org/apache/pekko/persistence/jdbc/journal/dao/JournalQueries.scala

Lines 105 to 108 in d04a145

    
           .filter(_.sequenceNumber >= fromSequenceNr) 
        
           .filter(_.sequenceNumber <= toSequenceNr) 
        
           .sortBy(_.sequenceNumber.asc) 
        
           .take(max)

Add Scala 3 support

Just pinging @pjfanning and @nvollmar , I am assigning this to myself so we don't duplicate work.

migrator-integration-test tests are not run

Not run in CI

add release notes

to docs (paradox docs -- see incubator-pekko repo for an example)

replace use of deprecated statefulMapConcat

https://github.com/search?q=repo%3Aapache%2Fincubator-pekko-persistence-jdbc%20statefulMapConcat&type=code

any interest in taking this on @mdedetrich, @kerr ?

mima checks are breaking CI builds

need to disable them

investigate mssql jdbc jar upgrade

Tried using latest mssql jar but got errors. We are testing with an old jar and we should really be able to test successfully with newer jars.

https://github.com/apache/incubator-pekko-persistence-jdbc/actions/runs/7862784119

[info]   Cause: com.microsoft.sqlserver.jdbc.SQLServerException: "encrypt" property is set to "true" and "trustServerCertificate" property is set to "false" but the driver could not establish a secure connection to SQL Server by using Secure Sockets Layer (SSL) encryption: Error: PKIX path building failed: sun.security.provider.certpath.SunCertPathBuilderException: unable to find valid certification path to requested target. ClientConnectionId:ac43e56d-18b4-4434-9dc8-2b1cb114e7fa
[info]   at com.microsoft.sqlserver.jdbc.SQLServerConnection.terminate(SQLServerConnection.java:4266)
[info]   at com.microsoft.sqlserver.jdbc.TDSChannel.enableSSL(IOBuffer.java:1965)
[info]   at com.microsoft.sqlserver.jdbc.SQLServerConnection.connectHelper(SQLServerConnection.java:3792)
[info]   at com.microsoft.sqlserver.jdbc.SQLServerConnection.login(SQLServerConnection.java:3348)
[info]   at com.microsoft.sqlserver.jdbc.SQLServerConnection.connectInternal(SQLServerConnection.java:3179)
[info]   at com.microsoft.sqlserver.jdbc.SQLServerConnection.connect(SQLServerConnection.java:1953)
[info]   at com.microsoft.sqlserver.jdbc.SQLServerDriver.connect(SQLServerDriver.java:1263)
[info]   at com.zaxxer.hikari.util.DriverDataSource.getConnection(DriverDataSource.java:136)
[info]   at com.zaxxer.hikari.pool.PoolBase.newConnection(PoolBase.java:369)
[info]   at com.zaxxer.hikari.pool.PoolBase.newPoolEntry(PoolBase.java:198)
[info]   at com.zaxxer.hikari.pool.HikariPool.createPoolEntry(HikariPool.java:467)
[info]   at com.zaxxer.hikari.pool.HikariPool.access$100(HikariPool.java:71)
[info]   at com.zaxxer.hikari.pool.HikariPool$PoolEntryCreator.call(HikariPool.java:706)
[info]   at com.zaxxer.hikari.pool.HikariPool$PoolEntryCreator.call(HikariPool.java:692)
[info]   at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
[info]   at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
[info]   at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
[info]   at java.base/java.lang.Thread.run(Thread.java:829)
[info]   Cause: javax.net.ssl.SSLHandshakeException: PKIX path building failed: sun.security.provider.certpath.SunCertPathBuilderException: unable to find valid certification path to requested target
[info]   at java.base/sun.security.ssl.Alert.createSSLException(Alert.java:131)
[info]   at java.base/sun.security.ssl.TransportContext.fatal(TransportContext.java:360)
[info]   at java.base/sun.security.ssl.TransportContext.fatal(TransportContext.java:303)
[info]   at java.base/sun.security.ssl.TransportContext.fatal(TransportContext.java:298)
[info]   at java.base/sun.security.ssl.CertificateMessage$T12CertificateConsumer.checkServerCerts(CertificateMessage.java:654)
[info]   at java.base/sun.security.ssl.CertificateMessage$T12CertificateConsumer.onCertificate(CertificateMessage.java:473)
[info]   at java.base/sun.security.ssl.CertificateMessage$T12CertificateConsumer.consume(CertificateMessage.java:369)
[info]   at java.base/sun.security.ssl.SSLHandshake.consume(SSLHandshake.java:392)
[info]   at java.base/sun.security.ssl.HandshakeContext.dispatch(HandshakeContext.java:443)
[info]   at java.base/sun.security.ssl.HandshakeContext.dispatch(HandshakeContext.java:421)
[info]   at java.base/sun.security.ssl.TransportContext.dispatch(TransportContext.java:183)
[info]   at java.base/sun.security.ssl.SSLTransport.decode(SSLTransport.java:172)
[info]   at java.base/sun.security.ssl.SSLSocketImpl.decode(SSLSocketImpl.java:1511)
[info]   at java.base/sun.security.ssl.SSLSocketImpl.readHandshakeRecord(SSLSocketImpl.java:1421)
[info]   at java.base/sun.security.ssl.SSLSocketImpl.startHandshake(SSLSocketImpl.java:456)
[info]   at java.base/sun.security.ssl.SSLSocketImpl.startHandshake(SSLSocketImpl.java:427)
[info]   at com.microsoft.sqlserver.jdbc.TDSChannel.enableSSL(IOBuffer.java:1854)
[info]   at com.microsoft.sqlserver.jdbc.SQLServerConnection.connectHelper(SQLServerConnection.java:3792)
[info]   at com.microsoft.sqlserver.jdbc.SQLServerConnection.login(SQLServerConnection.java:3348)
[info]   at com.microsoft.sqlserver.jdbc.SQLServerConnection.connectInternal(SQLServerConnection.java:3179)
[info]   at com.microsoft.sqlserver.jdbc.SQLServerConnection.connect(SQLServerConnection.java:1953)
[info]   at com.microsoft.sqlserver.jdbc.SQLServerDriver.connect(SQLServerDriver.java:1263)
[info]   at com.zaxxer.hikari.util.DriverDataSource.getConnection(DriverDataSource.java:136)
[info]   at com.zaxxer.hikari.pool.PoolBase.newConnection(PoolBase.java:369)
[info]   at com.zaxxer.hikari.pool.PoolBase.newPoolEntry(PoolBase.java:198)
[info]   at com.zaxxer.hikari.pool.HikariPool.createPoolEntry(HikariPool.java:467)
[info]   at com.zaxxer.hikari.pool.HikariPool.access$100(HikariPool.java:71)
[info]   at com.zaxxer.hikari.pool.HikariPool$PoolEntryCreator.call(HikariPool.java:706)
[info]   at com.zaxxer.hikari.pool.HikariPool$PoolEntryCreator.call(HikariPool.java:692)
[info]   at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
[info]   at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
[info]   at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
[info]   at java.base/java.lang.Thread.run(Thread.java:829)
[info]   Cause: sun.security.validator.ValidatorException: PKIX path building failed: sun.security.provider.certpath.SunCertPathBuilderException: unable to find valid certification path to requested target
[info]   at java.base/sun.security.validator.PKIXValidator.doBuild(PKIXValidator.java:439)
[info]   at java.base/sun.security.validator.PKIXValidator.engineValidate(PKIXValidator.java:306)
[info]   at java.base/sun.security.validator.Validator.validate(Validator.java:264)
[info]   at java.base/sun.security.ssl.X509TrustManagerImpl.validate(X509TrustManagerImpl.java:313)
[info]   at java.base/sun.security.ssl.X509TrustManagerImpl.checkTrusted(X509TrustManagerImpl.java:233)
[info]   at java.base/sun.security.ssl.X509TrustManagerImpl.checkServerTrusted(X509TrustManagerImpl.java:110)
[info]   at com.microsoft.sqlserver.jdbc.HostNameOverrideX509TrustManager.checkServerTrusted(SQLServerTrustManager.java:88)
[info]   at java.base/sun.security.ssl.AbstractTrustManagerWrapper.checkServerTrusted(SSLContextImpl.java:1492)
[info]   at java.base/sun.security.ssl.CertificateMessage$T12CertificateConsumer.checkServerCerts(CertificateMessage.java:638)
[info]   at java.base/sun.security.ssl.CertificateMessage$T12CertificateConsumer.onCertificate(CertificateMessage.java:473)
[info]   at java.base/sun.security.ssl.CertificateMessage$T12CertificateConsumer.consume(CertificateMessage.java:369)
[info]   at java.base/sun.security.ssl.SSLHandshake.consume(SSLHandshake.java:392)
[info]   at java.base/sun.security.ssl.HandshakeContext.dispatch(HandshakeContext.java:443)
[info]   at java.base/sun.security.ssl.HandshakeContext.dispatch(HandshakeContext.java:421)
[info]   at java.base/sun.security.ssl.TransportContext.dispatch(TransportContext.java:183)
[info]   at java.base/sun.security.ssl.SSLTransport.decode(SSLTransport.java:172)
[info]   at java.base/sun.security.ssl.SSLSocketImpl.decode(SSLSocketImpl.java:1511)
[info]   at java.base/sun.security.ssl.SSLSocketImpl.readHandshakeRecord(SSLSocketImpl.java:1421)
[info]   at java.base/sun.security.ssl.SSLSocketImpl.startHandshake(SSLSocketImpl.java:456)
[info]   at java.base/sun.security.ssl.SSLSocketImpl.startHandshake(SSLSocketImpl.java:427)
[info]   at com.microsoft.sqlserver.jdbc.TDSChannel.enableSSL(IOBuffer.java:1854)
[info]   at com.microsoft.sqlserver.jdbc.SQLServerConnection.connectHelper(SQLServerConnection.java:3792)
[info]   at com.microsoft.sqlserver.jdbc.SQLServerConnection.login(SQLServerConnection.java:3348)
[info]   at com.microsoft.sqlserver.jdbc.SQLServerConnection.connectInternal(SQLServerConnection.java:3179)
[info]   at com.microsoft.sqlserver.jdbc.SQLServerConnection.connect(SQLServerConnection.java:1953)
[info]   at com.microsoft.sqlserver.jdbc.SQLServerDriver.connect(SQLServerDriver.java:1263)
[info]   at com.zaxxer.hikari.util.DriverDataSource.getConnection(DriverDataSource.java:136)
[info]   at com.zaxxer.hikari.pool.PoolBase.newConnection(PoolBase.java:369)
[info]   at com.zaxxer.hikari.pool.PoolBase.newPoolEntry(PoolBase.java:198)
[info]   at com.zaxxer.hikari.pool.HikariPool.createPoolEntry(HikariPool.java:467)
[info]   at com.zaxxer.hikari.pool.HikariPool.access$100(HikariPool.java:71)
[info]   at com.zaxxer.hikari.pool.HikariPool$PoolEntryCreator.call(HikariPool.java:706)
[info]   at com.zaxxer.hikari.pool.HikariPool$PoolEntryCreator.call(HikariPool.java:692)
[info]   at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
[info]   at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
[info]   at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
[info]   at java.base/java.lang.Thread.run(Thread.java:829)
[info]   Cause: sun.security.provider.certpath.SunCertPathBuilderException: unable to find valid certification path to requested target
[info]   at java.base/sun.security.provider.certpath.SunCertPathBuilder.build(SunCertPathBuilder.java:148)
[info]   at java.base/sun.security.provider.certpath.SunCertPathBuilder.engineBuild(SunCertPathBuilder.java:129)
[info]   at java.base/java.security.cert.CertPathBuilder.build(CertPathBuilder.java:297)
[info]   at java.base/sun.security.validator.PKIXValidator.doBuild(PKIXValidator.java:434)
[info]   at java.base/sun.security.validator.PKIXValidator.engineValidate(PKIXValidator.java:306)
[info]   at java.base/sun.security.validator.Validator.validate(Validator.java:264)
[info]   at java.base/sun.security.ssl.X509TrustManagerImpl.validate(X509TrustManagerImpl.java:313)
[info]   at java.base/sun.security.ssl.X509TrustManagerImpl.checkTrusted(X509TrustManagerImpl.java:233)
[info]   at java.base/sun.security.ssl.X509TrustManagerImpl.checkServerTrusted(X509TrustManagerImpl.java:110)
[info]   at com.microsoft.sqlserver.jdbc.HostNameOverrideX509TrustManager.checkServerTrusted(SQLServerTrustManager.java:88)
[info]   at java.base/sun.security.ssl.AbstractTrustManagerWrapper.checkServerTrusted(SSLContextImpl.java:1492)
[info]   at java.base/sun.security.ssl.CertificateMessage$T12CertificateConsumer.checkServerCerts(CertificateMessage.java:638)
[info]   at java.base/sun.security.ssl.CertificateMessage$T12CertificateConsumer.onCertificate(CertificateMessage.java:473)
[info]   at java.base/sun.security.ssl.CertificateMessage$T12CertificateConsumer.consume(CertificateMessage.java:369)
[info]   at java.base/sun.security.ssl.SSLHandshake.consume(SSLHandshake.java:392)
[info]   at java.base/sun.security.ssl.HandshakeContext.dispatch(HandshakeContext.java:443)
[info]   at java.base/sun.security.ssl.HandshakeContext.dispatch(HandshakeContext.java:421)
[info]   at java.base/sun.security.ssl.TransportContext.dispatch(TransportContext.java:183)
[info]   at java.base/sun.security.ssl.SSLTransport.decode(SSLTransport.java:172)
[info]   at java.base/sun.security.ssl.SSLSocketImpl.decode(SSLSocketImpl.java:1511)
[info]   at java.base/sun.security.ssl.SSLSocketImpl.readHandshakeRecord(SSLSocketImpl.java:[1421](https://github.com/apache/incubator-pekko-persistence-jdbc/actions/runs/7862784119/job/21452649542#step:7:1422))
[info]   at java.base/sun.security.ssl.SSLSocketImpl.startHandshake(SSLSocketImpl.java:456)
[info]   at java.base/sun.security.ssl.SSLSocketImpl.startHandshake(SSLSocketImpl.java:427)
[info]   at com.microsoft.sqlserver.jdbc.TDSChannel.enableSSL(IOBuffer.java:1854)
[info]   at com.microsoft.sqlserver.jdbc.SQLServerConnection.connectHelper(SQLServerConnection.java:3792)
[info]   at com.microsoft.sqlserver.jdbc.SQLServerConnection.login(SQLServerConnection.java:3348)
[info]   at com.microsoft.sqlserver.jdbc.SQLServerConnection.connectInternal(SQLServerConnection.java:3179)
[info]   at com.microsoft.sqlserver.jdbc.SQLServerConnection.connect(SQLServerConnection.java:1953)
[info]   at com.microsoft.sqlserver.jdbc.SQLServerDriver.connect(SQLServerDriver.java:1263)
[info]   at com.zaxxer.hikari.util.DriverDataSource.getConnection(DriverDataSource.java:136)
[info]   at com.zaxxer.hikari.pool.PoolBase.newConnection(PoolBase.java:369)
[info]   at com.zaxxer.hikari.pool.PoolBase.newPoolEntry(PoolBase.java:198)
[info]   at com.zaxxer.hikari.pool.HikariPool.createPoolEntry(HikariPool.java:467)
[info]   at com.zaxxer.hikari.pool.HikariPool.access$100(HikariPool.java:71)
[info]   at com.zaxxer.hikari.pool.HikariPool$PoolEntryCreator.call(HikariPool.java:706)
[info]   at com.zaxxer.hikari.pool.HikariPool$PoolEntryCreator.call(HikariPool.java:692)
[info]   at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
[info]   at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
[info]   at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
[info]   at java.base/java.lang.Thread.run(Thread.java:829)

Move away from using sbt integration config

See https://github.com/sbt/sbt/releases/tag/v1.9.0-RC3, specifically

Deprecation of IntegrationTest configuration
sbt 1.9.0 deprecates IntegrationTest configuration. (RFC-3 proposes to deprecate general use of configuration axis beyond Compile and Test, and this is the first installment of the change.)

The recommended migration path is to create a subproject named "integration", or "foo-integration" etc.

lazy val integration = (project in file("integration"))
  .dependsOn(core) // your current subproject
  .settings(
    publish / skip := true,
    // test dependencies
    libraryDependencies += something % Test,
  )

Slow JDBC query for eventsByTag

The eventsByTag query is very slow because of a subquery and the incorrect placement of the where clause.

Dumped from our logs

select x2.x3, x2.x4, x2.x5, x2.x6, x2.x7, x2.x8, x2.x9, x2.x10, x2.x11, x2.x12, x2.x13, x2.x14, x2.x15 from (select "deleted" as x4, "meta_ser_id" as x14, "meta_ser_manifest" as x15, "sequence_number" as x6, "writer" as x7, "event_ser_manifest" as x12, "write_timestamp" as x8, "persistence_id" as x5, "adapter_manifest" as x9, "ordering" as x3, "event_payload" as x10, "event_ser_id" as x11, "meta_payload" as x13 from "event_journal" where "deleted" = false) x2, "event_tag" x16 where ((x16."tag" = ?) and ((x2.x3 > ?) and (x2.x3 <= ?))) and (x2.x3 = x16."event_id") order by x2.x3 limit ?

This subquery
select "deleted" as x4, "meta_ser_id" as x14, "meta_ser_manifest" as x15, "sequence_number" as x6, "writer" as x7, "event_ser_manifest" as x12, "write_timestamp" as x8, "persistence_id" as x5, "adapter_manifest" as x9, "ordering" as x3, "event_payload" as x10, "event_ser_id" as x11, "meta_payload" as x13 from "event_journal" where "deleted" = false

Doesn't have the where clause for the Ordering field. It will run the entire event-journal each time.

Initial PR that seems to have created the issue. akka/akka-persistence-jdbc#467

The query in question
https://github.com/apache/incubator-pekko-persistence-jdbc/blob/a9ebfe6da617f8cd2abc5813d2a530b1e1de70bb/core/src/main/scala/org/apache/pekko/persistence/jdbc/query/dao/ReadJournalQueries.scala#L61

The issue is the Join happens before the filters on the event_journal table. This pushes the where clause to the main query forcing a full event journal query in the subquery.

private def baseTableWithTagsQuery() = {
baseTableQuery().join(TagTable).on(_.ordering === _.eventId)
}

private def eventsByTag(
tag: Rep[String],
offset: ConstColumn[Long],
maxOffset: ConstColumn[Long],
max: ConstColumn[Long]) = {
baseTableWithTagsQuery()
.filter(.2.tag === tag)
.sortBy(._1.ordering.asc)
.filter(row => row._1.ordering > offset && row.1.ordering <= maxOffset)
.take(max)
.map(._1)
}

I think the fix is as simple as reordering the query to push the filter above the join.

private def eventsByTag(
tag: Rep[String],
offset: ConstColumn[Long],
maxOffset: ConstColumn[Long],
max: ConstColumn[Long]
) = {
baseTableQuery()
.filter(row => row.ordering > offset && row.ordering <= maxOffset)
.sortBy(.ordering.asc)
.join(TagTable)
.on(_.ordering === .eventId)
.filter(.2.tag === tag)
.take(max)
.map(._1)
}

scala3 build issue

https://github.com/apache/incubator-pekko-persistence-jdbc/actions/runs/8190639534/job/22398044719

Seems like we need to fix the skip-packages setting - changed in Scala 3.

fyi @mdedetrich

Make DurableState possible to use with other databases

I know the first priority is getting the 1.0.0 version out, so this won't be handled for a while, but I figured I'd go ahead and submit this issue so it might could be looked at in the future.

One of the things currently unimplemented even in the original akka-persistence-jdbc is Durable State for any database other than Postgres or H2. This limitation is not even mentioned in the documentation, it just throws errors at runtime if you try to use a different database (Oracle in our case) for Durable State. I'd love to see it finally get fully implemented in Pekko.

come up with new release strategy

will be based on what we end up doing in incubator-pekko
will need to remove the release train docs
and remove/replace RELEASING.md
possibly replace https://github.com/akka/akka-persistence-jdbc/blob/master/.github/release-drafter.yml

acknowledge Dennis Vriend contributions

The Akka repository on which this Pekko repository is based was itself derived from a separate project created by Dennis Vriend.

Everything is Apache 2.0 licensed but the question is whether we need to add more details to our LICENSE.

We acknowledge this copyright in https://github.com/apache/incubator-pekko-persistence-jdbc/blob/main/NOTICE

We have retained all the source headers from the Akka repository including those that contain a Dennis Vriend copyright statement.

@justinmclean What do you think? Do we need to list all the files with a Dennis Vriend copyright header in our LICENSE or is the NOTICE attribution enough?

CI lacks environmental configuration sometimes.

./scripts/launch-postgres.sh: line 34: docker-compose: command not found
./scripts/launch-postgres.sh: line 35: docker-compose: command not found
./scripts/launch-postgres.sh: line 36: docker-compose: command not found
Postgres not available, retrying...
Postgres not available, retrying...
Postgres not available, retrying...
Postgres not available, retrying...
Postgres not available, retrying...
Postgres not available, retrying...

See:

add apache license headers

Implement `deleteObject(persistenceId: String, revision: Long)` properly

As can be seen from https://github.com/apache/incubator-pekko-persistence-jdbc/blob/db5733e60fffb99d3d822d3d60199ec2f0890278/core/src/main/scala/org/apache/pekko/persistence/jdbc/state/scaladsl/JdbcDurableStateStore.scala#L117-L118 the current implementation of deleteObject(persistenceId: String, revision: Long) just defers to deleteObject(persistenceId: String) (which means it ignores the revision).

The implementation should instead be improved so it only deletes correct revision.

publish 1.0.0 docs

see incubator-pekko repo for an example GitHub CI job

Journal does logical and physical deletion in same transaction

DefaultJournalDao.delete(persistenceId: String, maxSequenceNr: Long) does both logical and physical deletion in the same transaction. We did some load-testing recently (with PostgreSQL) and found that these operations both used a lot of resources.

There's probably a case for doing logical deletion in terms of performance, but the way it's implemented now, the logical deletion doesn't make sense, and should probably be removed.

DefaultJournalDao.delete

MySQL support for DurableState

Motivation

supplement #158, continue solves #56

In #158, we notice that other databases support sequence but MySQL does not, but MySQL does have auto-increment coumn.

I used SELECT AUTO_INCREMENT FROM information_schema.TABLES for retrieving recent automatically generated AUTO_INCREMENT value, but the bigger problem is the sequence number only assigned from MySQL, which can not increase them via SequenceNextValUpdater.

use approach used in incubator-pekko - new @mdedetrich sbt plugin
may choose to publish nightly (CI action with cron) or after every successful build on main branch

Migrator Integration tests are failing for multiple db servers

see https://github.com/apache/pekko-persistence-jdbc/actions/runs/9247671484/job/25475898553, https://github.com/apache/pekko-persistence-jdbc/actions/runs/9247671480/job/25445435838, https://github.com/apache/pekko-persistence-jdbc/actions/runs/9247671477/job/25445328062

(MySQL, Oracle, SQL Server)

@Roiocam is this a release blocker or can we just disable the broken tests?

need to review if we can remove ojdbc6 jar

see https://github.com/apache/incubator-pekko-persistence-jdbc/tree/main/core/lib
we should not have jars checked in
we can get the build to script to download the jar from an external web site, if necessary

	xs <- messages(persistenceId, from, toSequenceNr, batchSize).runWith(Sink.seq)
	} yield {
	val hasMoreEvents = xs.size == batchSize
	// Events are ordered by sequence number, therefore the last one is the largest)
	val lastSeqNrInBatch: Option[Long] = xs.lastOption match {
	case Some(Success((repr, _))) => Some(repr.sequenceNr)
	case Some(Failure(e)) => throw e // fail the returned Future
	case None => None
	}
	val hasLastEvent = lastSeqNrInBatch.exists(_ >= toSequenceNr)
	val nextControl: FlowControl =
	if (hasLastEvent \|\| from > toSequenceNr) Stop
	else if (hasMoreEvents) Continue
	else if (refreshInterval.isEmpty) Stop
	else ContinueDelayed

	val nextFrom: Long = lastSeqNrInBatch match {
	// Continue querying from the last sequence number (the events are ordered)
	case Some(lastSeqNr) => lastSeqNr + 1
	case None => from
	}
	Some(((nextFrom, nextControl), xs))

	.filter(_.sequenceNumber >= fromSequenceNr)
	.filter(_.sequenceNumber <= toSequenceNr)
	.sortBy(_.sequenceNumber.asc)
	.take(max)

apache / pekko-persistence-jdbc Goto Github PK

pekko-persistence-jdbc's People

Stargazers

Watchers

Forkers

pekko-persistence-jdbc's Issues

Motivation

Motivation

Recommend Projects

Recommend Topics

Recommend Org