Add support for querying last known sequence number by persistenceId.

core/src/main/scala/org/apache/pekko/persistence/jdbc/query/dao/DefaultReadJournalDao.scala

@@ -60,6 +60,12 @@ class DefaultReadJournalDao(
   override def journalSequence(offset: Long, limit: Long): Source[Long, NotUsed] =
     Source.fromPublisher(db.stream(queries.journalSequenceQuery((offset, limit)).result))
 
+  override def lastPersistenceIdSequenceNumber(
+      persistenceId: String
+  ): Future[Option[Long]] = {
+    db.run(queries.lastPersistenceIdSequenceNumberQuery(persistenceId).result)
+  }
+
   override def maxJournalSequence(): Future[Long] =
     db.run(queries.maxJournalSequenceQuery.result)
 

core/src/main/scala/org/apache/pekko/persistence/jdbc/query/dao/ReadJournalDao.scala

@@ -49,6 +49,16 @@ trait ReadJournalDao extends JournalDaoWithReadMessages {
    */
   def journalSequence(offset: Long, limit: Long): Source[Long, NotUsed]
 
+  /**
+   * Returns the last known sequence number for the given `persistenceId`. Empty if the `persistenceId` is unknown.
+   *
+   * @param persistenceId The `persistenceId` for which the last known sequence number should be returned.
+   * @return Some sequence number or None if the `persistenceId` is unknown.
+   */
+  def lastPersistenceIdSequenceNumber(
+      persistenceId: String
+  ): Future[Option[Long]]
+
   /**
    * @return The value of the maximum (ordering) id in the journal
    */

core/src/main/scala/org/apache/pekko/persistence/jdbc/query/dao/ReadJournalQueries.scala

@@ -29,6 +29,9 @@ class ReadJournalQueries(val profile: JdbcProfile, val readJournalConfig: ReadJo
   val messagesQuery = Compiled(_messagesQuery _)
   val eventsByTag = Compiled(_eventsByTag _)
   val journalSequenceQuery = Compiled(_journalSequenceQuery _)
+  val lastPersistenceIdSequenceNumberQuery = Compiled {
+    _lastPersistenceIdSequenceNumberQuery _
+  }
   val maxJournalSequenceQuery = Compiled {
     JournalTable.map(_.ordering).max.getOrElse(0L)
   }
@@ -43,6 +46,16 @@ class ReadJournalQueries(val profile: JdbcProfile, val readJournalConfig: ReadJo
     baseTableQuery().join(TagTable).on(_.ordering === _.eventId)
   }
 
+  private def _lastPersistenceIdSequenceNumberQuery(
+      persistenceId: Rep[String]
+  ) = {
+
+    baseTableQuery()
+      .filter(_.persistenceId === persistenceId)
+      .map(_.sequenceNumber)
+      .max
+  }
+
   private def _messagesQuery(
       persistenceId: Rep[String],
       fromSequenceNr: Rep[Long],

core/src/main/scala/org/apache/pekko/persistence/jdbc/query/dao/legacy/ByteArrayReadJournalDao.scala

@@ -59,6 +59,12 @@ trait BaseByteArrayReadJournalDao extends ReadJournalDao with BaseJournalDaoWith
       .via(serializer.deserializeFlow)
   }
 
+  override def lastPersistenceIdSequenceNumber(
+      persistenceId: String
+  ): Future[Option[Long]] = {
+    db.run(queries.lastPersistenceIdSequenceNumberQuery(persistenceId).result)
+  }
+
   override def messages(
       persistenceId: String,
       fromSequenceNr: Long,

core/src/main/scala/org/apache/pekko/persistence/jdbc/query/dao/legacy/ReadJournalQueries.scala

@@ -28,6 +28,9 @@ class ReadJournalQueries(val profile: JdbcProfile, val readJournalConfig: ReadJo
   val messagesQuery = Compiled(_messagesQuery _)
   val eventsByTag = Compiled(_eventsByTag _)
   val journalSequenceQuery = Compiled(_journalSequenceQuery _)
+  val lastPersistenceIdSequenceNumberQuery = Compiled {
+    _lastPersistenceIdSequenceNumberQuery _
+  }
   val maxJournalSequenceQuery = Compiled {
     JournalTable.map(_.ordering).max.getOrElse(0L)
   }
@@ -38,6 +41,16 @@ class ReadJournalQueries(val profile: JdbcProfile, val readJournalConfig: ReadJo
   private def _allPersistenceIdsDistinct(max: ConstColumn[Long]): Query[Rep[String], String, Seq] =
     baseTableQuery().map(_.persistenceId).distinct.take(max)
 
+  private def _lastPersistenceIdSequenceNumberQuery(
+      persistenceId: Rep[String]
+  ) = {
+
+    baseTableQuery()
+      .filter(_.persistenceId === persistenceId)
+      .map(_.sequenceNumber)
+      .max
+  }
+
   private def _messagesQuery(
       persistenceId: Rep[String],
       fromSequenceNr: Rep[Long],

core/src/main/scala/org/apache/pekko/persistence/jdbc/query/javadsl/CurrentLastKnownSequenceNumberByPersistenceIdQuery.scala

@@ -0,0 +1,29 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+package org.apache.pekko.persistence.jdbc.query.javadsl
+
+import java.util.Optional
+import java.util.concurrent.CompletionStage
+
+/**
+ * A trait that enables querying the current last known sequence number for a given `persistenceId`.
+ */
+trait CurrentLastKnownSequenceNumberByPersistenceIdQuery {
+
+  /**
+   * Returns the last known sequence number for the given `persistenceId`. Empty if the `persistenceId` is unknown.
+   *
+   * @param persistenceId The `persistenceId` for which the last known sequence number should be returned.
+   * @return Some sequence number or None if the `persistenceId` is unknown.
+   */
+  def currentLastKnownSequenceNumberByPersistenceId(
+      persistenceId: String
+  ): CompletionStage[Optional[Long]]
+}

core/src/main/scala/org/apache/pekko/persistence/jdbc/query/javadsl/JdbcReadJournal.scala

@@ -15,11 +15,17 @@
 package org.apache.pekko.persistence.jdbc.query.javadsl
 
 import org.apache.pekko
+
+import java.util.Optional
+import java.util.concurrent.CompletionStage
+
 import pekko.NotUsed
 import pekko.persistence.jdbc.query.scaladsl.{ JdbcReadJournal => ScalaJdbcReadJournal }
 import pekko.persistence.query.{ EventEnvelope, Offset }
 import pekko.persistence.query.javadsl._
 import pekko.stream.javadsl.Source
+import pekko.util.FutureConverters._
+import pekko.util.OptionConverters._
 
 object JdbcReadJournal {
   final val Identifier = ScalaJdbcReadJournal.Identifier
@@ -32,7 +38,8 @@ class JdbcReadJournal(journal: ScalaJdbcReadJournal)
     with CurrentEventsByPersistenceIdQuery
     with EventsByPersistenceIdQuery
     with CurrentEventsByTagQuery
-    with EventsByTagQuery {
+    with EventsByTagQuery
+    with CurrentLastKnownSequenceNumberByPersistenceIdQuery {
 
   /**
    * Same type of query as `persistenceIds` but the event stream
@@ -132,4 +139,14 @@ class JdbcReadJournal(journal: ScalaJdbcReadJournal)
    */
   override def eventsByTag(tag: String, offset: Offset): Source[EventEnvelope, NotUsed] =
     journal.eventsByTag(tag, offset).asJava
+
+  override def currentLastKnownSequenceNumberByPersistenceId(
+      persistenceId: String
+  ): CompletionStage[Optional[Long]] = {
+
+    journal
+      .currentLastKnownSequenceNumberByPersistenceId(persistenceId)
+      .asJava
+      .thenApply(_.toJava)
+  }
 }

core/src/main/scala/org/apache/pekko/persistence/jdbc/query/scaladsl/CurrentLastKnownSequenceNumberByPersistenceIdQuery.scala

@@ -0,0 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+package org.apache.pekko.persistence.jdbc.query.scaladsl
+
+import scala.concurrent.Future
+
+/**
+ * A trait that enables querying the current last known sequence number for a given `persistenceId`.
+ */
+trait CurrentLastKnownSequenceNumberByPersistenceIdQuery {
+
+  /**
+   * Returns the last known sequence number for the given `persistenceId`. Empty if the `persistenceId` is unknown.
+   *
+   * @param persistenceId The `persistenceId` for which the last known sequence number should be returned.
+   * @return Some sequence number or None if the `persistenceId` is unknown.
+   */
+  def currentLastKnownSequenceNumberByPersistenceId(
+      persistenceId: String
+  ): Future[Option[Long]]
+}

core/src/main/scala/org/apache/pekko/persistence/jdbc/query/scaladsl/JdbcReadJournal.scala

@@ -52,7 +52,8 @@ class JdbcReadJournal(config: Config, configPath: String)(implicit val system: E
     with CurrentEventsByPersistenceIdQuery
     with EventsByPersistenceIdQuery
     with CurrentEventsByTagQuery
-    with EventsByTagQuery {
+    with EventsByTagQuery
+    with CurrentLastKnownSequenceNumberByPersistenceIdQuery {
 
   PluginVersionChecker.check()
 
@@ -317,4 +318,11 @@ class JdbcReadJournal(config: Config, configPath: String)(implicit val system: E
 
   def eventsByTag(tag: String, offset: Long): Source[EventEnvelope, NotUsed] =
     eventsByTag(tag, offset, terminateAfterOffset = None)
+
+  override def currentLastKnownSequenceNumberByPersistenceId(
+    persistenceId: String
+  ): Future[Option[Long]] = {
+
+    readJournalDao.lastPersistenceIdSequenceNumber(persistenceId)
+  }
 }

core/src/test/scala/org/apache/pekko/persistence/jdbc/query/CurrentLastKnownSequenceNumberByPersistenceIdTest.scala

@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+package org.apache.pekko.persistence.jdbc.query
+
+import org.scalatest.concurrent.ScalaFutures
+
+abstract class CurrentLastKnownSequenceNumberByPersistenceIdTest(
+    config: String
+) extends QueryTestSpec(config) with ScalaFutures {
+
+  it should "return None for unknown persistenceId" in withActorSystem { implicit system =>
+    val journalOps = new ScalaJdbcReadJournalOperations(system)
+
+    journalOps
+      .currentLastKnownSequenceNumberByPersistenceId("unknown")
+      .futureValue shouldBe None
+  }
+
+  it should "return last sequence number for known persistenceId" in withActorSystem { implicit system =>
+    val journalOps = new ScalaJdbcReadJournalOperations(system)
+
+    withTestActors() { (actor1, _, _) =>
+      actor1 ! 1
+      actor1 ! 2
+      actor1 ! 3
+      actor1 ! 4
+
+      eventually {
+        journalOps
+          .currentLastKnownSequenceNumberByPersistenceId("my-1")
+          .futureValue shouldBe Some(4)
+
+        // Just ensuring that query targets the correct persistenceId.
+        journalOps
+          .currentLastKnownSequenceNumberByPersistenceId("my-2")
+          .futureValue shouldBe None
+      }
+    }
+  }
+}
+
+class H2ScalaCurrentLastKnownSequenceNumberByPersistenceIdTest
+    extends CurrentLastKnownSequenceNumberByPersistenceIdTest("h2-shared-db-application.conf")
+    with H2Cleaner

core/src/test/scala/org/apache/pekko/persistence/jdbc/query/QueryTestSpec.scala

@@ -110,6 +110,13 @@ class ScalaJdbcReadJournalOperations(readJournal: JdbcReadJournal)(implicit syst
     tp.within(within)(f(tp))
   }
 
+  def currentLastKnownSequenceNumberByPersistenceId(
+      persistenceId: String
+  ): Future[Option[Long]] = {
+
+    readJournal.currentLastKnownSequenceNumberByPersistenceId(persistenceId)
+  }
+
   override def countJournal: Future[Long] =
     readJournal
       .currentPersistenceIds()

core/src/test/scala/org/apache/pekko/persistence/jdbc/query/dao/TestProbeReadJournalDao.scala

@@ -54,6 +54,10 @@ class TestProbeReadJournalDao(val probe: TestProbe) extends ReadJournalDao {
       maxOffset: Long,
       max: Long): Source[Try[(PersistentRepr, Set[String], Long)], NotUsed] = ???
 
+  override def lastPersistenceIdSequenceNumber(
+    persistenceId: String
+  ): Future[Option[Long]] = ???
+
   /**
    * Returns a Source of bytes for a certain persistenceId
    */