Removing an unseen tuple actually inserts it

[stefAIno]

Hi everyone, I have a question about the behaviour of differential dataflow.
I have set up a query implementation and tried to play a little with the framework. I have noticed that trying to remove an unseen tuple will actually result in an insertion of the same. Trying to remove again the same tuple gives no result, same as trying to inserting it. If I try to make use of that tuple to fire rules attivation, it is actually present. Why is there such a behaviour?
Here an example:

rule:    r(X,Z) :- r(X,Y), r(Y,Z).

remove r(1,2)
r(("1", "2"), 0, 1)

remove r(1,2)
no out

add r(1,2)
no out

remove r(1,2)
no out

add r(2,3)
r(("2", "3"), 4, 1)
r(("1", "3"), 4, 1)

remove r(1,2)
no out

[frankmcsherry]

I'm not sure what your output corresponds to, I'm sorry. My best guess is that you are writing a datalog-alike, and using distinct, which has the property that it maps all non-zero occurrences to one occurrence. If that isn't what you are seeing, feel welcome to provide some more detail! Good luck.

[stefAIno]

Thanks for the fast reply @frankmcsherry .
To implement the rule I've stated I have used:

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()
        });

I make use of distinct() to avoid duplicate tuples. It is still not clear to me why is this affecting the removal behaviour, could you please elaborate more?

[frankmcsherry]

If you insert an inspect operator in the dataflow you will be able to see the updates that flow past, which may clarify things. Your retraction make the record accumulate to a -1 multiplicity, and distinct maps all non-zero multiplicities to a multiplicity of one.

…

Sent from my iPhone

On Nov 18, 2021, at 09:49, stefAIno ***@***.***> wrote:  Thanks for the fast reply @frankmcsherry . To implement the rule I've stated I have used: 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() }); I make use of distinct() to avoid duplicate tuples. It is still not clear to me why is this affecting the removal behaviour, could you please elaborate more? — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub, or unsubscribe. Triage notifications on the go with GitHub Mobile for iOS or Android.

[stefAIno]

Ok, I understand the effect of distinct() now. However, in the following program (reads in inputs like +1,a,b or -1,b,c) I do not use the distinct operator, and still see that removing two tuples which do represent a join match, actually do yield an update with multiplicity "1" ...which looks a bit odd, but is due to the multiset logic/semantic?

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();
}