Query2 !

Hey ! Still obsessing on query2 :)

            let mut input = root.scoped::<u64, _, _>(move |builder| {
                let (input, stream) = builder.new_input::<((K, _), i32)>();
                let collections = Collection::new(stream);

                let group:Collection<_, (K,SaneF32)> = collections.group(|_,vs,o| {
                    let v:f32 = vs.map(|(sane,weight):(&SaneF32,i32)| sane.0*weight as f32).sum();
                    o.push( (SaneF32(v), 1) );
                });
                let count:Collection<_, (bool, u32)> = group.map(|(_,_):(K,SaneF32)| (true, 1)).consolidate();
                let count:Collection<_, (bool, i32)> = count.group(|_,vs,o| {
                    let c:i32 = vs.map(|(c,weight):(&u32,i32)| *c as i32 * weight).sum();
                    o.push((c, 1));
                });
                count.inspect(move |rec| result.store((rec.0).1 as usize, Relaxed));

            input
            });

            for (_i,chunk) in bibi.enumerate() {
                for item in chunk {
                    input.send(((item.0, SaneF32(item.1)), 1));
                }
            //    input.advance_to(i as u64 + 1);
            //    root.step();
            }

This is what I tried so far (with variation on the way to push the input as suggested byt the commented bit). Happy to say it is produces the correct result.

The problem I have is, it makes a gigantic memory consumption, compared to the timely impl, so it swaps and is quite slow.

diff-d:

timely-d:

This is for query 2A (X=8). Timely and the hand made impl made it under 500MB, differential activates 25GB...

Is it something that is expected ? Can i mitigate it somehow ?

Start by reproducing query 2A with X=8 using the shown differential dataflow code and compare its memory use with the timely implementation. Inspect the input advancement and root.step calls, including the commented lines, and determine whether they affect retention. Done means the memory difference is explained and a mitigation is identified.