Stateless pipeline resuming

[chubei]

Stateless pipeline resuming

For stateless pipelines, we store the pipeline state in the sink.

The pipeline state is a collection of connection states.

A connection state consists of two parts:

• Connection level state. Example: Postgres replication slot name.
• Record level state. Example: Postgres LSN and sequence number. We call it the OpIdentifier. It's always two u64s.

Connection level state is stored in a separate table in the sink, and is written only once. Record level state is written as part of the record.

To be more specific, every record in the sink has a __dozer_record_state field which stores the OpIdentifier. We should be able to get the maximum OpIdentifier of a sink. (How a specific sink stores and aggregates the field is out of scope for this document.)

For inserts, the field is set to the record's OpIdentifier if it has one (during replication), or None if it doesn't (during snapshotting).

For deletes, the field is deleted with the record.

For updates, the field is updated to the new OpIdentifier of the update.

Upon restart, Dozer first aggregates the sink for the maximum OpIdentifier of the sink. If it's None, Dozer starts moving the whole source tables. Otherwise, Dozer passes the OpIdentifier and the connection level state to the connector to continue replication.

Why this is correct

If the last operation that was written to the sink is an insert or update, then the connector restarts from the correct position.

If the last one or more operations are deletes, the connector will send these deletes again, but the deletes will not modify the sink because these records are already deleted.

[karolisg]

How should connector behave when ingestion resumes from middle of source transaction?

Now, in every connector we need to have logic which drops those events before sending them to pipeline. Maybe, it would be better that in pipeline we drop those messages instead of doing that in every connector.

[chubei]

Yes, connectors need to drop certain events in the first transaction. This is necessary because some sinks are not transactional (like Aerospike). Writes to Aerospike can break at any point.

[karolisg]

I mean, maybe we can do that not in ingestion, but for example insink_node.on_op.

[chubei]

Sure we can do that

[chubei]

Actually it's necessary for the sink to do that. Imagine we're writing two postgres tables to two aerospike sets. One set may be ahead of the other, so postgres has to resume from the set that's behind, which means the set that's ahead must discard operations.

[karolisg]

Right, then we don't need to do that in connector, because it would be just implementation of the same logic in multiple places.