make edge references cause scheduling

[MingweiSamuel]

as pointed out by @shadaj

When state changes it should trigger (the state/sources that feed into) the operators that reference that state

[jhellerstein]

Plan is to achieve this by rewriting references to be prepared via an n-way cross-join bimorphism, and fed downstream as a tuple. That means the consumer needs to be rewritten to handle that tuple rather than its original input.

x = source_stream(...) -> fold(||...);
// good = source_stream(...) 
//   -> map(|(i, j)| i*j + x);
x -> [1]cross;
good = source_stream(...) -> [0]cross;
cross = cross_join() 
    -> map(|((i, j), x)| i*j + x);

[shadaj]

Hmm I'm a bit worried about actually implementing this as a cross join. While this is the correct mental model I wonder if it might get in the way of Rust's closure capturing logic since often the closure will be fine with just a borrow of the singleton rather than a copy.

[MingweiSamuel]

That's a good point