perf: use subqueries over series for latest event timestamp (#671)

* perf: use subqueries over series for latest event timestamp

* adjust test for multiple partitions

Author: pvlugter

core/src/main/scala/akka/persistence/r2dbc/internal/postgres/PostgresQueryDao.scala

@@ -135,14 +135,20 @@ private[r2dbc] class PostgresQueryDao(executorProvider: R2dbcExecutorProvider) e
       WHERE persistence_id = ? AND seq_nr = ? AND deleted = $sqlFalse"""
     }
 
-  protected def selectLatestEventTimestampSql(minSlice: Int, maxSlice: Int): String =
-    sqlCache.get(minSlice, s"selectLatestEventTimestampSql-$minSlice-$maxSlice") {
+  protected def selectLatestEventTimestampSql(slice: Int): String =
+    sqlCache.get(slice, "selectLatestEventTimestampSql") {
       sql"""
-      SELECT MAX(db_timestamp) AS latest_timestamp
-      FROM ${journalTable(minSlice)}
-      WHERE entity_type = ?
-      AND ${sliceCondition(minSlice, maxSlice)}
-      AND deleted = $sqlFalse
+      SELECT MAX(per_slice.latest_timestamp) AS latest_timestamp
+      FROM (
+        SELECT
+          (SELECT MAX(db_timestamp)
+           FROM ${journalTable(slice)}
+           WHERE entity_type = ?
+           AND slice = slice_range.slice
+           AND deleted = $sqlFalse) AS latest_timestamp
+        FROM (SELECT * FROM generate_series(?, ?)) AS slice_range(slice)
+      ) per_slice
+      WHERE per_slice.latest_timestamp IS NOT NULL;
       """
     }
 
@@ -412,8 +418,10 @@ private[r2dbc] class PostgresQueryDao(executorProvider: R2dbcExecutorProvider) e
       .selectOne(s"select latest event timestamp for entity type [$entityType] slice range [$minSlice - $maxSlice]")(
         connection =>
           connection
-            .createStatement(selectLatestEventTimestampSql(minSlice, maxSlice))
-            .bind(0, entityType),
+            .createStatement(selectLatestEventTimestampSql(minSlice))
+            .bind(0, entityType)
+            .bind(1, minSlice)
+            .bind(2, maxSlice),
         row => Option.when(row.get("latest_timestamp") ne null)(row.getTimestamp("latest_timestamp")))
       .map(_.flatten)(ExecutionContext.parasitic)
 

core/src/test/scala/akka/persistence/r2dbc/query/EventsBySliceSpec.scala

@@ -292,24 +292,29 @@ class EventsBySliceSpec
 
         query shouldBe a[LatestEventTimestampQuery]
 
-        {
-          // test all slice ranges, with the events expected in one of the ranges
-          val numRanges = 4
-          val rangeSize = 1024 / numRanges
-          val expectedRangeIndex = slice / rangeSize
-
-          def sliceRange(rangeIndex: Int): (Int, Int) = {
-            val minSlice = rangeIndex * rangeSize
-            val maxSlice = minSlice + rangeSize - 1
-            minSlice -> maxSlice
-          }
+        val partitions = settings.numberOfDataPartitions
+        val testNumRanges =
+          if (partitions > 1) List(partitions, partitions * 2, 1024)
+          else List(1, 4, 1024)
+        testNumRanges.foreach { numRanges =>
+          withClue(s"numRanges=$numRanges: ") {
+            // test all slice ranges, with the events expected in one of the ranges
+            val rangeSize = 1024 / numRanges
+            val expectedRangeIndex = slice / rangeSize
+
+            def sliceRange(rangeIndex: Int): (Int, Int) = {
+              val minSlice = rangeIndex * rangeSize
+              val maxSlice = minSlice + rangeSize - 1
+              minSlice -> maxSlice
+            }
 
-          for (rangeIndex <- 0 until numRanges) {
-            val (minSlice, maxSlice) = sliceRange(rangeIndex)
-            val expectedTimestamp =
-              if (rangeIndex != expectedRangeIndex) None
-              else query.timestampOf(persistenceId, 3L).futureValue
-            query.latestEventTimestamp(entityType, minSlice, maxSlice).futureValue shouldBe expectedTimestamp
+            for (rangeIndex <- 0 until numRanges) {
+              val (minSlice, maxSlice) = sliceRange(rangeIndex)
+              val expectedTimestamp =
+                if (rangeIndex != expectedRangeIndex) None
+                else query.timestampOf(persistenceId, 3L).futureValue
+              query.latestEventTimestamp(entityType, minSlice, maxSlice).futureValue shouldBe expectedTimestamp
+            }
           }
         }
       }