Possibility to opt out from all_persistence_ids table, #784

akka · May 26, 2020 · 40c7615 · 40c7615
1 parent 68aa9a6
commit 40c7615
Show file tree

Hide file tree

Showing 6 changed files with 39 additions and 12 deletions.
diff --git a/core/src/main/resources/reference.conf b/core/src/main/resources/reference.conf
@@ -225,6 +225,15 @@ akka.persistence.cassandra {
     # replays.
     deserialization-parallelism = 1
 
+    # For applications that are not using persistenceIds or currentPersistenceIds queries
+    # this can be set to 'off', which will optimize the write of the first event for each
+    # persistent actor since the all_persistence_id table doesn't have to be populated.
+    # Note that the Cleanup and Reconcilation tools may also use persistenceIds queries
+    # and those will not work if this was disabled when the events were written. In
+    # that case the all_persistence_id table can be reconstructed with
+    # Reconcilation.rebuildAllPersistenceIds.
+    support-all-persistence-ids = on
+
   }
   //#query
 

diff --git a/core/src/main/scala/akka/persistence/cassandra/journal/CassandraJournal.scala b/core/src/main/scala/akka/persistence/cassandra/journal/CassandraJournal.scala
@@ -197,7 +197,8 @@ import akka.stream.scaladsl.Source
       preparedWriteMessageWithMeta
       preparedSelectMessages
       preparedSelectHighestSequenceNr
-      preparedInsertIntoAllPersistenceIds
+      if (settings.querySettings.supportAllPersistenceIds)
+        preparedInsertIntoAllPersistenceIds
       if (settings.journalSettings.supportDeletes) {
         preparedDeleteMessages
         preparedSelectDeletedTo
@@ -348,7 +349,7 @@ import akka.stream.scaladsl.Source
   private def writeMessages(atomicWrites: Seq[SerializedAtomicWrite]): Future[Unit] = {
     // insert into the all_persistence_ids table for the first event, used by persistenceIds query
     val allPersistenceId =
-      if (atomicWrites.head.payload.head.sequenceNr == 1L)
+      if (settings.querySettings.supportAllPersistenceIds && atomicWrites.head.payload.head.sequenceNr == 1L)
         preparedInsertIntoAllPersistenceIds.map(_.bind(atomicWrites.head.persistenceId)).flatMap(execute(_))
       else
         FutureUnit

diff --git a/core/src/main/scala/akka/persistence/cassandra/journal/CassandraJournalStatements.scala b/core/src/main/scala/akka/persistence/cassandra/journal/CassandraJournalStatements.scala
@@ -353,7 +353,12 @@ import akka.persistence.cassandra.FutureDone
         _ <- keyspace
         _ <- session.executeAsync(createTable).toScala
         _ <- session.executeAsync(createMetadataTable).toScala
-        _ <- session.executeAsync(createAllPersistenceIdsTable).toScala
+        _ <- {
+          if (settings.querySettings.supportAllPersistenceIds)
+            session.executeAsync(createAllPersistenceIdsTable).toScala
+          else
+            FutureDone
+        }
         _ <- tagStatements
       } yield {
         session.setSchemaMetadataEnabled(null)

diff --git a/core/src/main/scala/akka/persistence/cassandra/query/QuerySettings.scala b/core/src/main/scala/akka/persistence/cassandra/query/QuerySettings.scala
@@ -41,4 +41,5 @@ import com.typesafe.config.Config
   val eventsByPersistenceIdEventTimeout: FiniteDuration =
     queryConfig.getDuration("events-by-persistence-id-gap-timeout", MILLISECONDS).millis
 
+  val supportAllPersistenceIds: Boolean = queryConfig.getBoolean("support-all-persistence-ids")
 }
diff --git a/core/src/main/scala/akka/persistence/cassandra/query/scaladsl/CassandraReadJournal.scala b/core/src/main/scala/akka/persistence/cassandra/query/scaladsl/CassandraReadJournal.scala
@@ -307,7 +307,10 @@ class CassandraReadJournal protected (
    */
   @InternalApi private[akka] def eventsByTagInternal(tag: String, offset: Offset): Source[UUIDPersistentRepr, NotUsed] =
     if (!eventsByTagSettings.eventsByTagEnabled)
-      Source.failed(new IllegalStateException("Events by tag queries are disabled"))
+      Source.failed(
+        new IllegalStateException(
+          "Events by tag queries are disabled with configuration " +
+          "events-by-tag.enabled=off"))
     else {
       try {
         val (fromOffset, usingOffset) = offsetToInternalOffset(offset)
@@ -696,14 +699,21 @@ class CassandraReadJournal protected (
     persistenceIds(None, "currentPersistenceIds")
 
   private def persistenceIds(refreshInterval: Option[FiniteDuration], name: String): Source[String, NotUsed] =
-    createSource[String, PreparedStatement](
-      preparedSelectAllPersistenceIds,
-      (s, ps) =>
-        Source
-          .fromGraph(new AllPersistenceIdsStage(refreshInterval, ps, s, querySettings.readProfile))
-          .withAttributes(ActorAttributes.dispatcher(querySettings.pluginDispatcher))
-          .mapMaterializedValue(_ => NotUsed)
-          .named(name))
+    if (!querySettings.supportAllPersistenceIds)
+      Source.failed(
+        new IllegalStateException(
+          "persistenceIds queries are disabled with configuration " +
+          "support-all-persistence-ids=off"))
+    else {
+      createSource[String, PreparedStatement](
+        preparedSelectAllPersistenceIds,
+        (s, ps) =>
+          Source
+            .fromGraph(new AllPersistenceIdsStage(refreshInterval, ps, s, querySettings.readProfile))
+            .withAttributes(ActorAttributes.dispatcher(querySettings.pluginDispatcher))
+            .mapMaterializedValue(_ => NotUsed)
+            .named(name))
+    }
 
   /**
    * INTERNAL API: Needed for migration to 1.0

diff --git a/core/src/test/scala/akka/persistence/cassandra/journal/ManyActorsLoadSpec.scala b/core/src/test/scala/akka/persistence/cassandra/journal/ManyActorsLoadSpec.scala
@@ -19,6 +19,7 @@ object ManyActorsLoadSpec {
   val config = ConfigFactory.parseString(s"""
       akka.persistence.cassandra.journal.keyspace=ManyActorsLoadSpec
       akka.persistence.cassandra.events-by-tag.enabled = on
+      akka.persistence.cassandra.query.support-all-persistence-ids = off
       # increase this to 3s when benchmarking
       akka.persistence.cassandra.events-by-tag.scanning-flush-interval = 1s
       #akka.persistence.cassandra.log-queries = on