persistence semantics discussion (WIP)

[jhellerstein]

In Hydroflow we currently allow persistence to be specified per operator. This is causing some confusion on a few fronts.

To rescan or not to rescan

Consider the following code. It prints:

1. the ('static) join of two streams
2. the ('tick) join of two streams, each of which is persisted in a 'static group_by operator upstream.

Hypothesis: these should have the same results.

fn persist(
   left_recv: UnboundedReceiverStream<(usize, char)>,
   right_recv: UnboundedReceiverStream<(usize, char)>,
) -> Hydroflow {
   hydroflow_syntax! {
       left = source_stream(left_recv) -> tee();
       right = source_stream(right_recv) -> tee();
       // full symmetric join
       symmetric_join = join::<'static, 'static>()
           -> for_each(|(k, (v1, v2))| println!("symmetric join: ({}, ({}, {}))", k, v1, v2));
       left -> [0]symmetric_join;
       right -> [1]symmetric_join;
 
       // upstream persistence on both sides
       gbjoin = join::<'tick, 'tick>()
           -> for_each(|(k, (v1, v2))| println!("gb-streamjoin: ({}, ({}, {}))", k, v1, v2));
       left -> map(|(k,v)| ((k,v), 0))
           -> group_by::<'static, (usize, char), usize>(|| 0, |old: &mut usize, _new: usize| *old)
           -> map(|(k, _v): ((usize, char), _)| k) -> [0]gbjoin;
       right -> map(|(k,v)| ((k,v), 0))
           -> group_by::<'static, (usize, char), usize>(|| 0, |old: &mut usize, _new: usize| *old)
           -> map(|(k, _v): ((usize, char), _)| k) -> [1]gbjoin;
   }
}

We might think these would behave the same, but they don't. The 'static group_by op is rescanned each tick but the join is not, so the group_by outputs duplicates across ticks and the join does not:

pub fn main() {
    let opts = Opts::parse();

    let (left_send, left_recv) = hydroflow::util::unbounded_channel::<(usize, char)>();
    let (right_send, right_recv) = hydroflow::util::unbounded_channel::<(usize, char)>();
    let mut df =  persist(left_recv, right_recv);
    for i in 0..7 {
        left_send.send((i, 'a')).unwrap();
        right_send.send((i, 'x')).unwrap();
        if i > 0 {
            left_send.send((i - 1, 'b')).unwrap();
            right_send.send((i - 1, 'y')).unwrap();
        }
        println!("------------------- Tick {} -------------------", i);
        df.run_tick();
    }
}

produces:

log output

------------------- Tick 0 -------------------
symmetric join: (0, (a, x))
gb-streamjoin: (0, (a, x))
------------------- Tick 1 -------------------
------------------- Tick 2 -------------------
symmetric join: (0, (b, x))
symmetric join: (1, (a, x))
symmetric join: (1, (b, x))
symmetric join: (0, (a, y))
symmetric join: (0, (b, y))
symmetric join: (2, (a, x))
symmetric join: (1, (a, y))
symmetric join: (1, (b, y))
gb-streamjoin: (0, (b, y))
gb-streamjoin: (0, (a, y))
gb-streamjoin: (0, (b, x))
gb-streamjoin: (0, (a, x))
gb-streamjoin: (2, (a, x))
gb-streamjoin: (1, (a, y))
gb-streamjoin: (1, (b, y))
gb-streamjoin: (1, (a, x))
gb-streamjoin: (1, (b, x))
------------------- Tick 3 -------------------
------------------- Tick 4 -------------------
symmetric join: (2, (b, x))
symmetric join: (3, (a, x))
symmetric join: (3, (b, x))
symmetric join: (2, (a, y))
symmetric join: (2, (b, y))
symmetric join: (4, (a, x))
symmetric join: (3, (a, y))
symmetric join: (3, (b, y))
gb-streamjoin: (3, (b, x))
gb-streamjoin: (3, (a, x))
gb-streamjoin: (4, (a, x))
gb-streamjoin: (2, (a, x))
gb-streamjoin: (2, (b, x))
gb-streamjoin: (0, (b, y))
gb-streamjoin: (0, (a, y))
gb-streamjoin: (0, (b, x))
gb-streamjoin: (0, (a, x))
gb-streamjoin: (2, (a, y))
gb-streamjoin: (2, (b, y))
gb-streamjoin: (1, (a, y))
gb-streamjoin: (1, (b, y))
gb-streamjoin: (1, (a, x))
gb-streamjoin: (1, (b, x))
gb-streamjoin: (3, (b, y))
gb-streamjoin: (3, (a, y))
------------------- Tick 5 -------------------
------------------- Tick 6 -------------------
symmetric join: (4, (b, x))
symmetric join: (5, (a, x))
symmetric join: (5, (b, x))
symmetric join: (4, (a, y))
symmetric join: (4, (b, y))
symmetric join: (6, (a, x))
symmetric join: (5, (a, y))
symmetric join: (5, (b, y))
gb-streamjoin: (6, (a, x))
gb-streamjoin: (3, (b, x))
gb-streamjoin: (3, (a, x))
gb-streamjoin: (4, (a, x))
gb-streamjoin: (4, (b, x))
gb-streamjoin: (4, (a, y))
gb-streamjoin: (4, (b, y))
gb-streamjoin: (2, (a, x))
gb-streamjoin: (2, (b, x))
gb-streamjoin: (0, (b, y))
gb-streamjoin: (0, (a, y))
gb-streamjoin: (0, (b, x))
gb-streamjoin: (0, (a, x))
gb-streamjoin: (2, (a, y))
gb-streamjoin: (2, (b, y))
gb-streamjoin: (1, (a, y))
gb-streamjoin: (1, (b, y))
gb-streamjoin: (1, (a, x))
gb-streamjoin: (1, (b, x))
gb-streamjoin: (3, (b, y))
gb-streamjoin: (3, (a, y))
gb-streamjoin: (5, (b, x))
gb-streamjoin: (5, (a, x))
gb-streamjoin: (5, (b, y))
gb-streamjoin: (5, (a, y))

(as an aside, I'm not sure how the scheduler decides to dequeue channel inputs per tick. for debugging it would be nice if it drained all channels each time we call run_tick.)

Monotonicity and Lifetimes

Consider a flow containing just a join of two streams, but with different lifetimes:

fn joins(
    left_recv: UnboundedReceiverStream<(usize, char)>,
    right_recv: UnboundedReceiverStream<(usize, char)>,
) -> Hydroflow {
    hydroflow_syntax! {
        left = source_stream(left_recv) -> tee();
        right = source_stream(right_recv) -> tee();
        // full symmetric join
        symmetric_join = join::<'static, 'static>()
            -> for_each(|(k, (v1, v2))| println!("symmetric join: ({}, ({}, {}))", k, v1, v2));
        left -> [0]symmetric_join;
        right -> [1]symmetric_join;

        // stream-table join
        st_join = join::<'tick, 'static>()
            -> for_each(|(k, (v1, v2))| println!("stream-table join: ({}, ({}, {}))", k, v1, v2));
        left -> [0]st_join;
        right -> [1]st_join;

        // stream-stream join
        ss_join = join::<'tick, 'tick>()
            -> for_each(|(k, (v1, v2))| println!("stream-stream join: ({}, ({}, {}))", k, v1, v2));
        left -> [0]ss_join;
        right -> [1]ss_join;
    }
}

With the same inputs from main as above, we get:

log output

------------------- Tick 0 -------------------
symmetric join: (0, (a, x))
stream-table join: (0, (a, x))
stream-stream join: (0, (a, x))
------------------- Tick 1 -------------------
symmetric join: (0, (b, x))
symmetric join: (1, (a, x))
symmetric join: (0, (a, y))
symmetric join: (0, (b, y))
stream-table join: (0, (b, x))
stream-table join: (1, (a, x))
stream-table join: (0, (b, y))
stream-stream join: (1, (a, x))
stream-stream join: (0, (b, y))
------------------- Tick 2 -------------------
symmetric join: (1, (b, x))
symmetric join: (2, (a, x))
symmetric join: (1, (a, y))
symmetric join: (1, (b, y))
stream-table join: (1, (b, x))
stream-table join: (2, (a, x))
stream-table join: (1, (b, y))
stream-stream join: (2, (a, x))
stream-stream join: (1, (b, y))
------------------- Tick 3 -------------------
symmetric join: (2, (b, x))
symmetric join: (3, (a, x))
symmetric join: (2, (a, y))
symmetric join: (2, (b, y))
stream-table join: (2, (b, x))
stream-table join: (3, (a, x))
stream-table join: (2, (b, y))
stream-stream join: (3, (a, x))
stream-stream join: (2, (b, y))
------------------- Tick 4 -------------------
symmetric join: (3, (b, x))
symmetric join: (4, (a, x))
symmetric join: (3, (a, y))
symmetric join: (3, (b, y))
stream-table join: (3, (b, x))
stream-table join: (4, (a, x))
stream-table join: (3, (b, y))
stream-stream join: (4, (a, x))
stream-stream join: (3, (b, y))
------------------- Tick 5 -------------------
symmetric join: (4, (b, x))
symmetric join: (5, (a, x))
symmetric join: (4, (a, y))
symmetric join: (4, (b, y))
stream-table join: (4, (b, x))
stream-table join: (5, (a, x))
stream-table join: (4, (b, y))
stream-stream join: (5, (a, x))
stream-stream join: (4, (b, y))
------------------- Tick 6 -------------------
symmetric join: (5, (b, x))
symmetric join: (6, (a, x))
symmetric join: (5, (a, y))
symmetric join: (5, (b, y))
stream-table join: (5, (b, x))
stream-table join: (6, (a, x))
stream-table join: (5, (b, y))
stream-stream join: (6, (a, x))
stream-stream join: (5, (b, y))

Now let's add random delays:

pub fn main() {
    let opts = Opts::parse();

    let (left_send, left_recv) = hydroflow::util::unbounded_channel::<(usize, char)>();
    let (right_send, right_recv) = hydroflow::util::unbounded_channel::<(usize, char)>();
    let mut df = joins(left_recv, right_recv);

    let mut lefts = VecDeque::new();
    let mut rights = VecDeque::new();
    for i in 0..100 {
        if i < 6 {
            lefts.push_back((i, 'a'));
            rights.push_back((i, 'x'));
            if i > 0 {
                lefts.push_back((i - 1, 'b'));
                rights.push_back((i - 1, 'y'));
            }
        }
        while rand::random() && !lefts.is_empty() {
            left_send.send(lefts.pop_front().unwrap()).unwrap()
        }
        while rand::random() && !rights.is_empty() {
            right_send.send(rights.pop_front().unwrap()).unwrap()
        }
        println!("------------------- Tick {} -------------------", i);
        df.run_tick();
    }
}

Result (all but symmetric join vary across runs due to randomness of delays):

log output

------------------- Tick 0 -------------------
------------------- Tick 1 -------------------
symmetric join: (0, (a, x))
symmetric join: (0, (b, x))
symmetric join: (1, (a, x))
stream-table join: (0, (a, x))
stream-table join: (0, (b, x))
stream-table join: (1, (a, x))
stream-stream join: (1, (a, x))
------------------- Tick 2 -------------------
symmetric join: (1, (b, x))
stream-table join: (1, (b, x))
------------------- Tick 3 -------------------
symmetric join: (0, (a, y))
symmetric join: (0, (b, y))
------------------- Tick 4 -------------------
------------------- Tick 5 -------------------
symmetric join: (2, (a, x))
symmetric join: (2, (b, x))
symmetric join: (1, (a, y))
symmetric join: (1, (b, y))
------------------- Tick 6 -------------------
symmetric join: (3, (a, x))
symmetric join: (3, (b, x))
stream-table join: (3, (b, x))
stream-stream join: (3, (b, x))
------------------- Tick 7 -------------------
symmetric join: (2, (a, y))
symmetric join: (2, (b, y))
symmetric join: (4, (a, x))
symmetric join: (4, (b, x))
symmetric join: (3, (a, y))
symmetric join: (3, (b, y))
symmetric join: (5, (a, x))
------------------- Tick 8 -------------------
symmetric join: (4, (a, y))
symmetric join: (4, (b, y))
------------------- Tick 9 -------------------

Clearly the examples here that are deterministic are the ones that persist both sides, but that persistence is a transitive property along the graph as with the above group_by.

Getting the transitivity rules right is tricky. Here's an attempt. We have four properties:

1. Operator monotonicity (monotone/non-monotone): is the operator's output a monotone function of its input? E.g. join is monotone in both inputs; difference is non-monotone in its neg input.
2. Operator input persistence over time ('static/'tick): does the operator accumulate its input over time? if so that input is 'static else it's 'tick.
3. Edge delivery (cumulative/differential) over time: does the edge represent the upstream computation over all ticks, or just incremental change since the prior tick? If all it's cumulative, if just the last it's differential.
4. Edge monotonicity over time (t-monotone/t-non-monotone): does the edge deliver a monotone stream (in either cumulative or differential delivery mode)?

Axioms:

1. Op Enforces T-Monotonicity: If operator is monotone and all input persistence over time is static: output edge is t-monotone regardless of input edge properties.
2. Edge Replay Emulates 'static: If operator input [i] with persistence tick is 'cumulative' and t-monotone: can treat input [i] as having 'static semantics even though it does not persist (it gets repopulated each tick).
3. Non-Monotone Op = t-non-monotone output by definition.
4. Monotone+Tick Op = t-non-monotone output. Across ticks there is no guarantee that the op receives monotone inputs, so its outputs may not be monotone across ticks.

To be continued...

Reference Behavior: Bloom

For reference, in Bloom we have persistence on sources (table vs scratch/channel). In each Bloom tick we scan the tables/scratches/channels and rerun all the operators. At the end of the tick, the scratches and channels and operator state are all zeroed.

[MingweiSamuel]

With the run_tick() fix #350:
For the first example

log output

------------------- Tick 0 -------------------
symmetric join: (0, (a, x))
gb-streamjoin: (0, (a, x))
------------------- Tick 1 -------------------
symmetric join: (0, (b, x))
symmetric join: (1, (a, x))
symmetric join: (0, (a, y))
symmetric join: (0, (b, y))
gb-streamjoin: (0, (a, x))
gb-streamjoin: (0, (b, x))
gb-streamjoin: (1, (a, x))
gb-streamjoin: (0, (a, y))
gb-streamjoin: (0, (b, y))
------------------- Tick 2 -------------------
symmetric join: (1, (b, x))
symmetric join: (2, (a, x))
symmetric join: (1, (a, y))
symmetric join: (1, (b, y))
gb-streamjoin: (1, (b, x))
gb-streamjoin: (1, (a, x))
gb-streamjoin: (2, (a, x))
gb-streamjoin: (0, (a, x))
gb-streamjoin: (0, (b, x))
gb-streamjoin: (0, (a, y))
gb-streamjoin: (0, (b, y))
gb-streamjoin: (1, (b, y))
gb-streamjoin: (1, (a, y))
------------------- Tick 3 -------------------
symmetric join: (2, (b, x))
symmetric join: (3, (a, x))
symmetric join: (2, (a, y))
symmetric join: (2, (b, y))
gb-streamjoin: (3, (a, x))
gb-streamjoin: (1, (b, x))
gb-streamjoin: (1, (a, x))
gb-streamjoin: (2, (b, x))
gb-streamjoin: (2, (a, x))
gb-streamjoin: (2, (b, y))
gb-streamjoin: (2, (a, y))
gb-streamjoin: (0, (a, x))
gb-streamjoin: (0, (b, x))
gb-streamjoin: (0, (a, y))
gb-streamjoin: (0, (b, y))
gb-streamjoin: (1, (b, y))
gb-streamjoin: (1, (a, y))
------------------- Tick 4 -------------------
symmetric join: (3, (b, x))
symmetric join: (4, (a, x))
symmetric join: (3, (a, y))
symmetric join: (3, (b, y))
gb-streamjoin: (1, (a, x))
gb-streamjoin: (1, (b, x))
gb-streamjoin: (4, (a, x))
gb-streamjoin: (3, (a, y))
gb-streamjoin: (3, (b, y))
gb-streamjoin: (2, (b, y))
gb-streamjoin: (2, (a, y))
gb-streamjoin: (0, (a, x))
gb-streamjoin: (0, (b, x))
gb-streamjoin: (2, (b, x))
gb-streamjoin: (2, (a, x))
gb-streamjoin: (0, (a, y))
gb-streamjoin: (0, (b, y))
gb-streamjoin: (3, (a, x))
gb-streamjoin: (3, (b, x))
gb-streamjoin: (1, (a, y))
gb-streamjoin: (1, (b, y))
------------------- Tick 5 -------------------
symmetric join: (4, (b, x))
symmetric join: (5, (a, x))
symmetric join: (4, (a, y))
symmetric join: (4, (b, y))
gb-streamjoin: (4, (a, y))
gb-streamjoin: (4, (b, y))
gb-streamjoin: (1, (a, x))
gb-streamjoin: (1, (b, x))
gb-streamjoin: (4, (a, x))
gb-streamjoin: (4, (b, x))
gb-streamjoin: (3, (a, y))
gb-streamjoin: (3, (b, y))
gb-streamjoin: (2, (b, y))
gb-streamjoin: (2, (a, y))
gb-streamjoin: (0, (a, x))
gb-streamjoin: (0, (b, x))
gb-streamjoin: (5, (a, x))
gb-streamjoin: (2, (b, x))
gb-streamjoin: (2, (a, x))
gb-streamjoin: (0, (a, y))
gb-streamjoin: (0, (b, y))
gb-streamjoin: (3, (a, x))
gb-streamjoin: (3, (b, x))
gb-streamjoin: (1, (a, y))
gb-streamjoin: (1, (b, y))
------------------- Tick 6 -------------------
symmetric join: (5, (b, x))
symmetric join: (6, (a, x))
symmetric join: (5, (a, y))
symmetric join: (5, (b, y))
gb-streamjoin: (4, (a, y))
gb-streamjoin: (4, (b, y))
gb-streamjoin: (1, (a, x))
gb-streamjoin: (1, (b, x))
gb-streamjoin: (4, (a, x))
gb-streamjoin: (4, (b, x))
gb-streamjoin: (6, (a, x))
gb-streamjoin: (3, (a, y))
gb-streamjoin: (3, (b, y))
gb-streamjoin: (2, (b, y))
gb-streamjoin: (2, (a, y))
gb-streamjoin: (0, (a, x))
gb-streamjoin: (0, (b, x))
gb-streamjoin: (5, (b, x))
gb-streamjoin: (5, (a, x))
gb-streamjoin: (2, (b, x))
gb-streamjoin: (2, (a, x))
gb-streamjoin: (5, (b, y))
gb-streamjoin: (5, (a, y))
gb-streamjoin: (0, (a, y))
gb-streamjoin: (0, (b, y))
gb-streamjoin: (3, (a, x))
gb-streamjoin: (3, (b, x))
gb-streamjoin: (1, (a, y))
gb-streamjoin: (1, (b, y)

---

log output

------------------- Tick 0 -------------------
symmetric join: (0, (a, x))
stream-table join: (0, (a, x))
stream-stream join: (0, (a, x))
------------------- Tick 1 -------------------
symmetric join: (0, (b, x))
symmetric join: (1, (a, x))
symmetric join: (0, (a, y))
symmetric join: (0, (b, y))
stream-table join: (0, (b, x))
stream-table join: (1, (a, x))
stream-table join: (0, (b, y))
stream-stream join: (1, (a, x))
stream-stream join: (0, (b, y))
------------------- Tick 2 -------------------
symmetric join: (1, (b, x))
symmetric join: (2, (a, x))
symmetric join: (1, (a, y))
symmetric join: (1, (b, y))
stream-table join: (1, (b, x))
stream-table join: (2, (a, x))
stream-table join: (1, (b, y))
stream-stream join: (2, (a, x))
stream-stream join: (1, (b, y))
------------------- Tick 3 -------------------
symmetric join: (2, (b, x))
symmetric join: (3, (a, x))
symmetric join: (2, (a, y))
symmetric join: (2, (b, y))
stream-table join: (2, (b, x))
stream-table join: (3, (a, x))
stream-table join: (2, (b, y))
stream-stream join: (3, (a, x))
stream-stream join: (2, (b, y))
------------------- Tick 4 -------------------
symmetric join: (3, (b, x))
symmetric join: (4, (a, x))
symmetric join: (3, (a, y))
symmetric join: (3, (b, y))
stream-table join: (3, (b, x))
stream-table join: (4, (a, x))
stream-table join: (3, (b, y))
stream-stream join: (4, (a, x))
stream-stream join: (3, (b, y))
------------------- Tick 5 -------------------
symmetric join: (4, (b, x))
symmetric join: (5, (a, x))
symmetric join: (4, (a, y))
symmetric join: (4, (b, y))
stream-table join: (4, (b, x))
stream-table join: (5, (a, x))
stream-table join: (4, (b, y))
stream-stream join: (5, (a, x))
stream-stream join: (4, (b, y))
------------------- Tick 6 -------------------
symmetric join: (5, (b, x))
symmetric join: (6, (a, x))
symmetric join: (5, (a, y))
symmetric join: (5, (b, y))
stream-table join: (5, (b, x))
stream-table join: (6, (a, x))
stream-table join: (5, (b, y))
stream-stream join: (6, (a, x))
stream-stream join: (5, (b, y))

log output

------------------- Tick 0 -------------------
------------------- Tick 1 -------------------
------------------- Tick 2 -------------------
symmetric join: (0, (a, x))
symmetric join: (0, (a, y))
stream-table join: (0, (a, x))
stream-table join: (0, (a, y))
------------------- Tick 3 -------------------
symmetric join: (1, (a, x))
stream-table join: (1, (a, x))
------------------- Tick 4 -------------------
symmetric join: (0, (b, x))
symmetric join: (0, (b, y))
stream-table join: (0, (b, x))
stream-table join: (0, (b, y))
------------------- Tick 5 -------------------
symmetric join: (2, (a, x))
symmetric join: (1, (a, y))
stream-table join: (2, (a, x))
------------------- Tick 6 -------------------
symmetric join: (2, (a, y))
------------------- Tick 7 -------------------
symmetric join: (1, (b, x))
symmetric join: (1, (b, y))
stream-table join: (1, (b, x))
stream-table join: (1, (b, y))
------------------- Tick 8 -------------------
------------------- Tick 9 -------------------
symmetric join: (3, (a, x))
symmetric join: (2, (b, x))
symmetric join: (2, (b, y))
stream-table join: (3, (a, x))
stream-table join: (2, (b, x))
stream-table join: (2, (b, y))
------------------- Tick 10 -------------------
------------------- Tick 11 -------------------
symmetric join: (3, (a, y))
------------------- Tick 12 -------------------
symmetric join: (4, (a, x))
symmetric join: (4, (a, y))
stream-table join: (4, (a, x))
stream-table join: (4, (a, y))
------------------- Tick 13 -------------------
symmetric join: (3, (b, x))
symmetric join: (3, (b, y))
stream-table join: (3, (b, x))
stream-table join: (3, (b, y))
------------------- Tick 14 -------------------
symmetric join: (5, (a, x))
stream-table join: (5, (a, x))
------------------- Tick 15 -------------------
------------------- Tick 16 -------------------
------------------- Tick 17 -------------------
symmetric join: (4, (b, x))
symmetric join: (4, (b, y))
stream-table join: (4, (b, x))
stream-table join: (4, (b, y))
------------------- Tick 18 -------------------

[jhellerstein]

Thinking about rescanning logic today. Playing around with introducing more rescanning. It was easy to hack group_by to rescan each tick. Seems like doing that for join would require adding some rescan logic.

Results of the first example program above with a replaying group_by are pasted at bottom of this note. You can see how gb-streamjoin produces monotonically increasing output.

Playing with this, I’m trying to think both about correctness and about how complex our API is…

At the client API boundary:

1. For truly “stateless” clients, a monotone producer should literally reiterate the full lattice point each time — that gives the client the “experience” of a monotone producer.
2. If we assume a stateful client — e.g. a CLI client with scrollable buffer — it may be preferable to only output deltas over time.

Inside a dataflow:

1. A static consumer does not need a rescanning producer for correctness: it’s already persisting the state
2. Maybe a tick consumer should not have a rescanning producer? It sort of defeats the tick semantic. OTOH, if a developer plumbs a graph with a monotone operator feeding a tick-based operator, what are they trying to achieve? We need clarity here.
3. A stateless consumer (a la map) is agnostic to replay — it simply passes along the behavior of its upstream producers

Optimizations: In certain cases we may have static ops whose upstream producers are also static. It may be possible to optimize here in a few ways akin to materialized views:

• save space by removing the state from some of the static ops (intuition: if we are static upstream and replay, we could recompute as an unmaterialized view downstream)
• we could allow the upstream static to only forward deltas to the downstream static — this is akin to materialized view maintenance, and also client item 2 above.

Whatever the right answers here, I'm increasingly convinced that there are 3 bits of information here:

1. static vs tick memory in ops (a cross-tick issue)
2. streaming vs cumulative ports/edges for large objects like sets (a within-tick issue!)
3. for streaming edges, replay vs delta cross-tick semantics (a cross-tick issue)

log output

------------------- Tick 0 -------------------
stream-table join: (0, (a, x))
table-stream join: (0, (a, x))
symmetric join: (0, (a, x))
gb-streamjoin: (0, (a, x))
------------------- Tick 1 -------------------
symmetric join: (0, (b, x))
symmetric join: (1, (a, x))
symmetric join: (0, (a, y))
symmetric join: (0, (b, y))
stream-table join: (0, (b, x))
stream-table join: (1, (a, x))
stream-table join: (0, (b, y))
table-stream join: (1, (a, x))
table-stream join: (0, (a, y))
table-stream join: (0, (b, y))
gb-streamjoin: (1, (a, x))
gb-streamjoin: (0, (b, x))
gb-streamjoin: (0, (a, x))
gb-streamjoin: (0, (b, y))
gb-streamjoin: (0, (a, y))
------------------- Tick 2 -------------------
symmetric join: (1, (b, x))
symmetric join: (2, (a, x))
symmetric join: (1, (a, y))
symmetric join: (1, (b, y))
stream-table join: (1, (b, x))
stream-table join: (2, (a, x))
stream-table join: (1, (b, y))
table-stream join: (2, (a, x))
table-stream join: (1, (a, y))
table-stream join: (1, (b, y))
gb-streamjoin: (1, (a, x))
gb-streamjoin: (1, (b, x))
gb-streamjoin: (2, (a, x))
gb-streamjoin: (0, (a, y))
gb-streamjoin: (0, (b, y))
gb-streamjoin: (1, (a, y))
gb-streamjoin: (1, (b, y))
gb-streamjoin: (0, (a, x))
gb-streamjoin: (0, (b, x))
------------------- Tick 3 -------------------
symmetric join: (2, (b, x))
symmetric join: (3, (a, x))
symmetric join: (2, (a, y))
symmetric join: (2, (b, y))
stream-table join: (2, (b, x))
stream-table join: (3, (a, x))
stream-table join: (2, (b, y))
table-stream join: (3, (a, x))
table-stream join: (2, (a, y))
table-stream join: (2, (b, y))
gb-streamjoin: (1, (a, x))
gb-streamjoin: (1, (b, x))
gb-streamjoin: (2, (a, x))
gb-streamjoin: (2, (b, x))
gb-streamjoin: (0, (a, y))
gb-streamjoin: (0, (b, y))
gb-streamjoin: (1, (a, y))
gb-streamjoin: (1, (b, y))
gb-streamjoin: (3, (a, x))
gb-streamjoin: (2, (a, y))
gb-streamjoin: (2, (b, y))
gb-streamjoin: (0, (a, x))
gb-streamjoin: (0, (b, x))
------------------- Tick 4 -------------------
symmetric join: (3, (b, x))
symmetric join: (4, (a, x))
symmetric join: (3, (a, y))
symmetric join: (3, (b, y))
stream-table join: (3, (b, x))
stream-table join: (4, (a, x))
stream-table join: (3, (b, y))
table-stream join: (4, (a, x))
table-stream join: (3, (a, y))
table-stream join: (3, (b, y))
gb-streamjoin: (1, (b, x))
gb-streamjoin: (1, (a, x))
gb-streamjoin: (3, (a, x))
gb-streamjoin: (3, (b, x))
gb-streamjoin: (0, (a, y))
gb-streamjoin: (0, (b, y))
gb-streamjoin: (2, (b, y))
gb-streamjoin: (2, (a, y))
gb-streamjoin: (4, (a, x))
gb-streamjoin: (0, (a, x))
gb-streamjoin: (0, (b, x))
gb-streamjoin: (3, (a, y))
gb-streamjoin: (3, (b, y))
gb-streamjoin: (2, (b, x))
gb-streamjoin: (2, (a, x))
gb-streamjoin: (1, (b, y))
gb-streamjoin: (1, (a, y))
------------------- Tick 5 -------------------
symmetric join: (4, (b, x))
symmetric join: (5, (a, x))
symmetric join: (4, (a, y))
symmetric join: (4, (b, y))
stream-table join: (4, (b, x))
stream-table join: (5, (a, x))
stream-table join: (4, (b, y))
table-stream join: (5, (a, x))
table-stream join: (4, (a, y))
table-stream join: (4, (b, y))
gb-streamjoin: (1, (b, x))
gb-streamjoin: (1, (a, x))
gb-streamjoin: (3, (a, x))
gb-streamjoin: (3, (b, x))
gb-streamjoin: (0, (a, y))
gb-streamjoin: (0, (b, y))
gb-streamjoin: (2, (b, y))
gb-streamjoin: (2, (a, y))
gb-streamjoin: (4, (a, y))
gb-streamjoin: (4, (b, y))
gb-streamjoin: (4, (a, x))
gb-streamjoin: (4, (b, x))
gb-streamjoin: (0, (a, x))
gb-streamjoin: (0, (b, x))
gb-streamjoin: (3, (a, y))
gb-streamjoin: (3, (b, y))
gb-streamjoin: (2, (b, x))
gb-streamjoin: (2, (a, x))
gb-streamjoin: (5, (a, x))
gb-streamjoin: (1, (b, y))
gb-streamjoin: (1, (a, y))
------------------- Tick 6 -------------------
symmetric join: (5, (b, x))
symmetric join: (6, (a, x))
symmetric join: (5, (a, y))
symmetric join: (5, (b, y))
stream-table join: (5, (b, x))
stream-table join: (6, (a, x))
stream-table join: (5, (b, y))
table-stream join: (6, (a, x))
table-stream join: (5, (a, y))
table-stream join: (5, (b, y))
gb-streamjoin: (1, (b, x))
gb-streamjoin: (1, (a, x))
gb-streamjoin: (3, (a, x))
gb-streamjoin: (3, (b, x))
gb-streamjoin: (0, (a, y))
gb-streamjoin: (0, (b, y))
gb-streamjoin: (2, (b, y))
gb-streamjoin: (2, (a, y))
gb-streamjoin: (4, (a, y))
gb-streamjoin: (4, (b, y))
gb-streamjoin: (4, (a, x))
gb-streamjoin: (4, (b, x))
gb-streamjoin: (0, (a, x))
gb-streamjoin: (0, (b, x))
gb-streamjoin: (3, (a, y))
gb-streamjoin: (3, (b, y))
gb-streamjoin: (2, (b, x))
gb-streamjoin: (2, (a, x))
gb-streamjoin: (5, (a, x))
gb-streamjoin: (5, (b, x))
gb-streamjoin: (6, (a, x))
gb-streamjoin: (1, (b, y))
gb-streamjoin: (1, (a, y))
gb-streamjoin: (5, (a, y))
gb-streamjoin: (5, (b, y))

[jhellerstein]

Recall the 3 bits of information in the previous post:

1. streaming vs cumulative ports/edges for large objects like sets (a within-tick transport/compression issue!)
2. 'static vs 'tick memory in ops (a cross-tick semantic issue)
3. replay vs delta semantics (a cross-tick semantic issue)

Let's try to get notation for all this:

• $\sqcup$ is our standard lattice-join operator
• $\bigsqcup_\tau$ is the across-time lattice-join operator. (Github markdown won't support both subscript and superscript!)
• $\flat p$ flattens a single lattice point $p$ into a stream of lattice points $p_1 \ldots p_k$ s.t. $p_1 \sqcup ... \sqcup p_k = p$.
• $\Delta(p, q)$ is a minimal lattice point such that $q \sqcup \Delta(p, q) = p$.

Note that $\flat p$ is not uniquely defined -- we can flatten into more-smaller or few-bigger points. We also have not dictated whether the points themselves must be non-overlapping (i.e. $p_i \sqcap p_k = \bot$). For now I think that's fine. $\Delta$ is not well-defined on all lattices either, but that's OK too -- if it's not defined, it's not defined, and we forbid it.

Going back to our 3 bits, we can convert between their settings as follows:

1. Stream-Cumulative: If we have a stream of lattice points in our dataflow and we want a cumulative point, we simply use $\sqcup$. For example, the relational join of two streams $s_1 \bowtie s_2$ can be defined as a (monotone) function of the set of items in the two streams: i.e. $\sqcup(s_1) \bowtie \sqcup(s_2)$. Cumulative-Stream In the other direction, we use $\flat$ to turn a cumulative point into a stream.
2. 'tick-'static stream conversion is achieved via a $\bigsqcup_\tau$ operator. For example, a 'static-'static join takes incoming 'tick-lifetime data and persist it statically as $\bigsqcup_\tau(s_1) \bowtie \bigsqcup_\tau(s_2)$. (In this characterization, the join of two streams is a simple monotone binary function of the two sets accumulated by $\bigsqcup$. 'static-'tick conversion is effectively a no-op -- the upstream 'static memory will cause the downstream 'tick-based memory to be "rehydrated" with past information each tick.
3. By default, it would seem that monotone operators should release increasingly large lattice points. The invariant on a cumulative output $port$ $o$ of an operator is that $o_{t+1} \ge o_t$. The invariant on a stream output port would then be that $\sqcup(o)_{t+1} \ge \sqcup(o)_t$ -- i.e. the port should replay the data from last tick in the current tick (and perhaps more). replay-delta: if we have two stateful operators $op_1$ and $op_2$ where $op_1 \rightarrow op_2$ is in the flow, we should be able to interpose a $\Delta$ operator in between $op_1 \rightarrow \Delta \rightarrow op_2$ without changing semantics (if such an operator is defined on our lattice type). That $\Delta$ operator would need to know/remember what $op_1$ has produced and what $op_2$ has consumed. (Performing this rewrite with other operators in between $op_1$ and $op_2$ would require more careful analysis.) Note that this analysis could go across a network from server to client -- the client might prefer a stream of minimal deltas. delta-replay conversion seems to always be correct.

It would seem that $\Delta$ works correctly regardless of the lifetimes of the producer/consumer pair, as long as it knows the state of both. It's simply compressing the data on the connection between them.

[jhellerstein]

I think the above is enough to capture lifetime semantics. It models our 'tick and 'static operator decorations fairly naturally. The current implementation of hydroflow is all streaming, no cumulative edges. That seems OK, at least locally. To convert local streaming code to equivalent distributed code we'll need to implement a sealing protocol (e.g. ordered delivery with EOS, or unordered delivery with a manifest/seal like count).

The one thing we probably have wrong now is that our 'static operators should be replaying each tick to provide true monotonicity. To make that less annoying at the client API, we really need to (a) implement $\Delta$ and (b) offer a client choice between replaying or deltas. To make it more efficient we need to optimize our flows (e.g. minimize redundant state by removing redundant 'statics in a pipeline, or minimize iteration/data-passing by putting $\Delta$ between stateful ops).

[jhellerstein]

Remaining issues:

• Use this algebra to derive properties regarding monotonicity and order-insensitivity, non-monotonicity and order-sensitivity, and non-monotonicity + non-determinism as "taint".
• Write axioms for "taint transfer" downstream correctly.
• Non-monotone ops and stratification. Introduce a $\blacksquare$ operator that represents a sealed lattice-join.
• How do various sources of order (time, strata, others) fit in?
• Challenge problem: represent pipeline hash join as a flow of these operators that demonstrates monotonicity.
• Challenge problem: capture the notion of "catchup for late-joining consumers"
• Challenge problem: capture Dedalus semantics (ticks and strata) explicitly within a flow, e.g. using a lattice of $\mathbb{N}$ for each ordered "clock signature". Extra credit: can we make some argument about correctness? Extra extra credit: can we choose different "clocks" and still argue correctness?

[jhellerstein]

Proposal for Consistency Analysis Under Dedalus Time

Background on the Dedalus Time Model

In this doc we're going to embrace the Dedalus/Bloom notion of time in which:

• Each node has its own clock
• Each node is a fixpoint transducer: within a clock tick, it takes a batch of data from input, and continues to generate output for that tick until it reaches fixpoint.
• Non-monotonicity is handled via stratification
• Cycles through non-monotone operators must go across a tick boundary

It's interesting to try and unpack this model and remix it in different ways, but that is outside the scope of this document.

Notation

• $\sqcup$ is our standard lattice-join operator
• $\bigsqcup_\tau$ is the across-time lattice-join operator.
• $\flat p$ flattens a single lattice point $p$ into a stream of lattice points $p_1 \ldots p_k$ s.t. $p_1 \sqcup ... \sqcup p_k = p$.
• $\Delta(p, q)$ is a minimal lattice point such that $q \sqcup \Delta(p, q) = p$.
• Edges can be cumulative (one lattice point per tick) or streaming (a stream of points per tick to be interpreted as the merge $(\sqcup)$ of the points in the tick).

Edge Properties

We focus on two semantic properties of edges: monotonicity and non-determinism.

• An edge is monotonic if its output from stream $S_{s+1}$ is no smaller in lattice order than its output from $S_s$, a prefix of $S_{s+1}$.
  • A monotonic cumulative edge is, by definition, monotonic over time since it processes one item per tick. Effectively it represents $\bigsqcup_\tau$ for some (possibly composite) lattice.
  • By extension, a stream edge is monotonic over time if the merge $(\sqcup)$ of its stream at time $t+1$ is no smaller than the merge of its stream at time $t$. I.e. it provides the semantics of $\bigsqcup_t(\sqcup(tick_t))$.
• An edge is deterministic if it is guaranteed to produce the same stream of outputs given the same stream of inputs. Otherwise it is non-deterministic.

In the Dedalus-based Hydroflow model (and the rest of this document), when we speak of monotonicity and determinism, we are referring to the across-time monotonicity and determinism of an edge. That is, we are referring to monotonicity and determinism with respect to the $\bigsqcup_\tau$ of the edge's stream (over time).

Definition: An edge that is both non-monotonic and non-deterministic is inconsistent: it may produce different values across runs and replicas, not just different orders/associations

Operators

We will assume all operators consume streams of lattice points (i.e. multiple per tick) as in current Hydroflow.

Definition: We say an operator is monotonic if its output edges are monotonic whenever all of its input edges are monotonic.
Definition: We say an operator is deterministic if its output edges are deterministic whenever all of its input edges are deterministic.

Observation: Composability: Monotonicity and Determinism are both composable properties: we can assemble "pipelines" of monotonic edges and operators and they can be encapsulated as a monotonic operator. Similar for determinism.

Network input operators These operators perform "demonic" permutation and association $(\sqcup)$ on streams. Hence they are non-deterministic across runs and replicas. However they are monotonic, stamped as a vector of the clock lattice $(\mathbb{N},max)$ and an opaque payload.

• source_stream and source_stream_serde.

Deterministic generators These operators generate the same data across runs and replicas. They are deterministic.

• source_iter is deterministic but not monotonic.
• repeat_iter outputs the same data each tick. It is effectively the same as source_iter $\rightarrow \bigsqcup_\tau$ (see Property Inference section below).

Stateless (Standard) operators map from one lattice point to one or more other points (perhaps in another lattice). They may be monotone or non-monotone, deterministic or non-deterministic.

• The following stateless Hydroflow operators are always monotone and deterministic: flatten, identity, merge, null, tee, unzip.
• The following stateless Hydroflow operators inherit their monotonicity and determinism from their UDF (code block): demux, filter, filter_map, flat_map, for_each, map
• The inspect operator is non-monotone but deterministic.

Aggregation operators are "stateful" macro-operators above. That is, formally we can model them as composite operators involving either a $\sqcup$ or $\bigsqcup_\tau$ operator to accumulate state. For example, reduce is:

map(|x| into_accumulator_lattice) -> ⊔ -> map(|a| out_of_accumulator)

fold is the same, but into a lattice with an interesting bottom element.
And group_by is:

map(|x| into_hash_lattice) -> ⊔ -> ♭

If we represent hash-lattice map as $\sharp$, group_by "naturally" becomes $\sharp \rightarrow \sqcup \rightarrow \flat$.

Optionally replace $\sqcup$ with $\bigsqcup_\tau$ in the above discussion for cross-tick persistence.

These ops are monotone iff they use $\bigsqcup_\tau$ and monotone map functions.

• fold, group_by, reduce

Ordering operators are per-tick stateful operators. (They'd be non-terminating if they persisted across ticks). Effectively they operate on a single lattice point per tick, so require a $\sqcup$ operator as their input to accumulate state. They are deterministic but non-monotone: given input stream $S' > S$, we cannot guarantee that sort $(S') >$ sort $(S)$ because $S'$ may be a range of values that overlaps $S$ but extends lower and/or higher.

• sort sort_by

Joins generate elements of the cross-product space of their inputs. While they are well-defined as stateless operators (e.g. joining up singleton sets) across two streams, they are almost never intended to be used that way. We will assume accumulators $(\sqcup$ or $\bigsqcup_\tau)$ as their inputs, and a $\flat$ operator at the output to convert back to a stream. That is, join is a macro for $\flat(\sqcup(R) \bowtie \sqcup(S))$ (optionally replacing one or more of the $\sqcup$ with $\bigsqcup_\tau$).

Join is a monotone binary function of its inputs, so if both inputs are monotone $(\bigsqcup_\tau)$ then the output is monotone. Note that the outer_join family (including left_join and right_join) is not included here: they are not monotone, but rather composite operators that tee inputs to join and difference -> map pipelines, and merge the results. (We can quibble about this if we treat NULL as $\bot$ of some lattice, but let's save that discussion for another day.)

• join, cross_join

Negation operators are non-monotonic, deterministic operators. Like join they are well-defined as stateless operators (e.g. the difference of two singleton sets), but almost never intended to be used that way, so we will assume accumulators at their inputs, and $\flat$ at their output. E.g. difference is $\flat(\sqcup(R) - \sqcup(S))$, potentially substituting $\bigsqcup_\tau$ for $\sqcup$.

• difference, not_in, anti_join, not_exists, all

Stateful Ops and Persistence Annotations

For stateful ops (aggregations, ordering, and the typical uses of join and negation), we will "embed" the lattice-join operators into the operator's type signature, which contains a list of persistence annotations corresponding to the list of input edges; each annotation is either 'static or 'tick.
Inputs marked 'static implement $\bigsqcup_\tau$ of their inputs, while those marked 'tick start at $\bot$ each tick and implement the $\sqcup$ of their inputs during the tick.

As a corollary to the discussion above, if any input to a stateful operator is marked 'tick, the operator is non-monotonic by definition.

Property Inference

Here are the basic rules for inferring monotonicity and determinism of operators from their inputs:

• Enforcement of Monotonicity: The $\bigsqcup_\tau$ operator produces monotonic output regardless of its input.
• Composition of Monotonicity: If an operator is monotonic, and its inputs are all monotonic, its outputs are monotonic.
• Composition of Determinism: If an operator is deterministic, and its inputs are all deterministic, its outputs are deterministic.

Given a Hydroflow graph, we can use the three rules above and the operator definitions to infer the monotonicity and determinism of each edge in the graph. The inference pseudocode is as follows:

1. Conservatively mark all edges both non-monotonic and non-deterministic.
2. Iterate over the graph, propagating monotonicity and determinism as follows:
   • Mark the output edge of every occurrence of $\bigsqcup_\tau$ as monotone.
   • Mark the output edge of any deterministic source operator as deterministic.
   • For every other monotone operator, if all its input edges are monotone, mark its output edges monotone.
   • For every other deterministic operator, if all its input edges are deterministic, mark its output edges deterministic.
3. Repeat until fixpoint
4. Upon completion, any edge that remains both non-monotonic and non-deterministic is marked inconsistent.

Delta Optimization

In a fully monotonic subgraph that has multiple $\bigsqcup_\tau$ operators, it may be possible to interpose a $\Delta$ operator to minimize the data that needs to flow between them without changing semantics. Similarly it may be possible to remove one or more of the $\bigsqcup_\tau$ operators without changing semantics.

We'll set aside this topic for another discussion, but it's worth mentioning here if only to motivate the need for the $\Delta$ operator.

[shadaj]

Copying over from Slack for better, uh, persistence:

throwing this out there to gauge if this is an interesting idea or it’s already obvious…

So right now it’s up to the developer to determine when to use ’tick or ’static according to their application constraints, with some operators defaulting to ’tick semantics (difference / anti join) and some defaulting to ’static (join). I started to think what HF would look like if all operators were ’tick by default… then there’s a relatively simple mental model of “all state is flushed after processing each batch of inputs”

Of course there are scenarios where you want to accumulate data across ticks for, let’s say a join. Consider a snippet of a chat example…

message_distributor = join() -> messages_out
messages_in -> [0] message_distributor
recipients_in -> [1] message_distributor

Now let’s assume we do not have access to persistence lifetimes, then to ensure that messages are actually sent to the entire group, we need to accumulate them. Let’s say we have an operator persist() defined to accumulate the history of all values it has seen across all ticks and emits the entire history on every tick. Then we can have the appropriate app semantics with

message_distributor = join() -> messages_out
messages_in -> [0] message_distributor
recipients_in -> persist() -> [1] message_distributor

Of course, this will be very inefficient because the join state is reset every tick so we have rehash all the elements. But this program is behaviorally equivalent to one with lifetimes that is much faster!

message_distributor = join::<'tick, 'static>() -> messages_out
messages_in -> [0] message_distributor
recipients_in -> [1] message_distributor

If our application wants new recipients to be able to see old messages, we can persist both sides of the join:

message_distributor = join() -> messages_out
messages_in -> persist() -> [0] message_distributor
recipients_in -> persist() -> [1] message_distributor

But this will actually result in confusing behavior, as there will be redundant packets sent to each recipient on each tick. What we want is to send out new join results compared to the previous tick. So we can introduce another operator that is the opposite of persist: unpersist() treats the stream of inputs as a multiset and subtracts the inputs from the previous tick.

message_distributor = join() -> unpersist() -> messages_out
messages_in -> persist() -> [0] message_distributor
recipients_in -> persist() -> [1] message_distributor

This program is horrendously slow if implemented naively, but once again it is equivalent to lifetimes!

message_distributor = join::<'static, 'static>() -> messages_out
messages_in -> [0] message_distributor
recipients_in -> [1] message_distributor

The cool thing here is that a developer can reason about what persist() and unpersist() do locally. And so using lifetimes can be seen as a compiler optimization that is eliminating persist() and unpersist() from the compute graph. And furthermore, the situation in which persist() cannot be optimized away is precisely the presence of a non-monotone operator downstream, so this could also work well to help developers identify non-monotonicity that hurts the performance of their application.