Limit events by tag query ordering sizes

core/src/main/resources/reference.conf

@@ -396,6 +396,25 @@ jdbc-read-journal {
   # are delivered downstreams.
   max-buffer-size = "500"
 
+  # Number of 'max-buffer-size's to limit each events by tag query to
+  #
+  # Events by tag will fetch batches of elements limiting both using the DB LIMIT support and
+  # the "ordering" column of the journal. When executing a query starting from the beginning of the
+  # journal, for example adding a new projection to an existing application with a large number
+  # of already persisted events this can cause performance problems in some databases.
+  #
+  # This factor limits the "slices" of ordering the journal is queried for into smaller chunks,
+  # issuing more queries where each query covers a smaller slice of the journal instead of one
+  # covering the entire journal.
+  #
+  # Note that setting this too low will have a performance overhead in many queries being issued where
+  # each query returns no or very few entries, but what number is to low depends on how many tags are
+  # used and how well those are distributed, setting this value requires application specific benchmarking
+  # to find a good number.
+  #
+  # 0 means disable the factor and query the entire journal and limit to max-buffer-size elements
+  events-by-tag-buffer-sizes-per-query = 0
+
   # If enabled, automatically close the database connection when the actor system is terminated
   add-shutdown-hook = true
 

core/src/main/scala/akka/persistence/jdbc/config/AkkaPersistenceConfig.scala

@@ -187,6 +187,7 @@ class ReadJournalConfig(config: Config) {
   val pluginConfig = new ReadJournalPluginConfig(config)
   val refreshInterval: FiniteDuration = config.asFiniteDuration("refresh-interval")
   val maxBufferSize: Int = config.getInt("max-buffer-size")
+  val eventsByTagBufferSizesPerQuery: Long = config.getLong("events-by-tag-buffer-sizes-per-query")
   val addShutdownHook: Boolean = config.getBoolean("add-shutdown-hook")
 
   override def toString: String =

core/src/main/scala/akka/persistence/jdbc/query/scaladsl/JdbcReadJournal.scala

@@ -228,15 +228,29 @@ class JdbcReadJournal(config: Config, configPath: String)(implicit val system: E
     import FlowControl._
     implicit val askTimeout: Timeout = Timeout(readJournalConfig.journalSequenceRetrievalConfiguration.askTimeout)
     val batchSize = readJournalConfig.maxBufferSize
+    val maxOrderingRange = readJournalConfig.eventsByTagBufferSizesPerQuery match {
+      case 0 => None
+      case x => Some(x * batchSize)
+    }
+
+    def getLoopMaxOrderingId(offset: Long, latestOrdering: MaxOrderingId): Future[MaxOrderingId] =
+      Future.successful(maxOrderingRange match {
+        case None => latestOrdering
+        case Some(numberOfEvents) =>
+          val limitedMaxOrderingId = offset + numberOfEvents
+          if (limitedMaxOrderingId < 0 || limitedMaxOrderingId >= latestOrdering.maxOrdering) latestOrdering
+          else MaxOrderingId(limitedMaxOrderingId)
+      })
 
     Source
       .unfoldAsync[(Long, FlowControl), Seq[EventEnvelope]]((offset, Continue)) { case (from, control) =>
         def retrieveNextBatch() = {
           for {
             queryUntil <- journalSequenceActor.ask(GetMaxOrderingId).mapTo[MaxOrderingId]
-            xs <- currentJournalEventsByTag(tag, from, batchSize, queryUntil).runWith(Sink.seq)
+            loopMaxOrderingId <- getLoopMaxOrderingId(from, queryUntil)
+            xs <- currentJournalEventsByTag(tag, from, batchSize, loopMaxOrderingId).runWith(Sink.seq)
           } yield {
-            val hasMoreEvents = xs.size == batchSize
+            val hasMoreEvents = (xs.size == batchSize) || (loopMaxOrderingId.maxOrdering < queryUntil.maxOrdering)
             val nextControl: FlowControl =
               terminateAfterOffset match {
                 // we may stop if target is behind queryUntil and we don't have more events to fetch