Persist and restore pipeline state

[chubei]

Persist and restore pipeline state

Pipeline state includes two parts:

• The record store
• Individual processor state

Record store

Record store will be persisted in an incremental fashion. To enable this, we require the record store to be append only.

The record store is essentially a Arc<RwLock<Vec<Record>>. We serialize a slice of the Vec on every commit and upload to S3.

During restore, we download all the slices and concatenate them.

Processor state

Processor state can be arbitrary data and is usually not append only, so we persist it fully on every commit.

Every processor is responsible for serializing and deserializing its own state.

Persisting order

Because the record store is shared between all processors, and any processor can append records to the store, we need to make sure that the record store is persisted after all processors.

To achieve this, we'll add a finalize_epoch method to the EpochManager. All sinks will call this method on every commit. The EpochManager will persist the record store after all sinks have called finalize_epoch.

It doesn't matter if records belonging to the next epoch are appended before epoch is finalized, because it won't harm if the record store has more records than necessary.

Organizing checkpoints

The checkpoints of a pipeline looks like following:

record_store
    epoch_0.slice
    epoch_1.slice
epoch_0
    processor_0.bin
    processor_1.bin
epoch_1
    processor_0.bin
    processor_1.bin

Because we always persist processor states before the record store, all .bin files must exist if the corresponding .slice file exists.

Additional tweaking

Persisting doesn't need to happen on every commit. We can let EpochManager decide if persisting should happen during this commit by checking the record store size and the duration since last persisting.

@mediuminvader @snork-alt @v3g42

[snork-alt]

I'm assuming that when record store is loaded, records need to be looked up by a pk or something. How does this work ? Since record store is append only, it will need compaction at some point and we will have to store every operation. Is my assumption right ?

[chubei]

I'm thinking we change Arc<Record> to usize in the processors, so yes records need to be looked up, and can be done using index in the Vec<Recrod>.

Or we can keep the Arc<Record> in processors. The record store will then be Vec<Arc<Record>>. The record store can build a HashMap<*const Record, usize> and pass it to Processor::serialize to help the processor convert Arc<Recrod> to usize and serialize the usize. In this case, during restore, we convert usize back to Arc<Record>.

About compaction, one way to do it is to change Vec<Record> to Vec<Option<Record>> and run another process to remove the records that are not referenced. The index has to be stable because it's how processors reference records in this design.

[snork-alt]

Make sense..I was thinking the usize could actually be the offset on disk of the persisted record for faster lookup. Because we will have multiple slices how do we know in which slice to look for to find a specific record Id? Bloom filter ? Here each slice is acting like an immutable memtable ?

[chubei]

I don't get it. I think we're sticking with in memory record store, but you're talking like we're going to switch to a disk based record store.