Higher-level properties

[lalithsuresh]

• I echo Mihai's comment about describing the interfaces we need, the guarantees those interfaces should provide, and possible choices for implementations that can support that interface. Right now, Kafka and its terminology seems to permeate the algorithm, the exchange operators etc. I'm not very familiar with the dbsp internals, so it's a bit hard for me to follow the design.
• Let me try and summarize my high-level understanding of the model and the invariants you're trying to preserve. Happy to discuss:
  • H inputs. K outputs. 1 circuit. W workers. All workers evaluate the same circuit in a data-parallel manner. Producers send events to inputs. Outputs produce events to consumers. A circuit step processes one or more input events at a time.
  • There is only one top level property we want: exactly-once semantics for receiving an input event and producing an output event from a circuit (distributed or not).
  • Achieving this property while ignoring circuit distribution requires us to:
    • durably log all H inputs.
    • Whenever the circuit steps, it operates on an input frontier over the H inputs.
    • durably log all K outputs. Associate every log entry here with the input frontier that was processed for it.
    • the circuit, even after restarts can always look at the output log to know the last successfully processed input frontier, and replay input logs as needed.
    • If we could log forever, the above is sufficient to implement a fault-tolerant circuit. But we can't log forever. So at any point after executing a step, as an optimization, the circuit can checkpoint its internal state. This will correspond to the input frontier covered by the step. If this succeeds, the circuit can trim the portion of the input logs that is covered by that input frontier + checkpoint.
    • Safety Invariant 1, replay safety: input log replays should not create more than one entry for a frontier in the output log.
    • Safety invariant 2, trim safety: if an event in the input log is ever trimmed, then it must be covered by a circuit checkpoint.
  • Our goal is to now make this model work for a data-parallel circuit with W workers.
    • The model we are going for is that each worker works on a partition of the input data. A partition is, in the abstract, a subspace of input events.
    • So now, every worker needs the following operations: read(input, partition, frontier, num_events), write(output, partition, frontier), get_frontier(output, partition), checkpoint(partition, frontier) and trim(input, partition, frontier).
    • If my summary so far is accurate, it looks to me like this design can handle circuit replication too, assuming the above safety invariants hold --- multiple workers can do the same work on the same partition, but only one of them will succeed with the write to the (shared) output log
  • For now, it seems like Kafka can help us implement read/write/get_frontier/trim. RocksDB helps us implement checkpoint()?
• Re. the exchange operator, I suggest having an interface where a worker can locally query what its shard assignment is, with the assignment itself being made (potentially) by an external coordinator. This separates the sharding policy from mechanism.

[blp]

I rewrote most of the "fault tolerance" and a lot of "scale in and out". I think it should be clearer and more approachable, and I clarified it in my own mind too. Please take another look when you have time: https://github.com/feldera/dist-design/blob/main/README.md