This is a problem I would like to solve with a single query, as I am working on a large dataset, and the process as it is can take up to 2 hours to complete, and additional queries would create further bottlenecks.

This is roughly the Model I am working with

class Event(ModelBase):
 __tablename__ = "events"

 id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
 created: Mapped[Optional[datetime]] = mapped_column(DateTime, nullable=True, default=datetime.utcnow)
 global_sequence_number: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
 aggregate_uuid: Mapped[uuid.UUID] = mapped_column(
 UUID(as_uuid=True), nullable=False, index=True, default=uuid.uuid4
 )
 event_type: Mapped[str] = mapped_column(String, nullable=False)
 event_version: Mapped[int] = mapped_column(Integer, nullable=False)
 payload: Mapped[dict[str, Any]] = mapped_column(JSONB, nullable=False)
 unique_processing_hash: Mapped[Optional[str]] = mapped_column(String, nullable=True, unique=True)
 tenant_id: Mapped[Optional[int]] = mapped_column(BigInteger, nullable=True)

A scheduled task receives a partition of aggregates, such as `partition: list[UUID]`.

For each one a dictionary is generated with all events belonging to an aggregate, sorted as in

stmt = (
 select(Event.aggregate_uuid, Event.event_type, Event.payload)
 .filter(Event.tenant_id == self.tenant_id)
 .filter(Event.aggregate_uuid.in_(aggregate_uuids))
 .order_by(Event.aggregate_uuid, Event.global_sequence_number)
 )
events_by_aggregate = defaultdict(list)
 for row in self.session.execute(stmt):
 try:
 event_class = self._translate_to_event_class_from_raw(row.event_type, row.payload)
 if event_class:
 events_by_aggregate[row.aggregate_uuid].append(event_class)

Now, I am introducing a new event type, that will serve as a "snapshot". The snapshot will aggregate all data from previous events. Therefore, there can be two scenarios:

aggregate that does not have any snapshots: we should process ALL events from the very first one
- [ e1, e2, e3 ] => return all
aggregate that does have a snapshot: only process the latest snapshot + subsequent events
- [ e1, e2, s, e3 ] => return s + e3

The second scenario is up and running. The following filter is applied to the above query, and it always returns the lates snapshot + subsequent events

if start_from_snapshot:
 latest_snapshot = (
 select(
 Event.aggregate_uuid.label("aggregate"), 
 func.max(Event.created).label("latest")
 )
 .filter(Event.aggregate_uuid.in_(aggregate_uuids))
 .filter(Event.event_type == "CaseSnapshot")
 .group_by(Event.aggregate_uuid)
 .subquery()
 )
 stmt = stmt.where(
 Event.aggregate_uuid == latest_snapshot.c.aggregate, 
 latest_snapshot.c.latest <= Event.created
 )

What is not working is the first scenario, where I would expect all events to be selected.

What I have tried is to use logical conditions in the same filter, such as

stmt = stmt.filter(
 or_(
 and_(
 # aggregate has a snapshot
 Event.aggregate_uuid == latest_snapshot.c.aggregate,
 # filter by timestamp from latest snapshot onwards 
 latest_snapshot.c.latest <= Event.created
 ),
 # aggregate has no snapshot, allow all. THIS DOES NOT WORK
 not_(Event.aggregate_uuid == latest_snapshot.c.aggregate)
 )
 )

but the query does not select a single row. I believe the above filter is getting closer to the solution, but I am not sure how to make it work properly.

Stackoverflow

Message History

Delivery option(s)