ptrdom commented on code in PR #515:
URL:
https://github.com/apache/pekko-persistence-r2dbc/pull/515#discussion_r4375627011
##########
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:
Same issue as above for the live query: a fully filtered page means `latest`
doesn't advance, and the query never gets past that page.
##########
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:
The filter runs before `deserializeAndAddOffset`, which is what moves
`state.latest` and `rowCount` forward (via `nextOffset`). If every row of a
page is dropped, the page looks empty:
- here in `currentBySlices`, `nextQuery` stops on `rowCount == 0`, so the
query completes without the events after the snapshots;
- in `liveBySlices` (line 470) the offset doesn't move, so it keeps
re-reading the same page and never gets past it.
Repro with `pekko.persistence.r2dbc.query.buffer-size = 10`: entity A
persists `a-1-snap` (snapshot at seq 1, which becomes the init offset), then
entity B persists 33 events with a snapshot at seq 30.
`currentEventsBySlicesStartingFromSnapshots(entityType, 0, 1023, NoOffset, s =>
s)` emits the 2 snapshots and then completes without `b-31..b-33`; the live
variant emits the 2 snapshots and nothing more within 10 seconds. With
`buffer-size = 100` the same test passes.
With the default buffer size of 1000 it's enough that 1000 consecutive rows
are before their entities' snapshots, which is easy to hit with many entities.
The dropped rows still need to update the offset and row count, e.g. by
applying the filter after the state update (on the envelope), or by letting
`deserializeAndAddOffset` skip emitting them while still updating the state.
##########
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:
Minor: these defaults (and `start-from-snapshot.enabled` above) refer to the
default plugin paths, and `journalTableWithSchema` uses the snapshot plugin's
`schema`. With custom plugin ids or a journal in another schema, the snapshot
plugin silently stores `write_timestamp` and empty tags, because
`eventTimestampAndTags` just returns `None`. A note in the docs that these must
be set for custom plugin setups, and maybe a warning log in `saveAsync` when
enabled and the event isn't found, would make that easier to notice.
##########
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:
Minor: `eventTimestampAndTags` filters `deleted = false`, but the migration
`UPDATE`s (here, Yugabyte and MySQL) don't. Harmless in practice, but better to
be consistent. For large snapshot tables, a sentence in the migration guide
about running the update in batches could also help.
##########
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:
With a Zero offset the event query starts at the min snapshot timestamp, so
events of entities that have no snapshot, and that are older than that
timestamp, are skipped. The same risk the `else` comment describes for a
non-zero offset applies here too.
Repro: entity C persists `c-1` (no snapshot), then entity A persists
`a-1-snap`. `currentEventsBySlicesStartingFromSnapshots(entityType, 0, 1023,
NoOffset, s => s)` emits `a-1-snap` and completes without `c-1`.
The same gap applies to an entity whose snapshot row is moved past
`toTimestamp` by a newer snapshot while the snapshot query runs: it's then not
in `snapshotOffsets`, and its events before the min timestamp are skipped.
Either start from the given offset (correct once the filter issue above is
fixed, at the cost of scanning more events), or document that all entities of
the entity type must have a snapshot, and add a test either way. The docs
currently don't mention this.
--
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]
