Function test_basic_usage

src/timely-util/src/reclock.rs:728–793

()

    #[mz_ore::test]
    fn test_basic_usage() {
        let as_of = Antichain::from_elem(IntoTime::minimum());
        harness(
            as_of,
            |worker, mut bindings, (mut data, data_cap), reclocked| {
                // Reclock offsets 1 and 3 to timestamp 1000
                bindings.update_at(Partitioned::minimum(), 0, Diff::ONE);
                bindings.update_at(Partitioned::minimum(), 1000, Diff::MINUS_ONE);
                for time in partitioned_frontier([(0, 4)]) {
                    bindings.update_at(time, 1000, Diff::ONE);
                }
                bindings.advance_to(1001);
                bindings.flush();
                data.activate().session(&data_cap).give_iterator(
                    vec![
                        (1, Partitioned::new_singleton(0, 1), Diff::ONE),
                        (1, Partitioned::new_singleton(0, 1), Diff::ONE),
                        (3, Partitioned::new_singleton(0, 3), Diff::ONE),
                    ]
                    .into_iter(),
                );

                step(worker);
                assert_eq!(
                    reclocked.try_recv(),
                    Ok(Event::Messages(
                        0u64,
                        vec![
                            (1, 1000, Diff::ONE),
                            (1, 1000, Diff::ONE),
                            (3, 1000, Diff::ONE)
                        ]
                    ))
                );
                assert_eq!(
                    reclocked.try_recv(),
                    Ok(Event::Progress(vec![(0, -1), (1000, 1)]))
                );

                // Reclock more messages for offsets 3 to the same timestamp
                data.activate().session(&data_cap).give_iterator(
                    vec![
                        (3, Partitioned::new_singleton(0, 3), Diff::ONE),
                        (3, Partitioned::new_singleton(0, 3), Diff::ONE),
                    ]
                    .into_iter(),
                );
                step(worker);
                assert_eq!(
                    reclocked.try_recv(),
                    Ok(Event::Messages(
                        1000u64,
                        vec![(3, 1000, Diff::ONE), (3, 1000, Diff::ONE)]
                    ))
                );

                // Drop the capability which should advance the reclocked frontier to 1001.
                drop(data_cap);