You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@pekko.apache.org by "pjfanning (via GitHub)" <gi...@apache.org> on 2026/10/24 20:10:29 UTC
[PR] core: snapshots as starting points for eventsBySlices and eventsByPersistenceId [pekko-persistence-r2dbc]
pjfanning opened a new pull request, #515:
URL: https://github.com/apache/pekko-persistence-r2dbc/pull/515
### Motivation
Port of the akka-persistence-r2dbc "snapshot as starting point" feature and its follow-ups:
- [akka/akka-persistence-r2dbc#410](https://github.com/akka/akka-persistence-r2dbc/pull/410) Snapshots as starting point for eventsBySlices
- [akka/akka-persistence-r2dbc#412](https://github.com/akka/akka-persistence-r2dbc/pull/412) Snapshot as starting point for eventsByPersistenceId
- [akka/akka-persistence-r2dbc#438](https://github.com/akka/akka-persistence-r2dbc/pull/438) Harden EventsBySliceStartingFromSnapshotSpec
- [akka/akka-persistence-r2dbc#464](https://github.com/akka/akka-persistence-r2dbc/pull/464) Envelope source is different for durable state
- [akka/akka-persistence-r2dbc#469](https://github.com/akka/akka-persistence-r2dbc/pull/469) Mark snapshot envelopes with snapshot source
The code is copied from the akka-persistence-r2dbc 1.2.0 release, which is available under the Apache License, Version 2.0 as of 2026-10-24 (its LICENSE change date). New files copied from Akka (`StartingFromSnapshotStage` and three specs) keep the Lightbend copyright line, with the ASF header added by `sbt headerCreate`.
Pekko core 2.0.0-M4 already defines `EventsBySliceStartingFromSnapshotsQuery`, `EventsByPersistenceIdStartingFromSnapshotQuery` and their current variants, but the R2DBC read journal doesn't implement them.
### Modification
- **Queries:** `eventsBySlicesStartingFromSnapshots`, `currentEventsBySlicesStartingFromSnapshots`, `eventsByPersistenceIdStartingFromSnapshot` and `currentEventsByPersistenceIdStartingFromSnapshot`, in both the scaladsl and javadsl read journals.
- **`SnapshotDao`** implements `BySliceQuery.Dao`, with dialect hooks for Postgres, Yugabyte and MySQL (slice condition, bucket SQL, tag encoding, current timestamp). When enabled, it stores the `db_timestamp` and tags of the corresponding event.
- **Config (differs from upstream):** Pekko settings are per plugin, so:
- `query.start-from-snapshot.enabled` (default off), plus `query.snapshot-table` and `query.snapshot-payload-column-type`.
- `snapshot.start-from-snapshot.enabled` defaults to the query setting; `snapshot.journal-table` and `snapshot.buffer-size` are also new.
- The read journal derives its `SnapshotSettings` from the query settings.
- The snapshot plugin reads the event's timestamp and tags from the journal table itself, since it has no `QueryDao`.
- When disabled, the snapshot plugin never uses the new columns.
- **#464 / #469:** rows carry their own `source`, and snapshot envelopes have source `SN` (`EnvelopeOrigin.fromSnapshot`).
- **Migration tool (Pekko addition):** when enabled, migrated snapshots get the timestamp and tags of the already-migrated event. Upstream left this to a manual `UPDATE`.
- **Schema:** the `db_timestamp` and `tags` columns and `snapshot_slice_idx` are added to all `create_tables_*.sql` scripts. New migration scripts are in `ddl-scripts/migration/` for Postgres (covers BYTEA and JSONB), Yugabyte and MySQL.
- **Small fixes on top of upstream:** the snapshot by-slice query also bounds by `toTimestamp` when given, and the primary sub-sink in `StartingFromSnapshotStage` is now named `primary`.
- **Docs:** query sections with Java and Scala snippets, a snapshot store note, and a migration guide section with a rolling update procedure.
### Result
Projections and other consumers can start from snapshots instead of replaying all events. Existing deployments are unaffected until they enable the setting and migrate the schema.
### Tests
- New specs:
- `StartingFromSnapshotStageSpec`, `EventsBySliceStartingFromSnapshotSpec`, `EventsByPersistenceIdStartingFromSnapshotSpec`: ported from upstream, plus javadsl and `SN` source tests.
- `StartFromSnapshotDisabledSpec`: queries are rejected when the feature is disabled; snapshots save and load in a table with the **old** schema. That test fails with `column "db_timestamp" does not exist` if the new columns are written while disabled.
- `MigrationToolSpec`: snapshot timestamp equals the event's. It fails without the tool change.
- PostgreSQL (docker): `sbt core/test` 391 succeeded, 0 failed; `sbt migration/test` 8 succeeded.
- MySQL (docker): the snapshot-related specs pass (44 succeeded). `DurableStateBySliceSpec` fails locally on `upstream/main` too (it passes in CI), so it's environmental and unrelated.
- Yugabyte: not run locally; relying on CI.
- Migration scripts: applied to a database with the old schema on PostgreSQL and MySQL; existing snapshots got the event's timestamp and tags, and the index was created.
- `sbt core/mimaReportBinaryIssues migration/mimaReportBinaryIssues`: no issues. The changed internals are `private[r2dbc]`; MiMa was confirmed to flag a deliberate public removal.
- `sbt docs/Test/compile docs/paradox`; `paradoxValidateInternalLinks` reports only the existing `[[R2dbcReadJournal.loadEnvelope]]` error, which is also on `main`.
- scalafmt, `javafmtAll`, `headerCheckAll`, `git diff --check`.
- The new `.sql` migration scripts have no license header: sbt doesn't manage `.sql`, and the Postgres create scripts have none either.
### References
Refs akka/akka-persistence-r2dbc#410, akka/akka-persistence-r2dbc#412, akka/akka-persistence-r2dbc#438, akka/akka-persistence-r2dbc#464, akka/akka-persistence-r2dbc#469
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] core: snapshots as starting points for eventsBySlices and eventsByPersistenceId [pekko-persistence-r2dbc]
Posted by "ptrdom (via GitHub)" <gi...@apache.org>.
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] core: snapshots as starting points for eventsBySlices and eventsByPersistenceId [pekko-persistence-r2dbc]
Posted by "ptrdom (via GitHub)" <gi...@apache.org>.
ptrdom commented on PR #515:
URL: https://github.com/apache/pekko-persistence-r2dbc/pull/515#issuecomment-6519972362
Please let me know when I can review this again - I assume that since you have not re-requested review it means you are still working on it. Just double checking if we are on the same process.
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] core: snapshots as starting points for eventsBySlices and eventsByPersistenceId [pekko-persistence-r2dbc]
Posted by "ptrdom (via GitHub)" <gi...@apache.org>.
ptrdom commented on code in PR #515:
URL: https://github.com/apache/pekko-persistence-r2dbc/pull/515#discussion_r4422897906
##########
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:
Following up on my own suggestion here: filtering `deleted = false` has a side effect I missed. `deleteMessages(toSeqNr)` replaces the row at `toSeqNr` with a delete marker (`deleted = true`). So with the common "snapshot, then `deleteMessages(snapshotSeqNr)`" pattern, the update no longer matches and the snapshot's `db_timestamp` stays NULL. The by-slice snapshot query filters on `db_timestamp >= ?`, so that snapshot is never emitted. Its events are deleted as well, so `eventsBySlicesStartingFromSnapshots` misses the entity completely.
I reproduced it on PostgreSQL. I created the old schema and wrote two entities with the feature disabled: one that saves a snapshot at seq 2 and then calls `deleteMessages(2)`, and one that only saves a snapshot. After running the migration script with the feature enabled, `currentEventsBySlicesStartingFromSnapshots` emitted only the second entity's snapshot. The first entity's snapshot had `db_timestamp` NULL.
I still think the `deleted = false` filter is right, since a delete marker isn't the snapshot's event. Its timestamp is the time of the delete and it has no tags. Dropping the filter does make this case work, with the marker's timestamp. But it doesn't help when events were deleted past the snapshot's seqNr. I ran that case too: snapshot at seq 2, then `deleteMessages(3)`. The marker is at seq 3, the join matches nothing, and the snapshot stays NULL and is missed with or without the filter.
`saveAsync`, the `MigrationTool` and `collectSerializedSnapshot` all fall back to `write_timestamp` when the event isn't found. Could the scripts do the same after the join update? For example:
```sql
-- snapshots whose event is deleted or missing, same fallback as the snapshot plugin
UPDATE snapshot SET db_timestamp = to_timestamp(write_timestamp / 1000.0) WHERE db_timestamp IS NULL;
```
With this added, the snapshot was emitted in both experiments. MySQL could use `FROM_UNIXTIME(write_timestamp / 1000)`. Then the migration guide should say that snapshots without a matching event get the snapshot's write timestamp and no tags. The same applies to the Yugabyte and MySQL scripts.
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] core: snapshots as starting points for eventsBySlices and eventsByPersistenceId [pekko-persistence-r2dbc]
Posted by "pjfanning (via GitHub)" <gi...@apache.org>.
pjfanning commented on PR #515:
URL: https://github.com/apache/pekko-persistence-r2dbc/pull/515#issuecomment-6520318754
@ptrdom I think this is ready to review again
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] core: snapshots as starting points for eventsBySlices and eventsByPersistenceId [pekko-persistence-r2dbc]
Posted by "ptrdom (via GitHub)" <gi...@apache.org>.
ptrdom commented on code in PR #515:
URL: https://github.com/apache/pekko-persistence-r2dbc/pull/515#discussion_r4423962130
##########
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:
Sorry, the new spec broke CI on Postgres and Yugabyte, though not directly: `CurrentPersistenceIdsQuerySpec` "retrieve all ids" fails. The spec expects the ids in creation order, but the query returns them sorted by persistence id, and the entity types come from the `TestData` counter that all specs share. The one extra `nextEntityType()` call moved its entity types to `TestEntity-96` to `TestEntity-100`, and `TestEntity-100` sorts first. It isn't caused by this PR as such: on `main` the spec already fails when run alone (its entity types are then `TestEntity-6` to `TestEntity-10`), and the durable state `CurrentPersistenceIdsQuerySpec` has the same problem.
Fix for both specs, on the same branch: https://github.com/ptrdom/pekko-persistence-r2dbc/commit/0780dea540af7a44d7365a50013a08d5530099d8 (`git cherry-pick 0780dea`, it applies on top of 98b2cc6). The expected ids are sorted like the query results, and the "entity type after id" tests take the after id from the ids of that entity type.
- Both specs failed before the fix in a setup that crosses a digit boundary, and pass after.
- `sbt core/test` on PostgreSQL and Yugabyte: 452 succeeded, 0 failed.
- Both specs and `StartFromSnapshotMigrationScriptSpec` pass on MySQL, MariaDB and H2.
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] core: snapshots as starting points for eventsBySlices and eventsByPersistenceId [pekko-persistence-r2dbc]
Posted by "pjfanning (via GitHub)" <gi...@apache.org>.
pjfanning commented on code in PR #515:
URL: https://github.com/apache/pekko-persistence-r2dbc/pull/515#discussion_r4423459486
##########
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:
Thanks, cherry-picked as fea472e (plus a scalafmt line wrap in the new spec). The fallback matches what `saveAsync` and the `MigrationTool` do, and the spec covers both delete cases. On H2 it's cancelled as expected; CI will run it on the other databases.
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] core: snapshots as starting points for eventsBySlices and eventsByPersistenceId [pekko-persistence-r2dbc]
Posted by "pjfanning (via GitHub)" <gi...@apache.org>.
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] core: snapshots as starting points for eventsBySlices and eventsByPersistenceId [pekko-persistence-r2dbc]
Posted by "ptrdom (via GitHub)" <gi...@apache.org>.
ptrdom commented on code in PR #515:
URL: https://github.com/apache/pekko-persistence-r2dbc/pull/515#discussion_r4423051844
##########
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:
I put the fallback and a test for the migration scripts on a branch, so you can cherry-pick it if it looks right: https://github.com/ptrdom/pekko-persistence-r2dbc/commit/7a9b6e3761ee76a0c177314c0f492a931563e69f (`git cherry-pick 7a9b6e3` after fetching `snapshot-migration-script-spec` from my fork).
- The Postgres, Yugabyte and MySQL scripts set `db_timestamp` from `write_timestamp` for the snapshots that are still NULL after the join. For Postgres and Yugabyte I used `TIMESTAMP WITH TIME ZONE 'epoch' + write_timestamp * INTERVAL '1 millisecond'` instead of the `to_timestamp(... / 1000.0)` I suggested above, because it's exact to the millisecond. MySQL uses `FROM_UNIXTIME(write_timestamp / 1000)`.
- The migration guide shows the second update and explains it.
- New `StartFromSnapshotMigrationScriptSpec`, since the scripts aren't run anywhere in CI. It creates a legacy snapshot table and writes three entities with the snapshot plugin disabled: one whose event exists (with a tag), one after `deleteEventsTo(snapshotSeqNr)` and one after deleting past the snapshot. It then runs the dialect's script from `ddl-scripts/migration/` on that table and checks that `currentEventsBySlicesStartingFromSnapshots` emits all three. The first must have the event's timestamp and tags; the other two must have the write timestamp and no tags. It's cancelled on H2, which has no script.
It passes locally on PostgreSQL, MySQL, MariaDB and Yugabyte. Without the script change it fails on PostgreSQL and MySQL, and only the snapshot whose event exists is emitted.
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] core: snapshots as starting points for eventsBySlices and eventsByPersistenceId [pekko-persistence-r2dbc]
Posted by "pjfanning (via GitHub) via notifications" <no...@pekko.apache.org>.
pjfanning merged PR #515:
URL: https://github.com/apache/pekko-persistence-r2dbc/pull/515
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] core: snapshots as starting points for eventsBySlices and eventsByPersistenceId [pekko-persistence-r2dbc]
Posted by "pjfanning (via GitHub)" <gi...@apache.org>.
pjfanning commented on code in PR #515:
URL: https://github.com/apache/pekko-persistence-r2dbc/pull/515#discussion_r4424451324
##########
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:
Thanks. I hit the same failure and pushed an equivalent fix as 0d343a2 before I saw this: both specs sort the expected ids, and the "entity type after id" tests take the after id from that entity type's ids. It differs from 0780dea only in naming and the comment. Checked with the `TestData` counter starting at 95: both specs failed before and pass after.
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org