Custom DAO Implementation

The plugin supports loading a custom DAO for the journal and snapshot. You should implement a custom Data Access Object (DAO) if you wish to alter the default persistence strategy in any way, but wish to reuse all the logic that the plugin already has in place, eg. the Apache Pekko Persistence Query API. For example, the default persistence strategy that the plugin supports serializes journal and snapshot messages using a serializer of your choice and stores them as byte arrays in the database.

By means of configuration in application.conf a DAO can be configured, below the default DAOs are shown:

jdbc-journal {
  dao = "org.apache.pekko.persistence.jdbc.journal.dao.DefaultJournalDao"
}

jdbc-snapshot-store {
  dao = "org.apache.pekko.persistence.jdbc.snapshot.dao.DefaultSnapshotDao"
}

jdbc-read-journal {
  dao = "org.apache.pekko.persistence.jdbc.query.dao.DefaultReadJournalDao"
}

Storing messages as byte arrays in blobs is not the only way to store information in a database. For example, you could store messages with full type information as a normal database rows, each event type having its own table. For example, implementing a Journal Log table that stores all persistenceId, sequenceNumber and event type discriminator field, and storing the event data in another table with full typing

You only have to implement two interfaces org.apache.pekko.persistence.jdbc.journal.dao.JournalDao and/or org.apache.pekko.persistence.jdbc.snapshot.dao.SnapshotDao.

For example, take a look at the following two custom DAOs:

class MyCustomJournalDao(db: Database, val profile: JdbcProfile, journalConfig: JournalConfig, serialization: Serialization)(implicit ec: ExecutionContext, mat: Materializer) extends JournalDao {
    // snip
}

class MyCustomSnapshotDao(db: JdbcBackend#Database, val profile: JdbcProfile, snapshotConfig: SnapshotConfig, serialization: Serialization)(implicit ec: ExecutionContext, val mat: Materializer) extends SnapshotDao {
    // snip
}

As you can see, the custom DAOs get a Slick database, a Slick profile, the journal or snapshot configuration, an org.apache.pekko.serialization.Serialization, an ExecutionContext and Materializer injected after constructed. You should register the Fully Qualified Class Name in application.conf so that the custom DAOs will be used.

Snapshot selection criteria

Snapshot loading and criteria-based deletion use SnapshotDao.snapshotForCriteria and SnapshotDao.deleteByCriteria. Both built-in snapshot DAOs apply all four bounds in SnapshotSelectionCriteria: minSequenceNr, maxSequenceNr, minTimestamp, and maxTimestamp. Bounds are inclusive, and both the sequence number and timestamp must match. Loading returns the matching snapshot with the highest sequence number; deletion removes only matching snapshots for the requested persistence ID. An empty interval matches no snapshots.

The existing upper-bound methods remain available. The default snapshotForCriteria implementation delegates to those methods and post-filters their result against all bounds, preserving safe recovery for existing custom DAOs. It may return no snapshot when timestamps are not ordered by sequence number even if an older snapshot matches. The default deleteByCriteria implementation preserves the previous upper-bound deletion behavior and may delete snapshots below a minimum bound, including when the interval is empty. Custom DAOs should override both criteria methods to query all four bounds directly, as the built-in DAOs do.

Subclasses of a built-in snapshot DAO that override its upper-bound methods must also override snapshotForCriteria and deleteByCriteria if they need to apply custom loading or deletion behavior. Criteria-based operations in the built-in DAOs execute their four-bound queries directly.

For more information please review the two default implementations org.apache.pekko.persistence.jdbc.dao.bytea.journal.ByteArrayJournalDao and org.apache.pekko.persistence.jdbc.dao.bytea.snapshot.ByteArraySnapshotDao or the demo custom DAO example from the demo-akka-persistence site.