pjfanning commented on code in PR #515:
URL:
https://github.com/apache/pekko-persistence-r2dbc/pull/515#discussion_r4377812556
##########
core/src/main/scala/org/apache/pekko/persistence/r2dbc/internal/BySliceQuery.scala:
##########
@@ -257,6 +259,7 @@ import org.slf4j.Logger
toTimestamp = Some(toTimestamp),
behindCurrentTime = Duration.Zero,
backtracking = false)
+ .filter(row => filterEventsBeforeSnapshots(row.persistenceId,
row.seqNr, row.source))
Review Comment:
Fixed in fbbc33d: the filter now runs after `ContinuousQuery`, on the
envelopes, so dropped rows still move `latest` and `rowCount` forward.
`EventsBySliceStartingFromSnapshotSmallBufferSpec` is your repro with
`buffer-size = 10`, for both the current and the live query. Both tests fail
without the fix. One cost: the dropped events are now deserialized before they
are dropped. I put that in #528.
##########
core/src/main/scala/org/apache/pekko/persistence/r2dbc/internal/BySliceQuery.scala:
##########
@@ -463,6 +467,7 @@ import org.slf4j.Logger
toTimestamp,
behindCurrentTime,
backtracking = newState.backtracking)
+ .filter(row => filterEventsBeforeSnapshots(row.persistenceId,
row.seqNr, row.source))
Review Comment:
Fixed by the same change, covered by the live test in
`EventsBySliceStartingFromSnapshotSmallBufferSpec`. Without the fix, the live
query read the same page again and emitted the snapshot's event after the
snapshot.
##########
core/src/main/scala/org/apache/pekko/persistence/r2dbc/query/scaladsl/R2dbcReadJournal.scala:
##########
@@ -169,37 +216,203 @@ final class R2dbcReadJournal(system:
ExtendedActorSystem, config: Config, cfgPat
offset: Offset): Source[EventEnvelope[Event], NotUsed] = {
val dbSource = bySlice[Event].liveBySlices("eventsBySlices", entityType,
minSlice, maxSlice, offset)
if (settings.journalPublishEvents) {
- val pubSub = PubSub(typedSystem)
- val pubSubSource =
- Source
- .actorRef[EventEnvelope[Event]](
- completionMatcher = PartialFunction.empty,
- failureMatcher = PartialFunction.empty,
- bufferSize = settings.bufferSize,
- // dropHead drops the oldest buffered event when full; any dropped
pub/sub events are
- // recovered from the database source which is merged below, and
deduplication handles
- // any events received via both paths.
- // OverflowStrategy.dropNew is long deprecated and removed in
Pekko 2.0.0.
- overflowStrategy = OverflowStrategy.dropHead)
- .mapMaterializedValue { ref =>
- pubSub.eventTopics[Event](entityType, minSlice, maxSlice).foreach
{ topic =>
- import pekko.actor.typed.scaladsl.adapter._
- topic ! Topic.Subscribe(ref.toTyped[EventEnvelope[Event]])
- }
- }
- .filter { env =>
- val slice = sliceForPersistenceId(env.persistenceId)
- minSlice <= slice && slice <= maxSlice
- }
- dbSource
- .mergePrioritized(pubSubSource, leftPriority = 1, rightPriority = 10)
- .via(skipPubSubTooFarAhead(settings.backtrackingEnabled,
- settings.backtrackingWindow.toJava))
- .via(deduplicate(settings.deduplicateCapacity))
+ val pubSubSource = eventsBySlicesPubSubSource[Event](entityType,
minSlice, maxSlice)
+ mergeDbAndPubSubSources(dbSource, pubSubSource)
} else
dbSource
}
+ /**
+ * Same as `currentEventsBySlices` but with the purpose to use snapshots as
starting points and thereby reducing
+ * number of events that have to be loaded. This can be useful if the
consumer start from zero without any previously
+ * processed offset or if it has been disconnected for a long while and its
offset is far behind.
+ *
+ * First it loads all snapshots with timestamps greater than or equal to the
offset timestamp. There is at most one
+ * snapshot per persistenceId. The snapshots are transformed to events with
the given `transformSnapshot` function.
+ *
+ * After emitting the snapshot events the ordinary events with sequence
numbers after the snapshots are emitted.
+ *
+ * To use `currentEventsBySlicesStartingFromSnapshots` you must enable
configuration
+ * `pekko.persistence.r2dbc.query.start-from-snapshot.enabled` and follow
the instructions in the migration guide.
+ */
+ override def currentEventsBySlicesStartingFromSnapshots[Snapshot, Event](
+ entityType: String,
+ minSlice: Int,
+ maxSlice: Int,
+ offset: Offset,
+ transformSnapshot: Snapshot => Event): Source[EventEnvelope[Event],
NotUsed] = {
+ checkStartFromSnapshotEnabled("currentEventsBySlicesStartingFromSnapshots")
+ val timestampOffset = toTimestampOffset(offset)
+
+ val snapshotSource =
+ snapshotsBySlice[Snapshot, Event](transformSnapshot)
+ .currentBySlices("currentSnapshotsBySlices", entityType, minSlice,
maxSlice, offset)
+
+ Source.fromGraph(
+ new StartingFromSnapshotStage[Event](
+ snapshotSource,
+ { snapshotOffsets =>
+ val initOffset = initOffsetAfterSnapshots(timestampOffset, offset,
snapshotOffsets)
+
+ log.debug(
+ "currentEventsBySlicesStartingFromSnapshots initOffset [{}] with
[{}] snapshots",
+ initOffset,
+ snapshotOffsets.size)
+
+ bySlice.currentBySlices(
+ "currentEventsBySlices",
+ entityType,
+ minSlice,
+ maxSlice,
+ initOffset,
+ filterEventsBeforeSnapshots(snapshotOffsets, backtrackingEnabled =
false))
+ }))
+ }
+
+ /**
+ * Same as `eventsBySlices` but with the purpose to use snapshots as
starting points and thereby reducing number of
+ * events that have to be loaded. This can be useful if the consumer start
from zero without any previously processed
+ * offset or if it has been disconnected for a long while and its offset is
far behind.
+ *
+ * First it loads all snapshots with timestamps greater than or equal to the
offset timestamp. There is at most one
+ * snapshot per persistenceId. The snapshots are transformed to events with
the given `transformSnapshot` function.
+ *
+ * After emitting the snapshot events the ordinary events with sequence
numbers after the snapshots are emitted.
+ *
+ * To use `eventsBySlicesStartingFromSnapshots` you must enable configuration
+ * `pekko.persistence.r2dbc.query.start-from-snapshot.enabled` and follow
the instructions in the migration guide.
+ */
+ override def eventsBySlicesStartingFromSnapshots[Snapshot, Event](
+ entityType: String,
+ minSlice: Int,
+ maxSlice: Int,
+ offset: Offset,
+ transformSnapshot: Snapshot => Event): Source[EventEnvelope[Event],
NotUsed] = {
+ checkStartFromSnapshotEnabled("eventsBySlicesStartingFromSnapshots")
+ val timestampOffset = toTimestampOffset(offset)
+
+ val snapshotSource =
+ snapshotsBySlice[Snapshot, Event](transformSnapshot)
+ .currentBySlices("snapshotsBySlices", entityType, minSlice, maxSlice,
offset)
+
+ Source.fromGraph(
+ new StartingFromSnapshotStage[Event](
+ snapshotSource,
+ { snapshotOffsets =>
+ val initOffset = initOffsetAfterSnapshots(timestampOffset, offset,
snapshotOffsets)
+
+ log.debug(
+ "eventsBySlicesStartingFromSnapshots initOffset [{}] with [{}]
snapshots",
+ initOffset,
+ snapshotOffsets.size)
+
+ val dbSource =
+ bySlice[Event].liveBySlices(
+ "eventsBySlices",
+ entityType,
+ minSlice,
+ maxSlice,
+ initOffset,
+ filterEventsBeforeSnapshots(snapshotOffsets,
settings.backtrackingEnabled))
+
+ if (settings.journalPublishEvents) {
+ // Note that events via PubSub are not filtered by
snapshotOffsets. It's unlikely that
+ // those will be earlier than the snapshots and duplicates must be
handled downstream in that case.
+ // If we would use the filterEventsBeforeSnapshots function for
PubSub it would be difficult to
+ // know when memory of that Map can be released or the filter
function would have to be shared
+ // and thread safe, which is not worth it.
+ val pubSubSource = eventsBySlicesPubSubSource[Event](entityType,
minSlice, maxSlice)
+ mergeDbAndPubSubSources(dbSource, pubSubSource)
+ } else
+ dbSource
+ }))
+ }
+
+ private def initOffsetAfterSnapshots(
+ timestampOffset: TimestampOffset,
+ offset: Offset,
+ snapshotOffsets: Map[String, (Long, Instant)]): Offset =
+ if (timestampOffset == TimestampOffset.Zero && snapshotOffsets.nonEmpty) {
Review Comment:
Fixed in fbbc33d: the event query now always starts at the given offset.
That also covers a snapshot that moved past `toTimestamp` while the snapshot
query was running. New test: `include older events of entities without a
snapshot for NoOffset` (current and live). The docs now say that all events are
still read from the offset. Akka found the same problem and fixed it in 1.3.12
(akka/akka-persistence-r2dbc#717), but that code is under the BSL until
2029-02-06, so this fix is our own. #717 also fixes a related problem:
snapshots are emitted before events that have earlier timestamps, so the
offsets go backwards. A projection that restarts from a saved offset can miss
events. I documented this for now and opened #528 for a clean-room redesign.
##########
core/src/main/resources/reference.conf:
##########
@@ -131,6 +131,16 @@ pekko.persistence.r2dbc {
use-connection-factory = ${pekko.persistence.r2dbc.use-connection-factory}
log-db-calls-exceeding = ${pekko.persistence.r2dbc.log-db-calls-exceeding}
+
+ # When enabled, the timestamp and tags of the corresponding event are
stored with the snapshot.
+ # Required by the queries that use snapshots as starting points, see
query.start-from-snapshot.
+ start-from-snapshot.enabled =
${pekko.persistence.r2dbc.query.start-from-snapshot.enabled}
+
+ # the journal table that the timestamp and tags of the corresponding event
are read from
+ journal-table = ${pekko.persistence.r2dbc.journal.table}
Review Comment:
Done: `snapshots.md` now lists the snapshot plugin settings that default to
the default plugin paths and the journal `schema`. When start-from-snapshot is
enabled and the event isn't found, `eventTimestampAndTags` logs a warning that
names the journal table.
##########
ddl-scripts/migration/snapshot_start_from_snapshot_postgres.sql:
##########
@@ -0,0 +1,13 @@
+-- Adds the snapshot columns and index that are needed by the queries that use
snapshots as starting points
+-- (eventsBySlicesStartingFromSnapshots,
eventsByPersistenceIdStartingFromSnapshot and their current variants).
+-- Use this for both the BYTEA and the JSONB schema. See the migration guide
for a rolling update procedure.
+
+ALTER TABLE snapshot ADD COLUMN IF NOT EXISTS db_timestamp timestamp with time
zone;
+ALTER TABLE snapshot ADD COLUMN IF NOT EXISTS tags TEXT ARRAY;
+
+-- populate the new columns from the corresponding events
+UPDATE snapshot s SET db_timestamp = e.db_timestamp, tags = e.tags
+ FROM event_journal e
+ WHERE s.persistence_id = e.persistence_id AND s.seq_nr = e.seq_nr;
Review Comment:
Done: the Postgres, Yugabyte and MySQL updates now filter `deleted = false`.
I checked the Postgres script against a deleted event. The migration guide now
suggests running the update in batches by `slice` range for large tables.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
