Removing an unseen tuple actually inserts it about differential-dataflow HOT 4 OPEN

let mut r_in = InputSession::new();

        let probe = worker.dataflow(|transitive_closure| {
            let r = r_in.to_collection(transitive_closure);

            r
                .iterate(|inner| {
                    let join = inner
                        .map(|(x,y)| (y,x))
                        .join(&inner)
                        .map(|(_, (x,z))| (x,z));

                    inner
                        .concat(&join)
                        .consolidate()
                        .distinct()
                })
                .inspect(|x| println!("r{:?}", x))
                .probe()
        });

fn main() {
    timely::execute_from_args(std::env::args(), |worker| {
        let mut input = InputSession::new();

        let probe = worker.dataflow(|transitive_closure| {
            let input_collection = input.to_collection(transitive_closure);

            input_collection
                .inspect(|tuple| println!("in: {:?}", tuple))
                .map(|(x,y)| (y,x))
                .inspect(|tuple| println!("flip in: {:?}", tuple))
                .join(&input_collection)
                .inspect(|tuple| println!("join match: {:?}", tuple)).probe()
        });

        let mut time = 0;
        input.advance_to(time);
        loop {
            let mut line = String::new();
            let _ = io::stdin().read_line(&mut line);

            let args: Vec<&str> = line.trim().split(",").collect();
            match args[0].to_string().trim() {
                "+1" | "1" => input.insert((args[1].to_string().clone(), args[2].to_string().clone())),
                "-1" => input.remove((args[1].to_string().clone(), args[2].to_string().clone())),
                _ => unimplemented!(),
            };


            time = time+1;
            input.advance_to(time);
            input.flush();

            while probe.less_than(&input.time()) { worker.step(); }
        }
    }).unwrap();
}