Annotated queries to review as part of #155

Author: fredfp

Diff for: core/src/main/scala/akka/persistence/postgres/journal/dao/JournalQueries.scala

@@ -20,7 +20,7 @@ class JournalQueries(journalTable: TableQuery[JournalTable]) {
     compiledJournalTable ++= xs.sortBy(_.sequenceNumber)
 
   private def selectAllJournalForPersistenceId(persistenceId: Rep[String]) =
-    journalTable.filter(_.persistenceId === persistenceId).sortBy(_.sequenceNumber.desc)
+    journalTable.filter(_.persistenceId === persistenceId).sortBy(_.sequenceNumber.desc) // why .desc ?
 
   def delete(persistenceId: String, toSequenceNr: Long): FixedSqlAction[Int, NoStream, slick.dbio.Effect.Write] = {
     journalTable.filter(_.persistenceId === persistenceId).filter(_.sequenceNumber <= toSequenceNr).delete
@@ -54,20 +54,28 @@ class JournalQueries(journalTable: TableQuery[JournalTable]) {
   private def _highestMarkedSequenceNrForPersistenceId(persistenceId: Rep[String]): Rep[Option[Long]] =
     journalTable.filter(_.deleted === true).filter(_.persistenceId === persistenceId).map(_.sequenceNumber).max
 
+  // metadata lookup
   val highestSequenceNrForPersistenceId = Compiled(_highestSequenceNrForPersistenceId _)
 
+  // metadata lookup
   val highestMarkedSequenceNrForPersistenceId = Compiled(_highestMarkedSequenceNrForPersistenceId _)
 
+  // not used?
   private def _selectByPersistenceIdAndMaxSequenceNumber(persistenceId: Rep[String], maxSequenceNr: Rep[Long]) =
     selectAllJournalForPersistenceId(persistenceId).filter(_.sequenceNumber <= maxSequenceNr)
 
+  // not used?
   val selectByPersistenceIdAndMaxSequenceNumber = Compiled(_selectByPersistenceIdAndMaxSequenceNumber _)
 
+  // not used, remove or dedup with read journal
   private def _allPersistenceIdsDistinct: Query[Rep[String], String, Seq] =
     journalTable.map(_.persistenceId).distinct
 
+  // not used, remove or dedup with read journal
   val allPersistenceIdsDistinct = Compiled(_allPersistenceIdsDistinct)
 
+  // for partition pruning, how to integrate min/max ordering?
+  // at least do where ordering >= metadata.min_ordering
   private def _messagesQuery(
       persistenceId: Rep[String],
       fromSequenceNr: Rep[Long],

Diff for: core/src/main/scala/akka/persistence/postgres/query/dao/ByteArrayReadJournalDao.scala

@@ -32,6 +32,8 @@ trait BaseByteArrayReadJournalDao extends ReadJournalDao with BaseJournalDaoWith
   import akka.persistence.postgres.db.ExtendedPostgresProfile.api._
 
   override def allPersistenceIdsSource(max: Long): Source[String, NotUsed] =
+  // this is fully buffered and not streamed, see note at https://scala-slick.org/doc/3.3.2/dbio.html#streaming
+  // either batch or stream
     Source.fromPublisher(db.stream(queries.allPersistenceIdsDistinct(max).result))
 
   override def eventsByTag(

Diff for: core/src/main/scala/akka/persistence/postgres/query/dao/ReadJournalQueries.scala

@@ -14,6 +14,7 @@ class ReadJournalQueries(val readJournalConfig: ReadJournalConfig) {
 
   private val journalTable: TableQuery[JournalTable] = FlatJournalTable(readJournalConfig.journalTableConfiguration)
 
+  // from metadata table to avoid distinct call
   private def _allPersistenceIdsDistinct(max: ConstColumn[Long]): Query[Rep[String], String, Seq] =
     baseTableQuery().map(_.persistenceId).distinct.take(max)
 
@@ -23,6 +24,7 @@ class ReadJournalQueries(val readJournalConfig: ReadJournalConfig) {
 
   val allPersistenceIdsDistinct = Compiled(_allPersistenceIdsDistinct _)
 
+  // for partition pruning, consider adding a filter on ordering >= metadata.min_ordering
   private def _messagesQuery(
       persistenceId: Rep[String],
       fromSequenceNr: Rep[Long],
@@ -54,6 +56,7 @@ class ReadJournalQueries(val readJournalConfig: ReadJournalConfig) {
 
   val orderingByOrdering = Compiled(_journalSequenceQuery _)
 
+  // this could also be done using metadata table, to be evaluated
   val maxOrdering = Compiled {
     journalTable.map(_.ordering).max.getOrElse(0L)
   }

Diff for: core/src/main/scala/akka/persistence/postgres/query/scaladsl/PostgresReadJournal.scala

@@ -111,7 +111,7 @@ class PostgresReadJournal(config: Config, configPath: String)(implicit val syste
       .repeat(0)
       .flatMapConcat(_ => delaySource.flatMapConcat(_ => currentPersistenceIds()))
       .statefulMapConcat[String] { () =>
-        var knownIds = Set.empty[String]
+        var knownIds = Set.empty[String] // get rid of that and instead use natural ordering of persistenceIds in metadata when sorted by id
         def next(id: String): Iterable[String] = {
           val xs = Set(id).diff(knownIds)
           knownIds += id