You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pekko.apache.org by ro...@apache.org on 2024/04/24 08:03:26 UTC
(pekko-persistence-jdbc) branch main updated: perf: improve journal deletion (#169)
This is an automated email from the ASF dual-hosted git repository.
roiocam pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko-persistence-jdbc.git
The following commit(s) were added to refs/heads/main by this push:
new 8e170c2 perf: improve journal deletion (#169)
8e170c2 is described below
commit 8e170c21e6741a088cfe2356ea9371045f8d6b2e
Author: AndyChen(Jingzhang) <iR...@gmail.com>
AuthorDate: Wed Apr 24 16:03:22 2024 +0800
perf: improve journal deletion (#169)
* feat: remove logic delete for optimization
---
.../1.1.x.backwards.excludes/BaseDao.excludes | 17 +++++++++
.../deletion-optimize.excludes | 29 ++++++++++++++
.../jdbc/journal/dao/DefaultJournalDao.scala | 12 +++---
.../jdbc/journal/dao/JournalQueries.scala | 44 +++++++---------------
.../journal/dao/legacy/ByteArrayJournalDao.scala | 10 ++---
.../jdbc/journal/dao/legacy/JournalQueries.scala | 42 ++++++---------------
.../jdbc/query/dao/ReadJournalQueries.scala | 31 +++++----------
.../jdbc/query/dao/legacy/ReadJournalQueries.scala | 30 +++++----------
8 files changed, 102 insertions(+), 113 deletions(-)
diff --git a/core/src/main/mima-filters/1.1.x.backwards.excludes/BaseDao.excludes b/core/src/main/mima-filters/1.1.x.backwards.excludes/BaseDao.excludes
index 79eeae1..1468d36 100644
--- a/core/src/main/mima-filters/1.1.x.backwards.excludes/BaseDao.excludes
+++ b/core/src/main/mima-filters/1.1.x.backwards.excludes/BaseDao.excludes
@@ -1,2 +1,19 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
ProblemFilters.exclude[IncompatibleResultTypeProblem]("org.apache.pekko.persistence.jdbc.journal.dao.BaseDao.writeQueue")
ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.persistence.jdbc.journal.dao.BaseDao.writeQueue")
diff --git a/core/src/main/mima-filters/1.1.x.backwards.excludes/deletion-optimize.excludes b/core/src/main/mima-filters/1.1.x.backwards.excludes/deletion-optimize.excludes
new file mode 100644
index 0000000..462d014
--- /dev/null
+++ b/core/src/main/mima-filters/1.1.x.backwards.excludes/deletion-optimize.excludes
@@ -0,0 +1,29 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.persistence.jdbc.journal.dao.JournalQueries.markJournalMessagesAsDeleted")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.persistence.jdbc.journal.dao.legacy.JournalQueries.markJournalMessagesAsDeleted")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.persistence.jdbc.journal.dao.JournalQueries.highestMarkedSequenceNrForPersistenceId")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.persistence.jdbc.journal.dao.legacy.JournalQueries.highestMarkedSequenceNrForPersistenceId")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.persistence.jdbc.journal.dao.JournalQueries.selectByPersistenceIdAndMaxSequenceNumber")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.persistence.jdbc.journal.dao.legacy.JournalQueries.selectByPersistenceIdAndMaxSequenceNumber")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.persistence.jdbc.journal.dao.JournalQueries.allPersistenceIdsDistinct")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.persistence.jdbc.journal.dao.legacy.JournalQueries.allPersistenceIdsDistinct")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.persistence.jdbc.journal.dao.JournalQueries.journalRowByPersistenceIds")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.persistence.jdbc.journal.dao.legacy.JournalQueries.journalRowByPersistenceIds")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.persistence.jdbc.query.dao.ReadJournalQueries.journalRowByPersistenceIds")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.persistence.jdbc.query.dao.legacy.ReadJournalQueries.journalRowByPersistenceIds")
diff --git a/core/src/main/scala/org/apache/pekko/persistence/jdbc/journal/dao/DefaultJournalDao.scala b/core/src/main/scala/org/apache/pekko/persistence/jdbc/journal/dao/DefaultJournalDao.scala
index e982f9d..dd67adf 100644
--- a/core/src/main/scala/org/apache/pekko/persistence/jdbc/journal/dao/DefaultJournalDao.scala
+++ b/core/src/main/scala/org/apache/pekko/persistence/jdbc/journal/dao/DefaultJournalDao.scala
@@ -59,10 +59,13 @@ class DefaultJournalDao(
new JournalQueries(profile, journalConfig.eventJournalTableConfiguration, journalConfig.eventTagTableConfiguration)
override def delete(persistenceId: String, maxSequenceNr: Long): Future[Unit] = {
+ // We should keep journal record with highest sequence number in order to be compliant
+ // with @see [[pekko.persistence.journal.JournalSpec]]
val actions: DBIOAction[Unit, NoStream, Effect.Write with Effect.Read] = for {
- _ <- queries.markJournalMessagesAsDeleted(persistenceId, maxSequenceNr)
- highestMarkedSequenceNr <- highestMarkedSequenceNr(persistenceId)
- _ <- queries.delete(persistenceId, highestMarkedSequenceNr.getOrElse(0L) - 1)
+ seq <- queries.highestSequenceNrForPersistenceIdBefore((persistenceId, maxSequenceNr)).result
+ highestSequenceNr = seq.headOption.getOrElse(0L)
+ _ <- queries.delete(persistenceId, highestSequenceNr - 1)
+ _ <- queries.markSeqNrJournalMessagesAsDeleted(persistenceId, highestSequenceNr)
} yield ()
db.run(actions.transactionally)
@@ -74,9 +77,6 @@ class DefaultJournalDao(
} yield maybeHighestSeqNo.getOrElse(0L)
}
- private def highestMarkedSequenceNr(persistenceId: String) =
- queries.highestMarkedSequenceNrForPersistenceId(persistenceId).result
-
override def asyncWriteMessages(messages: immutable.Seq[AtomicWrite]): Future[immutable.Seq[Try[Unit]]] = {
def serializeAtomicWrite(aw: AtomicWrite): Try[Seq[(JournalPekkoSerializationRow, Set[String])]] = {
diff --git a/core/src/main/scala/org/apache/pekko/persistence/jdbc/journal/dao/JournalQueries.scala b/core/src/main/scala/org/apache/pekko/persistence/jdbc/journal/dao/JournalQueries.scala
index 9dc6cca..751e11c 100644
--- a/core/src/main/scala/org/apache/pekko/persistence/jdbc/journal/dao/JournalQueries.scala
+++ b/core/src/main/scala/org/apache/pekko/persistence/jdbc/journal/dao/JournalQueries.scala
@@ -33,6 +33,10 @@ class JournalQueries(
private val insertAndReturn = JournalTable.returning(JournalTable.map(_.ordering))
private val TagTableC = Compiled(TagTable)
+ val highestSequenceNrForPersistenceId = Compiled(_highestSequenceNrForPersistenceId _)
+ val highestSequenceNrForPersistenceIdBefore = Compiled(_highestSequenceNrForPersistenceIdBefore _)
+ val messagesQuery = Compiled(_messagesQuery _)
+
def writeJournalRows(xs: Seq[(JournalPekkoSerializationRow, Set[String])])(implicit ec: ExecutionContext) = {
val sorted = xs.sortBy(event => event._1.sequenceNumber)
if (sorted.exists(_._2.nonEmpty)) {
@@ -50,9 +54,6 @@ class JournalQueries(
}
}
- private def selectAllJournalForPersistenceIdDesc(persistenceId: Rep[String]) =
- selectAllJournalForPersistenceId(persistenceId).sortBy(_.sequenceNumber.desc)
-
private def selectAllJournalForPersistenceId(persistenceId: Rep[String]) =
JournalTable.filter(_.persistenceId === persistenceId).sortBy(_.sequenceNumber.desc)
@@ -60,10 +61,10 @@ class JournalQueries(
JournalTable.filter(_.persistenceId === persistenceId).filter(_.sequenceNumber <= toSequenceNr).delete
}
- def markJournalMessagesAsDeleted(persistenceId: String, maxSequenceNr: Long) =
+ def markSeqNrJournalMessagesAsDeleted(persistenceId: String, sequenceNr: Long) =
JournalTable
.filter(_.persistenceId === persistenceId)
- .filter(_.sequenceNumber <= maxSequenceNr)
+ .filter(_.sequenceNumber === sequenceNr)
.filter(_.deleted === false)
.map(_.deleted)
.update(true)
@@ -71,28 +72,13 @@ class JournalQueries(
private def _highestSequenceNrForPersistenceId(persistenceId: Rep[String]): Rep[Option[Long]] =
selectAllJournalForPersistenceId(persistenceId).take(1).map(_.sequenceNumber).max
- private def _highestMarkedSequenceNrForPersistenceId(persistenceId: Rep[String]): Rep[Option[Long]] =
- selectAllJournalForPersistenceId(persistenceId).filter(_.deleted === true).take(1).map(_.sequenceNumber).max
-
- val highestSequenceNrForPersistenceId = Compiled(_highestSequenceNrForPersistenceId _)
-
- val highestMarkedSequenceNrForPersistenceId = Compiled(_highestMarkedSequenceNrForPersistenceId _)
-
- private def _selectByPersistenceIdAndMaxSequenceNumber(persistenceId: Rep[String], maxSequenceNr: Rep[Long]) =
- selectAllJournalForPersistenceIdDesc(persistenceId).filter(_.sequenceNumber <= maxSequenceNr)
-
- val selectByPersistenceIdAndMaxSequenceNumber = Compiled(_selectByPersistenceIdAndMaxSequenceNumber _)
-
- private def _allPersistenceIdsDistinct: Query[Rep[String], String, Seq] =
- JournalTable.map(_.persistenceId).distinct
-
- val allPersistenceIdsDistinct = Compiled(_allPersistenceIdsDistinct)
-
- def journalRowByPersistenceIds(persistenceIds: Iterable[String]): Query[Rep[String], String, Seq] =
- for {
- query <- JournalTable.map(_.persistenceId)
- if query.inSetBind(persistenceIds)
- } yield query
+ private def _highestSequenceNrForPersistenceIdBefore(
+ persistenceId: Rep[String],
+ maxSequenceNr: Rep[Long]): Query[Rep[Long], Long, Seq] =
+ selectAllJournalForPersistenceId(persistenceId)
+ .filter(_.sequenceNumber <= maxSequenceNr)
+ .map(_.sequenceNumber)
+ .take(1)
private def _messagesQuery(
persistenceId: Rep[String],
@@ -103,10 +89,8 @@ class JournalQueries(
.filter(_.persistenceId === persistenceId)
.filter(_.deleted === false)
.filter(_.sequenceNumber >= fromSequenceNr)
- .filter(_.sequenceNumber <= toSequenceNr)
+ .filter(_.sequenceNumber <= toSequenceNr) // TODO optimized this avoid large offset query.
.sortBy(_.sequenceNumber.asc)
.take(max)
- val messagesQuery = Compiled(_messagesQuery _)
-
}
diff --git a/core/src/main/scala/org/apache/pekko/persistence/jdbc/journal/dao/legacy/ByteArrayJournalDao.scala b/core/src/main/scala/org/apache/pekko/persistence/jdbc/journal/dao/legacy/ByteArrayJournalDao.scala
index d72bbfe..4eaf4fa 100644
--- a/core/src/main/scala/org/apache/pekko/persistence/jdbc/journal/dao/legacy/ByteArrayJournalDao.scala
+++ b/core/src/main/scala/org/apache/pekko/persistence/jdbc/journal/dao/legacy/ByteArrayJournalDao.scala
@@ -88,9 +88,10 @@ trait BaseByteArrayJournalDao
// We should keep journal record with highest sequence number in order to be compliant
// with @see [[pekko.persistence.journal.JournalSpec]]
val actions: DBIOAction[Unit, NoStream, Effect.Write with Effect.Read] = for {
- _ <- queries.markJournalMessagesAsDeleted(persistenceId, maxSequenceNr)
- highestMarkedSequenceNr <- highestMarkedSequenceNr(persistenceId)
- _ <- queries.delete(persistenceId, highestMarkedSequenceNr.getOrElse(0L) - 1)
+ seq <- queries.highestSequenceNrForPersistenceIdBefore((persistenceId, maxSequenceNr)).result
+ highestSequenceNr = seq.headOption.getOrElse(0L)
+ _ <- queries.delete(persistenceId, highestSequenceNr - 1)
+ _ <- queries.markSeqNrJournalMessagesAsDeleted(persistenceId, highestSequenceNr)
} yield ()
db.run(actions.transactionally)
@@ -108,9 +109,6 @@ trait BaseByteArrayJournalDao
db.run(queries.update(persistenceId, sequenceNr, serializedRow.message).map(_ => Done))
}
- private def highestMarkedSequenceNr(persistenceId: String) =
- queries.highestMarkedSequenceNrForPersistenceId(persistenceId).result
-
override def highestSequenceNr(persistenceId: String, fromSequenceNr: Long): Future[Long] =
for {
maybeHighestSeqNo <- db.run(queries.highestSequenceNrForPersistenceId(persistenceId).result)
diff --git a/core/src/main/scala/org/apache/pekko/persistence/jdbc/journal/dao/legacy/JournalQueries.scala b/core/src/main/scala/org/apache/pekko/persistence/jdbc/journal/dao/legacy/JournalQueries.scala
index 4fb7248..df2529f 100644
--- a/core/src/main/scala/org/apache/pekko/persistence/jdbc/journal/dao/legacy/JournalQueries.scala
+++ b/core/src/main/scala/org/apache/pekko/persistence/jdbc/journal/dao/legacy/JournalQueries.scala
@@ -24,12 +24,13 @@ class JournalQueries(val profile: JdbcProfile, override val journalTableCfg: Leg
private val JournalTableC = Compiled(JournalTable)
+ val highestSequenceNrForPersistenceId = Compiled(_highestSequenceNrForPersistenceId _)
+ val highestSequenceNrForPersistenceIdBefore = Compiled(_highestSequenceNrForPersistenceIdBefore _)
+ val messagesQuery = Compiled(_messagesQuery _)
+
def writeJournalRows(xs: Seq[JournalRow]) =
JournalTableC ++= xs.sortBy(_.sequenceNumber)
- private def selectAllJournalForPersistenceIdDesc(persistenceId: Rep[String]) =
- selectAllJournalForPersistenceId(persistenceId).sortBy(_.sequenceNumber.desc)
-
private def selectAllJournalForPersistenceId(persistenceId: Rep[String]) =
JournalTable.filter(_.persistenceId === persistenceId).sortBy(_.sequenceNumber.desc)
@@ -47,10 +48,10 @@ class JournalQueries(val profile: JdbcProfile, override val journalTableCfg: Leg
baseQuery.map(_.message).update(replacement)
}
- def markJournalMessagesAsDeleted(persistenceId: String, maxSequenceNr: Long) =
+ def markSeqNrJournalMessagesAsDeleted(persistenceId: String, sequenceNr: Long) =
JournalTable
.filter(_.persistenceId === persistenceId)
- .filter(_.sequenceNumber <= maxSequenceNr)
+ .filter(_.sequenceNumber === sequenceNr)
.filter(_.deleted === false)
.map(_.deleted)
.update(true)
@@ -58,28 +59,12 @@ class JournalQueries(val profile: JdbcProfile, override val journalTableCfg: Leg
private def _highestSequenceNrForPersistenceId(persistenceId: Rep[String]): Rep[Option[Long]] =
selectAllJournalForPersistenceId(persistenceId).take(1).map(_.sequenceNumber).max
- private def _highestMarkedSequenceNrForPersistenceId(persistenceId: Rep[String]): Rep[Option[Long]] =
- selectAllJournalForPersistenceId(persistenceId).filter(_.deleted === true).take(1).map(_.sequenceNumber).max
-
- val highestSequenceNrForPersistenceId = Compiled(_highestSequenceNrForPersistenceId _)
-
- val highestMarkedSequenceNrForPersistenceId = Compiled(_highestMarkedSequenceNrForPersistenceId _)
-
- private def _selectByPersistenceIdAndMaxSequenceNumber(persistenceId: Rep[String], maxSequenceNr: Rep[Long]) =
- selectAllJournalForPersistenceIdDesc(persistenceId).filter(_.sequenceNumber <= maxSequenceNr)
-
- val selectByPersistenceIdAndMaxSequenceNumber = Compiled(_selectByPersistenceIdAndMaxSequenceNumber _)
-
- private def _allPersistenceIdsDistinct: Query[Rep[String], String, Seq] =
- JournalTable.map(_.persistenceId).distinct
-
- val allPersistenceIdsDistinct = Compiled(_allPersistenceIdsDistinct)
-
- def journalRowByPersistenceIds(persistenceIds: Iterable[String]): Query[Rep[String], String, Seq] =
- for {
- query <- JournalTable.map(_.persistenceId)
- if query.inSetBind(persistenceIds)
- } yield query
+ private def _highestSequenceNrForPersistenceIdBefore(
+ persistenceId: Rep[String], maxSequenceNr: Rep[Long]): Query[Rep[Long], Long, Seq] =
+ selectAllJournalForPersistenceId(persistenceId)
+ .filter(_.sequenceNumber <= maxSequenceNr)
+ .map(_.sequenceNumber)
+ .take(1)
private def _messagesQuery(
persistenceId: Rep[String],
@@ -93,7 +78,4 @@ class JournalQueries(val profile: JdbcProfile, override val journalTableCfg: Leg
.filter(_.sequenceNumber <= toSequenceNr)
.sortBy(_.sequenceNumber.asc)
.take(max)
-
- val messagesQuery = Compiled(_messagesQuery _)
-
}
diff --git a/core/src/main/scala/org/apache/pekko/persistence/jdbc/query/dao/ReadJournalQueries.scala b/core/src/main/scala/org/apache/pekko/persistence/jdbc/query/dao/ReadJournalQueries.scala
index fa0af92..4d2884b 100644
--- a/core/src/main/scala/org/apache/pekko/persistence/jdbc/query/dao/ReadJournalQueries.scala
+++ b/core/src/main/scala/org/apache/pekko/persistence/jdbc/query/dao/ReadJournalQueries.scala
@@ -25,24 +25,24 @@ class ReadJournalQueries(val profile: JdbcProfile, val readJournalConfig: ReadJo
import profile.api._
- def journalRowByPersistenceIds(persistenceIds: Iterable[String]) =
- for {
- query <- JournalTable.map(_.persistenceId)
- if query.inSetBind(persistenceIds)
- } yield query
-
- private def _allPersistenceIdsDistinct(max: ConstColumn[Long]): Query[Rep[String], String, Seq] =
- baseTableQuery().map(_.persistenceId).distinct.take(max)
+ val allPersistenceIdsDistinct = Compiled(_allPersistenceIdsDistinct _)
+ val messagesQuery = Compiled(_messagesQuery _)
+ val eventsByTag = Compiled(_eventsByTag _)
+ val journalSequenceQuery = Compiled(_journalSequenceQuery _)
+ val maxJournalSequenceQuery = Compiled {
+ JournalTable.map(_.ordering).max.getOrElse(0L)
+ }
private def baseTableQuery() =
JournalTable.filter(_.deleted === false)
+ private def _allPersistenceIdsDistinct(max: ConstColumn[Long]): Query[Rep[String], String, Seq] =
+ baseTableQuery().map(_.persistenceId).distinct.take(max)
+
private def baseTableWithTagsQuery() = {
baseTableQuery().join(TagTable).on(_.ordering === _.eventId)
}
- val allPersistenceIdsDistinct = Compiled(_allPersistenceIdsDistinct _)
-
private def _messagesQuery(
persistenceId: Rep[String],
fromSequenceNr: Rep[Long],
@@ -52,12 +52,9 @@ class ReadJournalQueries(val profile: JdbcProfile, val readJournalConfig: ReadJo
.filter(_.persistenceId === persistenceId)
.filter(_.sequenceNumber >= fromSequenceNr)
.filter(_.sequenceNumber <= toSequenceNr)
- .filter(!_.deleted)
.sortBy(_.sequenceNumber.asc)
.take(max)
- val messagesQuery = Compiled(_messagesQuery _)
-
private def _eventsByTag(
tag: Rep[String],
offset: ConstColumn[Long],
@@ -71,15 +68,7 @@ class ReadJournalQueries(val profile: JdbcProfile, val readJournalConfig: ReadJo
.map(_._1)
}
- val eventsByTag = Compiled(_eventsByTag _)
-
private def _journalSequenceQuery(from: ConstColumn[Long], limit: ConstColumn[Long]) =
JournalTable.filter(_.ordering > from).map(_.ordering).sorted.take(limit)
- val journalSequenceQuery = Compiled(_journalSequenceQuery _)
-
- val maxJournalSequenceQuery = Compiled {
- JournalTable.map(_.ordering).max.getOrElse(0L)
- }
-
}
diff --git a/core/src/main/scala/org/apache/pekko/persistence/jdbc/query/dao/legacy/ReadJournalQueries.scala b/core/src/main/scala/org/apache/pekko/persistence/jdbc/query/dao/legacy/ReadJournalQueries.scala
index ace288a..7640678 100644
--- a/core/src/main/scala/org/apache/pekko/persistence/jdbc/query/dao/legacy/ReadJournalQueries.scala
+++ b/core/src/main/scala/org/apache/pekko/persistence/jdbc/query/dao/legacy/ReadJournalQueries.scala
@@ -24,19 +24,19 @@ class ReadJournalQueries(val profile: JdbcProfile, val readJournalConfig: ReadJo
import profile.api._
- def journalRowByPersistenceIds(persistenceIds: Iterable[String]) =
- for {
- query <- JournalTable.map(_.persistenceId)
- if query.inSetBind(persistenceIds)
- } yield query
-
- private def _allPersistenceIdsDistinct(max: ConstColumn[Long]): Query[Rep[String], String, Seq] =
- baseTableQuery().map(_.persistenceId).distinct.take(max)
+ val allPersistenceIdsDistinct = Compiled(_allPersistenceIdsDistinct _)
+ val messagesQuery = Compiled(_messagesQuery _)
+ val eventsByTag = Compiled(_eventsByTag _)
+ val journalSequenceQuery = Compiled(_journalSequenceQuery _)
+ val maxJournalSequenceQuery = Compiled {
+ JournalTable.map(_.ordering).max.getOrElse(0L)
+ }
private def baseTableQuery() =
JournalTable.filter(_.deleted === false)
- val allPersistenceIdsDistinct = Compiled(_allPersistenceIdsDistinct _)
+ private def _allPersistenceIdsDistinct(max: ConstColumn[Long]): Query[Rep[String], String, Seq] =
+ baseTableQuery().map(_.persistenceId).distinct.take(max)
private def _messagesQuery(
persistenceId: Rep[String],
@@ -46,12 +46,10 @@ class ReadJournalQueries(val profile: JdbcProfile, val readJournalConfig: ReadJo
baseTableQuery()
.filter(_.persistenceId === persistenceId)
.filter(_.sequenceNumber >= fromSequenceNr)
- .filter(_.sequenceNumber <= toSequenceNr)
+ .filter(_.sequenceNumber <= toSequenceNr) // TODO perf: optimized this avoid large offset query.
.sortBy(_.sequenceNumber.asc)
.take(max)
- val messagesQuery = Compiled(_messagesQuery _)
-
private def _eventsByTag(
tag: Rep[String],
offset: ConstColumn[Long],
@@ -64,14 +62,6 @@ class ReadJournalQueries(val profile: JdbcProfile, val readJournalConfig: ReadJo
.take(max)
}
- val eventsByTag = Compiled(_eventsByTag _)
-
private def _journalSequenceQuery(from: ConstColumn[Long], limit: ConstColumn[Long]) =
JournalTable.filter(_.ordering > from).map(_.ordering).sorted.take(limit)
-
- val journalSequenceQuery = Compiled(_journalSequenceQuery _)
-
- val maxJournalSequenceQuery = Compiled {
- JournalTable.map(_.ordering).max.getOrElse(0L)
- }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org