---
source: hydroflow_datalog_core/src/lib.rs
expression: "prettyplease::unparse(&wrapped)"
---
fn main() {
    {
        #[allow(unused_qualifications)]
        {
            use hydroflow::{var_expr, var_args};
            let mut df = hydroflow::scheduled::graph::Hydroflow::new();
            df.__assign_meta_graph(
                "{\"nodes\":[{\"value\":null,\"version\":0},{\"value\":null,\"version\":2},{\"value\":{\"Operator\":\"unique :: < 'tick > ()\"},\"version\":1},{\"value\":{\"Operator\":\"tee ()\"},\"version\":1},{\"value\":null,\"version\":2},{\"value\":{\"Operator\":\"unique :: < 'tick > ()\"},\"version\":1},{\"value\":{\"Handoff\":{}},\"version\":3},{\"value\":null,\"version\":2},{\"value\":{\"Operator\":\"unique :: < 'tick > ()\"},\"version\":1},{\"value\":{\"Handoff\":{}},\"version\":3},{\"value\":{\"Handoff\":{}},\"version\":3},{\"value\":{\"Operator\":\"unique :: < 'tick > ()\"},\"version\":1},{\"value\":{\"Handoff\":{}},\"version\":3},{\"value\":{\"Handoff\":{}},\"version\":3},{\"value\":{\"Operator\":\"unique :: < 'tick > ()\"},\"version\":1},{\"value\":{\"Handoff\":{}},\"version\":3},{\"value\":{\"Handoff\":{}},\"version\":3},{\"value\":{\"Operator\":\"unique :: < 'tick > ()\"},\"version\":1},{\"value\":{\"Operator\":\"difference :: < 'tick , 'static > ()\"},\"version\":1},{\"value\":{\"Operator\":\"tee ()\"},\"version\":1},{\"value\":{\"Operator\":\"next_tick ()\"},\"version\":1},{\"value\":{\"Handoff\":{}},\"version\":3},{\"value\":{\"Operator\":\"unique :: < 'tick > ()\"},\"version\":1},{\"value\":{\"Operator\":\"difference :: < 'tick , 'static > ()\"},\"version\":1},{\"value\":{\"Operator\":\"tee ()\"},\"version\":1},{\"value\":{\"Operator\":\"next_tick ()\"},\"version\":1},{\"value\":{\"Operator\":\"source_stream (ints)\"},\"version\":1},{\"value\":{\"Operator\":\"for_each (| v | result . send (v) . unwrap ())\"},\"version\":1},{\"value\":{\"Operator\":\"for_each (| v | result2 . send (v) . unwrap ())\"},\"version\":1},{\"value\":{\"Operator\":\"for_each (| v | result3 . send (v) . unwrap ())\"},\"version\":1},{\"value\":{\"Operator\":\"for_each (| v | result4 . send (v) . unwrap ())\"},\"version\":1},{\"value\":{\"Operator\":\"persist ()\"},\"version\":1},{\"value\":{\"Operator\":\"for_each (| v | result5 . send (v) . unwrap ())\"},\"version\":1},{\"value\":{\"Operator\":\"map (| row : (_ , _ ,) | ((row . 0 , row . 1 ,) , ()))\"},\"version\":1},{\"value\":{\"Operator\":\"enumerate :: < 'tick > ()\"},\"version\":1},{\"value\":{\"Operator\":\"map (| (i , (g , a)) : (_ , ((_ , _ ,) , _)) | (g . 0 , g . 1 , i ,))\"},\"version\":1},{\"value\":{\"Operator\":\"map (| row : (_ , _ ,) | ((row . 0 ,) , ((row . 1) ,)))\"},\"version\":1},{\"value\":{\"Operator\":\"group_by :: < 'tick , (_ ,) , (Option < _ > ,) > (| | (None ,) , | old : & mut (Option < _ > ,) , val : (_ ,) | { old . 0 = if let Some (prev) = old . 0 . take () { Some ({ let prev : (hydroflow :: rustc_hash :: FxHashSet < _ > , _) = prev ; let mut set : hydroflow :: rustc_hash :: FxHashSet < _ > = prev . 0 ; if set . insert (val . 0) { (set , prev . 1 + 1) } else { (set , prev . 1) } }) } else { Some ({ let mut set = hydroflow :: rustc_hash :: FxHashSet :: < _ > :: default () ; set . insert (val . 0) ; (set , 1) }) } ; })\"},\"version\":1},{\"value\":{\"Operator\":\"enumerate :: < 'tick > ()\"},\"version\":1},{\"value\":{\"Operator\":\"map (| (i , (g , a)) : (_ , ((_ ,) , _)) | (g . 0 , a . 0 . unwrap () . 1 , i ,))\"},\"version\":1},{\"value\":{\"Operator\":\"map (| row : (_ , _ ,) | ((row . 0 , row . 1 ,) , ()))\"},\"version\":1},{\"value\":{\"Operator\":\"map (| (g , a) : ((_ , _ ,) , _) | (g . 0 , g . 1 ,))\"},\"version\":1},{\"value\":{\"Operator\":\"map (| row : (_ , _ ,) | ((row . 0 , row . 1 ,) , ()))\"},\"version\":1},{\"value\":{\"Operator\":\"enumerate :: < 'static > ()\"},\"version\":1},{\"value\":{\"Operator\":\"map (| (i , (g , a)) : (_ , ((_ , _ ,) , _)) | (g . 0 , g . 1 , i ,))\"},\"version\":1},{\"value\":{\"Operator\":\"persist ()\"},\"version\":1},{\"value\":{\"Operator\":\"map (| row : (_ , _ ,) | ((row . 0 ,) , ((row . 1) ,)))\"},\"version\":1},{\"value\":{\"Operator\":\"group_by :: < 'static , (_ ,) , (Option < _ > ,) > (| | (None ,) , | old : & mut (Option < _ > ,) , val : (_ ,) | { old . 0 = if let Some (prev) = old . 0 . take () { Some ({ let prev : (hydroflow :: rustc_hash :: FxHashSet < _ > , _) = prev ; let mut set : hydroflow :: rustc_hash :: FxHashSet < _ > = prev . 0 ; if set . insert (val . 0) { (set , prev . 1 + 1) } else { (set , prev . 1) } }) } else { Some ({ let mut set = hydroflow :: rustc_hash :: FxHashSet :: < _ > :: default () ; set . insert (val . 0) ; (set , 1) }) } ; })\"},\"version\":1},{\"value\":{\"Operator\":\"enumerate :: < 'tick > ()\"},\"version\":1},{\"value\":{\"Operator\":\"map (| (i , (g , a)) : (_ , ((_ ,) , _)) | (g . 0 , a . 0 . unwrap () . 1 , i ,))\"},\"version\":1},{\"value\":{\"Operator\":\"map (| row : (_ , _ ,) | ((row . 0 , row . 1 ,) , ()))\"},\"version\":1},{\"value\":{\"Operator\":\"enumerate :: < 'static > ()\"},\"version\":1},{\"value\":{\"Operator\":\"map (| (i , (g , a)) : (_ , ((_ , _ ,) , _)) | (g . 0 , g . 1 , i ,))\"},\"version\":1}],\"graph\":[{\"value\":null,\"version\":0},{\"value\":[{\"idx\":26,\"version\":1},{\"idx\":2,\"version\":1}],\"version\":3},{\"value\":[{\"idx\":2,\"version\":1},{\"idx\":3,\"version\":1}],\"version\":1},{\"value\":[{\"idx\":35,\"version\":1},{\"idx\":5,\"version\":1}],\"version\":3},{\"value\":[{\"idx\":6,\"version\":3},{\"idx\":25,\"version\":1}],\"version\":3},{\"value\":[{\"idx\":39,\"version\":1},{\"idx\":8,\"version\":1}],\"version\":3},{\"value\":[{\"idx\":9,\"version\":3},{\"idx\":23,\"version\":1}],\"version\":3},{\"value\":[{\"idx\":45,\"version\":1},{\"idx\":11,\"version\":1}],\"version\":3},{\"value\":[{\"idx\":12,\"version\":3},{\"idx\":20,\"version\":1}],\"version\":3},{\"value\":[{\"idx\":49,\"version\":1},{\"idx\":14,\"version\":1}],\"version\":3},{\"value\":[{\"idx\":15,\"version\":3},{\"idx\":18,\"version\":1}],\"version\":3},{\"value\":[{\"idx\":52,\"version\":1},{\"idx\":17,\"version\":1}],\"version\":3},{\"value\":[{\"idx\":18,\"version\":1},{\"idx\":19,\"version\":1}],\"version\":1},{\"value\":[{\"idx\":17,\"version\":1},{\"idx\":18,\"version\":1}],\"version\":1},{\"value\":[{\"idx\":20,\"version\":1},{\"idx\":15,\"version\":3}],\"version\":3},{\"value\":[{\"idx\":19,\"version\":1},{\"idx\":12,\"version\":3}],\"version\":3},{\"value\":[{\"idx\":41,\"version\":1},{\"idx\":22,\"version\":1}],\"version\":3},{\"value\":[{\"idx\":23,\"version\":1},{\"idx\":24,\"version\":1}],\"version\":1},{\"value\":[{\"idx\":22,\"version\":1},{\"idx\":23,\"version\":1}],\"version\":1},{\"value\":[{\"idx\":25,\"version\":1},{\"idx\":9,\"version\":3}],\"version\":3},{\"value\":[{\"idx\":24,\"version\":1},{\"idx\":6,\"version\":3}],\"version\":3},{\"value\":null,\"version\":2},{\"value\":[{\"idx\":5,\"version\":1},{\"idx\":27,\"version\":1}],\"version\":3},{\"value\":[{\"idx\":8,\"version\":1},{\"idx\":28,\"version\":1}],\"version\":3},{\"value\":[{\"idx\":11,\"version\":1},{\"idx\":29,\"version\":1}],\"version\":3},{\"value\":[{\"idx\":14,\"version\":1},{\"idx\":30,\"version\":1}],\"version\":3},{\"value\":[{\"idx\":31,\"version\":1},{\"idx\":32,\"version\":1}],\"version\":1},{\"value\":[{\"idx\":19,\"version\":1},{\"idx\":31,\"version\":1}],\"version\":1},{\"value\":null,\"version\":2},{\"value\":[{\"idx\":34,\"version\":1},{\"idx\":35,\"version\":1}],\"version\":1},{\"value\":[{\"idx\":33,\"version\":1},{\"idx\":34,\"version\":1}],\"version\":1},{\"value\":[{\"idx\":3,\"version\":1},{\"idx\":33,\"version\":1}],\"version\":1},{\"value\":null,\"version\":2},{\"value\":[{\"idx\":38,\"version\":1},{\"idx\":39,\"version\":1}],\"version\":1},{\"value\":[{\"idx\":37,\"version\":1},{\"idx\":38,\"version\":1}],\"version\":1},{\"value\":[{\"idx\":36,\"version\":1},{\"idx\":21,\"version\":3}],\"version\":3},{\"value\":[{\"idx\":3,\"version\":1},{\"idx\":36,\"version\":1}],\"version\":1},{\"value\":[{\"idx\":21,\"version\":3},{\"idx\":37,\"version\":1}],\"version\":3},{\"value\":[{\"idx\":40,\"version\":1},{\"idx\":16,\"version\":3}],\"version\":3},{\"value\":[{\"idx\":3,\"version\":1},{\"idx\":40,\"version\":1}],\"version\":1},{\"value\":[{\"idx\":10,\"version\":3},{\"idx\":51,\"version\":1}],\"version\":3},{\"value\":[{\"idx\":44,\"version\":1},{\"idx\":45,\"version\":1}],\"version\":1},{\"value\":[{\"idx\":43,\"version\":1},{\"idx\":44,\"version\":1}],\"version\":1},{\"value\":[{\"idx\":42,\"version\":1},{\"idx\":43,\"version\":1}],\"version\":1},{\"value\":[{\"idx\":24,\"version\":1},{\"idx\":42,\"version\":1}],\"version\":1},{\"value\":[{\"idx\":13,\"version\":3},{\"idx\":47,\"version\":1}],\"version\":3},{\"value\":[{\"idx\":48,\"version\":1},{\"idx\":49,\"version\":1}],\"version\":1},{\"value\":[{\"idx\":47,\"version\":1},{\"idx\":48,\"version\":1}],\"version\":1},{\"value\":[{\"idx\":46,\"version\":1},{\"idx\":13,\"version\":3}],\"version\":3},{\"value\":[{\"idx\":24,\"version\":1},{\"idx\":46,\"version\":1}],\"version\":1},{\"value\":[{\"idx\":16,\"version\":3},{\"idx\":41,\"version\":1}],\"version\":3},{\"value\":[{\"idx\":51,\"version\":1},{\"idx\":52,\"version\":1}],\"version\":1},{\"value\":[{\"idx\":50,\"version\":1},{\"idx\":10,\"version\":3}],\"version\":3},{\"value\":[{\"idx\":24,\"version\":1},{\"idx\":50,\"version\":1}],\"version\":1}],\"ports\":[{\"value\":null,\"version\":0},{\"value\":[\"Elided\",\"Elided\"],\"version\":3},{\"value\":[\"Elided\",\"Elided\"],\"version\":1},{\"value\":[\"Elided\",\"Elided\"],\"version\":3},{\"value\":[\"Elided\",\"Elided\"],\"version\":3},{\"value\":[\"Elided\",\"Elided\"],\"version\":3},{\"value\":[\"Elided\",{\"Path\":\"neg\"}],\"version\":3},{\"value\":[\"Elided\",\"Elided\"],\"version\":3},{\"value\":[\"Elided\",\"Elided\"],\"version\":3},{\"value\":[\"Elided\",\"Elided\"],\"version\":3},{\"value\":[\"Elided\",{\"Path\":\"neg\"}],\"version\":3},{\"value\":[\"Elided\",\"Elided\"],\"version\":3},{\"value\":[\"Elided\",\"Elided\"],\"version\":1},{\"value\":[\"Elided\",{\"Path\":\"pos\"}],\"version\":1},{\"value\":[\"Elided\",\"Elided\"],\"version\":3},{\"value\":[\"Elided\",\"Elided\"],\"version\":3},{\"value\":[\"Elided\",\"Elided\"],\"version\":3},{\"value\":[\"Elided\",\"Elided\"],\"version\":1},{\"value\":[\"Elided\",{\"Path\":\"pos\"}],\"version\":1},{\"value\":[\"Elided\",\"Elided\"],\"version\":3},{\"value\":[\"Elided\",\"Elided\"],\"version\":3},{\"value\":null,\"version\":0},{\"value\":[\"Elided\",\"Elided\"],\"version\":3},{\"value\":[\"Elided\",\"Elided\"],\"version\":3},{\"value\":[\"Elided\",\"Elided\"],\"version\":3},{\"value\":[\"Elided\",\"Elided\"],\"version\":3},{\"value\":[\"Elided\",\"Elided\"],\"version\":1},{\"value\":[{\"Int\":\"0\"},\"Elided\"],\"version\":1},{\"value\":null,\"version\":0},{\"value\":[\"Elided\",\"Elided\"],\"version\":1},{\"value\":[\"Elided\",\"Elided\"],\"version\":1},{\"value\":[{\"Int\":\"0\"},\"Elided\"],\"version\":1},{\"value\":null,\"version\":0},{\"value\":[\"Elided\",\"Elided\"],\"version\":1},{\"value\":[\"Elided\",\"Elided\"],\"version\":1},{\"value\":[\"Elided\",\"Elided\"],\"version\":3},{\"value\":[{\"Int\":\"1\"},\"Elided\"],\"version\":1},{\"value\":[\"Elided\",\"Elided\"],\"version\":3},{\"value\":[\"Elided\",\"Elided\"],\"version\":3},{\"value\":[{\"Int\":\"2\"},\"Elided\"],\"version\":1},{\"value\":[\"Elided\",\"Elided\"],\"version\":3},{\"value\":[\"Elided\",\"Elided\"],\"version\":1},{\"value\":[\"Elided\",\"Elided\"],\"version\":1},{\"value\":[\"Elided\",\"Elided\"],\"version\":1},{\"value\":[{\"Int\":\"0\"},\"Elided\"],\"version\":1},{\"value\":[\"Elided\",\"Elided\"],\"version\":3},{\"value\":[\"Elided\",\"Elided\"],\"version\":1},{\"value\":[\"Elided\",\"Elided\"],\"version\":1},{\"value\":[\"Elided\",\"Elided\"],\"version\":3},{\"value\":[{\"Int\":\"1\"},\"Elided\"],\"version\":1},{\"value\":[\"Elided\",\"Elided\"],\"version\":3},{\"value\":[\"Elided\",\"Elided\"],\"version\":1},{\"value\":[\"Elided\",\"Elided\"],\"version\":3},{\"value\":[{\"Int\":\"2\"},\"Elided\"],\"version\":1}],\"node_subgraph\":[{\"value\":null,\"version\":0},{\"value\":null,\"version\":0},{\"value\":{\"idx\":3,\"version\":1},\"version\":1},{\"value\":{\"idx\":3,\"version\":1},\"version\":1},{\"value\":null,\"version\":0},{\"value\":{\"idx\":3,\"version\":1},\"version\":1},{\"value\":null,\"version\":0},{\"value\":null,\"version\":0},{\"value\":{\"idx\":4,\"version\":1},\"version\":1},{\"value\":null,\"version\":0},{\"value\":null,\"version\":0},{\"value\":{\"idx\":5,\"version\":1},\"version\":1},{\"value\":null,\"version\":0},{\"value\":null,\"version\":0},{\"value\":{\"idx\":6,\"version\":1},\"version\":1},{\"value\":null,\"version\":0},{\"value\":null,\"version\":0},{\"value\":{\"idx\":7,\"version\":1},\"version\":1},{\"value\":{\"idx\":7,\"version\":1},\"version\":1},{\"value\":{\"idx\":7,\"version\":1},\"version\":1},{\"value\":{\"idx\":1,\"version\":1},\"version\":1},{\"value\":null,\"version\":0},{\"value\":{\"idx\":5,\"version\":1},\"version\":1},{\"value\":{\"idx\":5,\"version\":1},\"version\":1},{\"value\":{\"idx\":5,\"version\":1},\"version\":1},{\"value\":{\"idx\":2,\"version\":1},\"version\":1},{\"value\":{\"idx\":3,\"version\":1},\"version\":1},{\"value\":{\"idx\":3,\"version\":1},\"version\":1},{\"value\":{\"idx\":4,\"version\":1},\"version\":1},{\"value\":{\"idx\":5,\"version\":1},\"version\":1},{\"value\":{\"idx\":6,\"version\":1},\"version\":1},{\"value\":{\"idx\":7,\"version\":1},\"version\":1},{\"value\":{\"idx\":7,\"version\":1},\"version\":1},{\"value\":{\"idx\":3,\"version\":1},\"version\":1},{\"value\":{\"idx\":3,\"version\":1},\"version\":1},{\"value\":{\"idx\":3,\"version\":1},\"version\":1},{\"value\":{\"idx\":3,\"version\":1},\"version\":1},{\"value\":{\"idx\":4,\"version\":1},\"version\":1},{\"value\":{\"idx\":4,\"version\":1},\"version\":1},{\"value\":{\"idx\":4,\"version\":1},\"version\":1},{\"value\":{\"idx\":3,\"version\":1},\"version\":1},{\"value\":{\"idx\":5,\"version\":1},\"version\":1},{\"value\":{\"idx\":5,\"version\":1},\"version\":1},{\"value\":{\"idx\":5,\"version\":1},\"version\":1},{\"value\":{\"idx\":5,\"version\":1},\"version\":1},{\"value\":{\"idx\":5,\"version\":1},\"version\":1},{\"value\":{\"idx\":5,\"version\":1},\"version\":1},{\"value\":{\"idx\":6,\"version\":1},\"version\":1},{\"value\":{\"idx\":6,\"version\":1},\"version\":1},{\"value\":{\"idx\":6,\"version\":1},\"version\":1},{\"value\":{\"idx\":5,\"version\":1},\"version\":1},{\"value\":{\"idx\":7,\"version\":1},\"version\":1},{\"value\":{\"idx\":7,\"version\":1},\"version\":1}],\"subgraph_nodes\":[{\"value\":null,\"version\":0},{\"value\":[{\"idx\":20,\"version\":1}],\"version\":1},{\"value\":[{\"idx\":25,\"version\":1}],\"version\":1},{\"value\":[{\"idx\":26,\"version\":1},{\"idx\":2,\"version\":1},{\"idx\":3,\"version\":1},{\"idx\":33,\"version\":1},{\"idx\":34,\"version\":1},{\"idx\":35,\"version\":1},{\"idx\":5,\"version\":1},{\"idx\":27,\"version\":1},{\"idx\":36,\"version\":1},{\"idx\":40,\"version\":1}],\"version\":1},{\"value\":[{\"idx\":37,\"version\":1},{\"idx\":38,\"version\":1},{\"idx\":39,\"version\":1},{\"idx\":8,\"version\":1},{\"idx\":28,\"version\":1}],\"version\":1},{\"value\":[{\"idx\":41,\"version\":1},{\"idx\":22,\"version\":1},{\"idx\":23,\"version\":1},{\"idx\":24,\"version\":1},{\"idx\":42,\"version\":1},{\"idx\":43,\"version\":1},{\"idx\":44,\"version\":1},{\"idx\":45,\"version\":1},{\"idx\":11,\"version\":1},{\"idx\":29,\"version\":1},{\"idx\":46,\"version\":1},{\"idx\":50,\"version\":1}],\"version\":1},{\"value\":[{\"idx\":47,\"version\":1},{\"idx\":48,\"version\":1},{\"idx\":49,\"version\":1},{\"idx\":14,\"version\":1},{\"idx\":30,\"version\":1}],\"version\":1},{\"value\":[{\"idx\":51,\"version\":1},{\"idx\":52,\"version\":1},{\"idx\":17,\"version\":1},{\"idx\":18,\"version\":1},{\"idx\":19,\"version\":1},{\"idx\":31,\"version\":1},{\"idx\":32,\"version\":1}],\"version\":1}],\"subgraph_stratum\":[{\"value\":null,\"version\":0},{\"value\":0,\"version\":1},{\"value\":0,\"version\":1},{\"value\":0,\"version\":1},{\"value\":1,\"version\":1},{\"value\":1,\"version\":1},{\"value\":2,\"version\":1},{\"value\":1,\"version\":1}],\"node_varnames\":[{\"value\":null,\"version\":0},{\"value\":null,\"version\":0},{\"value\":\"ints_insert\",\"version\":1},{\"value\":\"ints\",\"version\":1},{\"value\":null,\"version\":0},{\"value\":\"result_insert\",\"version\":1},{\"value\":null,\"version\":0},{\"value\":null,\"version\":0},{\"value\":\"result2_insert\",\"version\":1},{\"value\":null,\"version\":0},{\"value\":null,\"version\":0},{\"value\":\"result3_insert\",\"version\":1},{\"value\":null,\"version\":0},{\"value\":null,\"version\":0},{\"value\":\"result4_insert\",\"version\":1},{\"value\":null,\"version\":0},{\"value\":null,\"version\":0},{\"value\":\"result5_insert\",\"version\":1},{\"value\":\"result5\",\"version\":1},{\"value\":\"result5\",\"version\":1},{\"value\":null,\"version\":0},{\"value\":null,\"version\":0},{\"value\":\"ints_persisted_insert\",\"version\":1},{\"value\":\"ints_persisted\",\"version\":1},{\"value\":\"ints_persisted\",\"version\":1}]}",
            );
            df.__assign_diagnostics("[]");
            let (hoff_6v3_send, hoff_6v3_recv) = df
                .make_edge::<
                    _,
                    hydroflow::scheduled::handoff::VecHandoff<_>,
                >("handoff GraphNodeId(6v3)");
            let (hoff_9v3_send, hoff_9v3_recv) = df
                .make_edge::<
                    _,
                    hydroflow::scheduled::handoff::VecHandoff<_>,
                >("handoff GraphNodeId(9v3)");
            let (hoff_10v3_send, hoff_10v3_recv) = df
                .make_edge::<
                    _,
                    hydroflow::scheduled::handoff::VecHandoff<_>,
                >("handoff GraphNodeId(10v3)");
            let (hoff_12v3_send, hoff_12v3_recv) = df
                .make_edge::<
                    _,
                    hydroflow::scheduled::handoff::VecHandoff<_>,
                >("handoff GraphNodeId(12v3)");
            let (hoff_13v3_send, hoff_13v3_recv) = df
                .make_edge::<
                    _,
                    hydroflow::scheduled::handoff::VecHandoff<_>,
                >("handoff GraphNodeId(13v3)");
            let (hoff_15v3_send, hoff_15v3_recv) = df
                .make_edge::<
                    _,
                    hydroflow::scheduled::handoff::VecHandoff<_>,
                >("handoff GraphNodeId(15v3)");
            let (hoff_16v3_send, hoff_16v3_recv) = df
                .make_edge::<
                    _,
                    hydroflow::scheduled::handoff::VecHandoff<_>,
                >("handoff GraphNodeId(16v3)");
            let (hoff_21v3_send, hoff_21v3_recv) = df
                .make_edge::<
                    _,
                    hydroflow::scheduled::handoff::VecHandoff<_>,
                >("handoff GraphNodeId(21v3)");
            let mut sg_3v1_node_26v1_stream = {
                #[inline(always)]
                fn check_stream<
                    Stream: hydroflow::futures::stream::Stream<Item = Item>,
                    Item,
                >(
                    stream: Stream,
                ) -> ::std::pin::Pin<
                    ::std::boxed::Box<
                        impl hydroflow::futures::stream::Stream<Item = Item>,
                    >,
                > {
                    ::std::boxed::Box::pin(stream)
                }
                check_stream(ints)
            };
            let sg_3v1_node_2v1_uniquedata = df
                .add_state(
                    ::std::cell::RefCell::new(
                        hydroflow::lang::monotonic_map::MonotonicMap::<
                            _,
                            hydroflow::rustc_hash::FxHashSet<_>,
                        >::default(),
                    ),
                );
            let sg_3v1_node_5v1_uniquedata = df
                .add_state(
                    ::std::cell::RefCell::new(
                        hydroflow::lang::monotonic_map::MonotonicMap::<
                            _,
                            hydroflow::rustc_hash::FxHashSet<_>,
                        >::default(),
                    ),
                );
            let sg_3v1_node_34v1_counterdata = df
                .add_state(
                    ::std::cell::RefCell::new(
                        hydroflow::lang::monotonic_map::MonotonicMap::new_init(0..),
                    ),
                );
            df.add_subgraph_stratified(
                "Subgraph GraphSubgraphId(3v1)",
                0,
                var_expr!(),
                var_expr!(hoff_16v3_send, hoff_21v3_send),
                move |context, var_args!(), var_args!(hoff_16v3_send, hoff_21v3_send)| {
                    let hoff_16v3_send = hydroflow::pusherator::for_each::ForEach::new(|
                        v|
                    {
                        hoff_16v3_send.give(Some(v));
                    });
                    let hoff_21v3_send = hydroflow::pusherator::for_each::ForEach::new(|
                        v|
                    {
                        hoff_21v3_send.give(Some(v));
                    });
                    let op_26v1 = std::iter::from_fn(|| {
                        match hydroflow::futures::stream::Stream::poll_next(
                            sg_3v1_node_26v1_stream.as_mut(),
                            &mut std::task::Context::from_waker(&context.waker()),
                        ) {
                            std::task::Poll::Ready(maybe) => maybe,
                            std::task::Poll::Pending => None,
                        }
                    });
                    let op_26v1 = {
                        #[allow(non_snake_case)]
                        #[inline(always)]
                        pub fn op_26v1__source_stream__loc_unknown_start_2_25_end_2_44<
                            Item,
                            Input: ::std::iter::Iterator<Item = Item>,
                        >(input: Input) -> impl ::std::iter::Iterator<Item = Item> {
                            struct Pull<
                                Item,
                                Input: ::std::iter::Iterator<Item = Item>,
                            > {
                                inner: Input,
                            }
                            impl<
                                Item,
                                Input: ::std::iter::Iterator<Item = Item>,
                            > Iterator for Pull<Item, Input> {
                                type Item = Item;
                                #[inline(always)]
                                fn next(&mut self) -> Option<Self::Item> {
                                    self.inner.next()
                                }
                                #[inline(always)]
                                fn size_hint(&self) -> (usize, Option<usize>) {
                                    self.inner.size_hint()
                                }
                            }
                            Pull { inner: input }
                        }
                        op_26v1__source_stream__loc_unknown_start_2_25_end_2_44(op_26v1)
                    };
                    let op_2v1 = op_26v1
                        .filter(|item| {
                            let mut borrow = context
                                .state_ref(sg_3v1_node_2v1_uniquedata)
                                .borrow_mut();
                            let set = borrow
                                .get_mut_clear((
                                    context.current_tick(),
                                    context.current_stratum(),
                                ));
                            if !set.contains(item) {
                                set.insert(::std::clone::Clone::clone(item));
                                true
                            } else {
                                false
                            }
                        });
                    let op_2v1 = {
                        #[allow(non_snake_case)]
                        #[inline(always)]
                        pub fn op_2v1__unique__loc_unknown_start_2_19_end_2_23<
                            Item,
                            Input: ::std::iter::Iterator<Item = Item>,
                        >(input: Input) -> impl ::std::iter::Iterator<Item = Item> {
                            struct Pull<
                                Item,
                                Input: ::std::iter::Iterator<Item = Item>,
                            > {
                                inner: Input,
                            }
                            impl<
                                Item,
                                Input: ::std::iter::Iterator<Item = Item>,
                            > Iterator for Pull<Item, Input> {
                                type Item = Item;
                                #[inline(always)]
                                fn next(&mut self) -> Option<Self::Item> {
                                    self.inner.next()
                                }
                                #[inline(always)]
                                fn size_hint(&self) -> (usize, Option<usize>) {
                                    self.inner.size_hint()
                                }
                            }
                            Pull { inner: input }
                        }
                        op_2v1__unique__loc_unknown_start_2_19_end_2_23(op_2v1)
                    };
                    let op_40v1 = hydroflow::pusherator::map::Map::new(
                        |row: (_, _)| ((row.0, row.1), ()),
                        hoff_16v3_send,
                    );
                    let op_40v1 = {
                        #[allow(non_snake_case)]
                        #[inline(always)]
                        pub fn op_40v1__map__loc_unknown_start_1_0_end_1_0<
                            Item,
                            Input: hydroflow::pusherator::Pusherator<Item = Item>,
                        >(
                            input: Input,
                        ) -> impl hydroflow::pusherator::Pusherator<Item = Item> {
                            struct Push<
                                Item,
                                Input: hydroflow::pusherator::Pusherator<Item = Item>,
                            > {
                                inner: Input,
                            }
                            impl<
                                Item,
                                Input: hydroflow::pusherator::Pusherator<Item = Item>,
                            > hydroflow::pusherator::Pusherator for Push<Item, Input> {
                                type Item = Item;
                                #[inline(always)]
                                fn give(&mut self, item: Self::Item) {
                                    self.inner.give(item)
                                }
                            }
                            Push { inner: input }
                        }
                        op_40v1__map__loc_unknown_start_1_0_end_1_0(op_40v1)
                    };
                    let op_36v1 = hydroflow::pusherator::map::Map::new(
                        |row: (_, _)| ((row.0,), ((row.1),)),
                        hoff_21v3_send,
                    );
                    let op_36v1 = {
                        #[allow(non_snake_case)]
                        #[inline(always)]
                        pub fn op_36v1__map__loc_unknown_start_1_0_end_1_0<
                            Item,
                            Input: hydroflow::pusherator::Pusherator<Item = Item>,
                        >(
                            input: Input,
                        ) -> impl hydroflow::pusherator::Pusherator<Item = Item> {
                            struct Push<
                                Item,
                                Input: hydroflow::pusherator::Pusherator<Item = Item>,
                            > {
                                inner: Input,
                            }
                            impl<
                                Item,
                                Input: hydroflow::pusherator::Pusherator<Item = Item>,
                            > hydroflow::pusherator::Pusherator for Push<Item, Input> {
                                type Item = Item;
                                #[inline(always)]
                                fn give(&mut self, item: Self::Item) {
                                    self.inner.give(item)
                                }
                            }
                            Push { inner: input }
                        }
                        op_36v1__map__loc_unknown_start_1_0_end_1_0(op_36v1)
                    };
                    let op_27v1 = hydroflow::pusherator::for_each::ForEach::new(|v| {
                        result.send(v).unwrap()
                    });
                    let op_27v1 = {
                        #[allow(non_snake_case)]
                        #[inline(always)]
                        pub fn op_27v1__for_each__loc_unknown_start_4_28_end_4_65<
                            Item,
                            Input: hydroflow::pusherator::Pusherator<Item = Item>,
                        >(
                            input: Input,
                        ) -> impl hydroflow::pusherator::Pusherator<Item = Item> {
                            struct Push<
                                Item,
                                Input: hydroflow::pusherator::Pusherator<Item = Item>,
                            > {
                                inner: Input,
                            }
                            impl<
                                Item,
                                Input: hydroflow::pusherator::Pusherator<Item = Item>,
                            > hydroflow::pusherator::Pusherator for Push<Item, Input> {
                                type Item = Item;
                                #[inline(always)]
                                fn give(&mut self, item: Self::Item) {
                                    self.inner.give(item)
                                }
                            }
                            Push { inner: input }
                        }
                        op_27v1__for_each__loc_unknown_start_4_28_end_4_65(op_27v1)
                    };
                    let op_5v1 = hydroflow::pusherator::filter::Filter::new(
                        |item| {
                            let mut borrow = context
                                .state_ref(sg_3v1_node_5v1_uniquedata)
                                .borrow_mut();
                            let set = borrow
                                .get_mut_clear((
                                    context.current_tick(),
                                    context.current_stratum(),
                                ));
                            if !set.contains(item) {
                                set.insert(::std::clone::Clone::clone(item));
                                true
                            } else {
                                false
                            }
                        },
                        op_27v1,
                    );
                    let op_5v1 = {
                        #[allow(non_snake_case)]
                        #[inline(always)]
                        pub fn op_5v1__unique__loc_unknown_start_4_20_end_4_26<
                            Item,
                            Input: hydroflow::pusherator::Pusherator<Item = Item>,
                        >(
                            input: Input,
                        ) -> impl hydroflow::pusherator::Pusherator<Item = Item> {
                            struct Push<
                                Item,
                                Input: hydroflow::pusherator::Pusherator<Item = Item>,
                            > {
                                inner: Input,
                            }
                            impl<
                                Item,
                                Input: hydroflow::pusherator::Pusherator<Item = Item>,
                            > hydroflow::pusherator::Pusherator for Push<Item, Input> {
                                type Item = Item;
                                #[inline(always)]
                                fn give(&mut self, item: Self::Item) {
                                    self.inner.give(item)
                                }
                            }
                            Push { inner: input }
                        }
                        op_5v1__unique__loc_unknown_start_4_20_end_4_26(op_5v1)
                    };
                    let op_35v1 = hydroflow::pusherator::map::Map::new(
                        |(i, (g, a)): (_, ((_, _), _))| (g.0, g.1, i),
                        op_5v1,
                    );
                    let op_35v1 = {
                        #[allow(non_snake_case)]
                        #[inline(always)]
                        pub fn op_35v1__map__loc_unknown_start_1_0_end_1_0<
                            Item,
                            Input: hydroflow::pusherator::Pusherator<Item = Item>,
                        >(
                            input: Input,
                        ) -> impl hydroflow::pusherator::Pusherator<Item = Item> {
                            struct Push<
                                Item,
                                Input: hydroflow::pusherator::Pusherator<Item = Item>,
                            > {
                                inner: Input,
                            }
                            impl<
                                Item,
                                Input: hydroflow::pusherator::Pusherator<Item = Item>,
                            > hydroflow::pusherator::Pusherator for Push<Item, Input> {
                                type Item = Item;
                                #[inline(always)]
                                fn give(&mut self, item: Self::Item) {
                                    self.inner.give(item)
                                }
                            }
                            Push { inner: input }
                        }
                        op_35v1__map__loc_unknown_start_1_0_end_1_0(op_35v1)
                    };
                    let op_34v1 = hydroflow::pusherator::map::Map::new(
                        |item| {
                            let mut borrow = context
                                .state_ref(sg_3v1_node_34v1_counterdata)
                                .borrow_mut();
                            let counter = borrow
                                .get_mut_with(
                                    (context.current_tick(), context.current_stratum()),
                                    || 0..,
                                );
                            (counter.next().unwrap(), item)
                        },
                        op_35v1,
                    );
                    let op_34v1 = {
                        #[allow(non_snake_case)]
                        #[inline(always)]
                        pub fn op_34v1__enumerate__loc_unknown_start_1_0_end_1_0<
                            Item,
                            Input: hydroflow::pusherator::Pusherator<Item = Item>,
                        >(
                            input: Input,
                        ) -> impl hydroflow::pusherator::Pusherator<Item = Item> {
                            struct Push<
                                Item,
                                Input: hydroflow::pusherator::Pusherator<Item = Item>,
                            > {
                                inner: Input,
                            }
                            impl<
                                Item,
                                Input: hydroflow::pusherator::Pusherator<Item = Item>,
                            > hydroflow::pusherator::Pusherator for Push<Item, Input> {
                                type Item = Item;
                                #[inline(always)]
                                fn give(&mut self, item: Self::Item) {
                                    self.inner.give(item)
                                }
                            }
                            Push { inner: input }
                        }
                        op_34v1__enumerate__loc_unknown_start_1_0_end_1_0(op_34v1)
                    };
                    let op_33v1 = hydroflow::pusherator::map::Map::new(
                        |row: (_, _)| ((row.0, row.1), ()),
                        op_34v1,
                    );
                    let op_33v1 = {
                        #[allow(non_snake_case)]
                        #[inline(always)]
                        pub fn op_33v1__map__loc_unknown_start_1_0_end_1_0<
                            Item,
                            Input: hydroflow::pusherator::Pusherator<Item = Item>,
                        >(
                            input: Input,
                        ) -> impl hydroflow::pusherator::Pusherator<Item = Item> {
                            struct Push<
                                Item,
                                Input: hydroflow::pusherator::Pusherator<Item = Item>,
                            > {
                                inner: Input,
                            }
                            impl<
                                Item,
                                Input: hydroflow::pusherator::Pusherator<Item = Item>,
                            > hydroflow::pusherator::Pusherator for Push<Item, Input> {
                                type Item = Item;
                                #[inline(always)]
                                fn give(&mut self, item: Self::Item) {
                                    self.inner.give(item)
                                }
                            }
                            Push { inner: input }
                        }
                        op_33v1__map__loc_unknown_start_1_0_end_1_0(op_33v1)
                    };
                    let op_3v1 = hydroflow::pusherator::tee::Tee::new(
                        op_33v1,
                        hydroflow::pusherator::tee::Tee::new(op_36v1, op_40v1),
                    );
                    let op_3v1 = {
                        #[allow(non_snake_case)]
                        #[inline(always)]
                        pub fn op_3v1__tee__loc_unknown_start_2_19_end_2_23<
                            Item,
                            Input: hydroflow::pusherator::Pusherator<Item = Item>,
                        >(
                            input: Input,
                        ) -> impl hydroflow::pusherator::Pusherator<Item = Item> {
                            struct Push<
                                Item,
                                Input: hydroflow::pusherator::Pusherator<Item = Item>,
                            > {
                                inner: Input,
                            }
                            impl<
                                Item,
                                Input: hydroflow::pusherator::Pusherator<Item = Item>,
                            > hydroflow::pusherator::Pusherator for Push<Item, Input> {
                                type Item = Item;
                                #[inline(always)]
                                fn give(&mut self, item: Self::Item) {
                                    self.inner.give(item)
                                }
                            }
                            Push { inner: input }
                        }
                        op_3v1__tee__loc_unknown_start_2_19_end_2_23(op_3v1)
                    };
                    #[inline(always)]
                    fn check_pivot_run<
                        Pull: ::std::iter::Iterator<Item = Item>,
                        Push: hydroflow::pusherator::Pusherator<Item = Item>,
                        Item,
                    >(pull: Pull, push: Push) {
                        hydroflow::pusherator::pivot::Pivot::new(pull, push).run();
                    }
                    check_pivot_run(op_2v1, op_3v1);
                },
            );
            df.add_subgraph_stratified(
                "Subgraph GraphSubgraphId(1v1)",
                0,
                var_expr!(hoff_12v3_recv),
                var_expr!(hoff_15v3_send),
                move |context, var_args!(hoff_12v3_recv), var_args!(hoff_15v3_send)| {
                    let mut hoff_12v3_recv = hoff_12v3_recv.borrow_mut_swap();
                    let hoff_12v3_recv = hoff_12v3_recv.drain(..);
                    let hoff_15v3_send = hydroflow::pusherator::for_each::ForEach::new(|
                        v|
                    {
                        hoff_15v3_send.give(Some(v));
                    });
                    let op_20v1 = {
                        fn check_input<Iter: ::std::iter::Iterator<Item = Item>, Item>(
                            iter: Iter,
                        ) -> impl ::std::iter::Iterator<Item = Item> {
                            iter
                        }
                        check_input::<_, _>(hoff_12v3_recv)
                    };
                    let op_20v1 = {
                        #[allow(non_snake_case)]
                        #[inline(always)]
                        pub fn op_20v1__next_tick__loc_unknown_start_9_21_end_9_28<
                            Item,
                            Input: ::std::iter::Iterator<Item = Item>,
                        >(input: Input) -> impl ::std::iter::Iterator<Item = Item> {
                            struct Pull<
                                Item,
                                Input: ::std::iter::Iterator<Item = Item>,
                            > {
                                inner: Input,
                            }
                            impl<
                                Item,
                                Input: ::std::iter::Iterator<Item = Item>,
                            > Iterator for Pull<Item, Input> {
                                type Item = Item;
                                #[inline(always)]
                                fn next(&mut self) -> Option<Self::Item> {
                                    self.inner.next()
                                }
                                #[inline(always)]
                                fn size_hint(&self) -> (usize, Option<usize>) {
                                    self.inner.size_hint()
                                }
                            }
                            Pull { inner: input }
                        }
                        op_20v1__next_tick__loc_unknown_start_9_21_end_9_28(op_20v1)
                    };
                    #[inline(always)]
                    fn check_pivot_run<
                        Pull: ::std::iter::Iterator<Item = Item>,
                        Push: hydroflow::pusherator::Pusherator<Item = Item>,
                        Item,
                    >(pull: Pull, push: Push) {
                        hydroflow::pusherator::pivot::Pivot::new(pull, push).run();
                    }
                    check_pivot_run(op_20v1, hoff_15v3_send);
                },
            );
            df.add_subgraph_stratified(
                "Subgraph GraphSubgraphId(2v1)",
                0,
                var_expr!(hoff_6v3_recv),
                var_expr!(hoff_9v3_send),
                move |context, var_args!(hoff_6v3_recv), var_args!(hoff_9v3_send)| {
                    let mut hoff_6v3_recv = hoff_6v3_recv.borrow_mut_swap();
                    let hoff_6v3_recv = hoff_6v3_recv.drain(..);
                    let hoff_9v3_send = hydroflow::pusherator::for_each::ForEach::new(|
                        v|
                    {
                        hoff_9v3_send.give(Some(v));
                    });
                    let op_25v1 = {
                        fn check_input<Iter: ::std::iter::Iterator<Item = Item>, Item>(
                            iter: Iter,
                        ) -> impl ::std::iter::Iterator<Item = Item> {
                            iter
                        }
                        check_input::<_, _>(hoff_6v3_recv)
                    };
                    let op_25v1 = {
                        #[allow(non_snake_case)]
                        #[inline(always)]
                        pub fn op_25v1__next_tick__loc_unknown_start_15_21_end_15_35<
                            Item,
                            Input: ::std::iter::Iterator<Item = Item>,
                        >(input: Input) -> impl ::std::iter::Iterator<Item = Item> {
                            struct Pull<
                                Item,
                                Input: ::std::iter::Iterator<Item = Item>,
                            > {
                                inner: Input,
                            }
                            impl<
                                Item,
                                Input: ::std::iter::Iterator<Item = Item>,
                            > Iterator for Pull<Item, Input> {
                                type Item = Item;
                                #[inline(always)]
                                fn next(&mut self) -> Option<Self::Item> {
                                    self.inner.next()
                                }
                                #[inline(always)]
                                fn size_hint(&self) -> (usize, Option<usize>) {
                                    self.inner.size_hint()
                                }
                            }
                            Pull { inner: input }
                        }
                        op_25v1__next_tick__loc_unknown_start_15_21_end_15_35(op_25v1)
                    };
                    #[inline(always)]
                    fn check_pivot_run<
                        Pull: ::std::iter::Iterator<Item = Item>,
                        Push: hydroflow::pusherator::Pusherator<Item = Item>,
                        Item,
                    >(pull: Pull, push: Push) {
                        hydroflow::pusherator::pivot::Pivot::new(pull, push).run();
                    }
                    check_pivot_run(op_25v1, hoff_9v3_send);
                },
            );
            let sg_4v1_node_37v1_groupbydata = df
                .add_state(
                    ::std::cell::RefCell::new(
                        hydroflow::rustc_hash::FxHashMap::<(_,), (Option<_>,)>::default(),
                    ),
                );
            let sg_4v1_node_38v1_counterdata = df
                .add_state(
                    ::std::cell::RefCell::new(
                        hydroflow::lang::monotonic_map::MonotonicMap::new_init(0..),
                    ),
                );
            let sg_4v1_node_8v1_uniquedata = df
                .add_state(
                    ::std::cell::RefCell::new(
                        hydroflow::lang::monotonic_map::MonotonicMap::<
                            _,
                            hydroflow::rustc_hash::FxHashSet<_>,
                        >::default(),
                    ),
                );
            df.add_subgraph_stratified(
                "Subgraph GraphSubgraphId(4v1)",
                1,
                var_expr!(hoff_21v3_recv),
                var_expr!(),
                move |context, var_args!(hoff_21v3_recv), var_args!()| {
                    let mut hoff_21v3_recv = hoff_21v3_recv.borrow_mut_swap();
                    let hoff_21v3_recv = hoff_21v3_recv.drain(..);
                    let mut sg_4v1_node_37v1_hashtable = context
                        .state_ref(sg_4v1_node_37v1_groupbydata)
                        .borrow_mut();
                    {
                        #[inline(always)]
                        fn check_input<
                            Iter: ::std::iter::Iterator<Item = (A, B)>,
                            A: ::std::clone::Clone,
                            B: ::std::clone::Clone,
                        >(iter: Iter) -> impl ::std::iter::Iterator<Item = (A, B)> {
                            iter
                        }
                        for kv in check_input(hoff_21v3_recv) {
                            let entry = sg_4v1_node_37v1_hashtable
                                .entry(kv.0)
                                .or_insert_with(|| (None,));
                            #[allow(clippy::redundant_closure_call)]
                            (|old: &mut (Option<_>,), val: (_,)| {
                                old
                                    .0 = if let Some(prev) = old.0.take() {
                                    Some({
                                        let prev: (hydroflow::rustc_hash::FxHashSet<_>, _) = prev;
                                        let mut set: hydroflow::rustc_hash::FxHashSet<_> = prev.0;
                                        if set.insert(val.0) {
                                            (set, prev.1 + 1)
                                        } else {
                                            (set, prev.1)
                                        }
                                    })
                                } else {
                                    Some({
                                        let mut set = hydroflow::rustc_hash::FxHashSet::<
                                            _,
                                        >::default();
                                        set.insert(val.0);
                                        (set, 1)
                                    })
                                };
                            })(entry, kv.1);
                        }
                    }
                    let op_37v1 = sg_4v1_node_37v1_hashtable.drain();
                    let op_37v1 = {
                        #[allow(non_snake_case)]
                        #[inline(always)]
                        pub fn op_37v1__group_by__loc_unknown_start_1_0_end_1_0<
                            Item,
                            Input: ::std::iter::Iterator<Item = Item>,
                        >(input: Input) -> impl ::std::iter::Iterator<Item = Item> {
                            struct Pull<
                                Item,
                                Input: ::std::iter::Iterator<Item = Item>,
                            > {
                                inner: Input,
                            }
                            impl<
                                Item,
                                Input: ::std::iter::Iterator<Item = Item>,
                            > Iterator for Pull<Item, Input> {
                                type Item = Item;
                                #[inline(always)]
                                fn next(&mut self) -> Option<Self::Item> {
                                    self.inner.next()
                                }
                                #[inline(always)]
                                fn size_hint(&self) -> (usize, Option<usize>) {
                                    self.inner.size_hint()
                                }
                            }
                            Pull { inner: input }
                        }
                        op_37v1__group_by__loc_unknown_start_1_0_end_1_0(op_37v1)
                    };
                    let op_38v1 = ::std::iter::Iterator::map(
                        op_37v1,
                        |item| {
                            let mut borrow = context
                                .state_ref(sg_4v1_node_38v1_counterdata)
                                .borrow_mut();
                            let counter = borrow
                                .get_mut_with(
                                    (context.current_tick(), context.current_stratum()),
                                    || 0..,
                                );
                            (counter.next().unwrap(), item)
                        },
                    );
                    let op_38v1 = {
                        #[allow(non_snake_case)]
                        #[inline(always)]
                        pub fn op_38v1__enumerate__loc_unknown_start_1_0_end_1_0<
                            Item,
                            Input: ::std::iter::Iterator<Item = Item>,
                        >(input: Input) -> impl ::std::iter::Iterator<Item = Item> {
                            struct Pull<
                                Item,
                                Input: ::std::iter::Iterator<Item = Item>,
                            > {
                                inner: Input,
                            }
                            impl<
                                Item,
                                Input: ::std::iter::Iterator<Item = Item>,
                            > Iterator for Pull<Item, Input> {
                                type Item = Item;
                                #[inline(always)]
                                fn next(&mut self) -> Option<Self::Item> {
                                    self.inner.next()
                                }
                                #[inline(always)]
                                fn size_hint(&self) -> (usize, Option<usize>) {
                                    self.inner.size_hint()
                                }
                            }
                            Pull { inner: input }
                        }
                        op_38v1__enumerate__loc_unknown_start_1_0_end_1_0(op_38v1)
                    };
                    let op_39v1 = op_38v1
                        .map(|(i, (g, a)): (_, ((_,), _))| (g.0, a.0.unwrap().1, i));
                    let op_39v1 = {
                        #[allow(non_snake_case)]
                        #[inline(always)]
                        pub fn op_39v1__map__loc_unknown_start_1_0_end_1_0<
                            Item,
                            Input: ::std::iter::Iterator<Item = Item>,
                        >(input: Input) -> impl ::std::iter::Iterator<Item = Item> {
                            struct Pull<
                                Item,
                                Input: ::std::iter::Iterator<Item = Item>,
                            > {
                                inner: Input,
                            }
                            impl<
                                Item,
                                Input: ::std::iter::Iterator<Item = Item>,
                            > Iterator for Pull<Item, Input> {
                                type Item = Item;
                                #[inline(always)]
                                fn next(&mut self) -> Option<Self::Item> {
                                    self.inner.next()
                                }
                                #[inline(always)]
                                fn size_hint(&self) -> (usize, Option<usize>) {
                                    self.inner.size_hint()
                                }
                            }
                            Pull { inner: input }
                        }
                        op_39v1__map__loc_unknown_start_1_0_end_1_0(op_39v1)
                    };
                    let op_8v1 = op_39v1
                        .filter(|item| {
                            let mut borrow = context
                                .state_ref(sg_4v1_node_8v1_uniquedata)
                                .borrow_mut();
                            let set = borrow
                                .get_mut_clear((
                                    context.current_tick(),
                                    context.current_stratum(),
                                ));
                            if !set.contains(item) {
                                set.insert(::std::clone::Clone::clone(item));
                                true
                            } else {
                                false
                            }
                        });
                    let op_8v1 = {
                        #[allow(non_snake_case)]
                        #[inline(always)]
                        pub fn op_8v1__unique__loc_unknown_start_5_20_end_5_27<
                            Item,
                            Input: ::std::iter::Iterator<Item = Item>,
                        >(input: Input) -> impl ::std::iter::Iterator<Item = Item> {
                            struct Pull<
                                Item,
                                Input: ::std::iter::Iterator<Item = Item>,
                            > {
                                inner: Input,
                            }
                            impl<
                                Item,
                                Input: ::std::iter::Iterator<Item = Item>,
                            > Iterator for Pull<Item, Input> {
                                type Item = Item;
                                #[inline(always)]
                                fn next(&mut self) -> Option<Self::Item> {
                                    self.inner.next()
                                }
                                #[inline(always)]
                                fn size_hint(&self) -> (usize, Option<usize>) {
                                    self.inner.size_hint()
                                }
                            }
                            Pull { inner: input }
                        }
                        op_8v1__unique__loc_unknown_start_5_20_end_5_27(op_8v1)
                    };
                    let op_28v1 = hydroflow::pusherator::for_each::ForEach::new(|v| {
                        result2.send(v).unwrap()
                    });
                    let op_28v1 = {
                        #[allow(non_snake_case)]
                        #[inline(always)]
                        pub fn op_28v1__for_each__loc_unknown_start_5_29_end_5_67<
                            Item,
                            Input: hydroflow::pusherator::Pusherator<Item = Item>,
                        >(
                            input: Input,
                        ) -> impl hydroflow::pusherator::Pusherator<Item = Item> {
                            struct Push<
                                Item,
                                Input: hydroflow::pusherator::Pusherator<Item = Item>,
                            > {
                                inner: Input,
                            }
                            impl<
                                Item,
                                Input: hydroflow::pusherator::Pusherator<Item = Item>,
                            > hydroflow::pusherator::Pusherator for Push<Item, Input> {
                                type Item = Item;
                                #[inline(always)]
                                fn give(&mut self, item: Self::Item) {
                                    self.inner.give(item)
                                }
                            }
                            Push { inner: input }
                        }
                        op_28v1__for_each__loc_unknown_start_5_29_end_5_67(op_28v1)
                    };
                    #[inline(always)]
                    fn check_pivot_run<
                        Pull: ::std::iter::Iterator<Item = Item>,
                        Push: hydroflow::pusherator::Pusherator<Item = Item>,
                        Item,
                    >(pull: Pull, push: Push) {
                        hydroflow::pusherator::pivot::Pivot::new(pull, push).run();
                    }
                    check_pivot_run(op_8v1, op_28v1);
                },
            );
            let sg_5v1_node_22v1_uniquedata = df
                .add_state(
                    ::std::cell::RefCell::new(
                        hydroflow::lang::monotonic_map::MonotonicMap::<
                            _,
                            hydroflow::rustc_hash::FxHashSet<_>,
                        >::default(),
                    ),
                );
            let sg_5v1_node_23v1_diffdata_handle = df
                .add_state(
                    ::std::cell::RefCell::new(
                        hydroflow::rustc_hash::FxHashSet::default(),
                    ),
                );
            let sg_5v1_node_11v1_uniquedata = df
                .add_state(
                    ::std::cell::RefCell::new(
                        hydroflow::lang::monotonic_map::MonotonicMap::<
                            _,
                            hydroflow::rustc_hash::FxHashSet<_>,
                        >::default(),
                    ),
                );
            let sg_5v1_node_45v1_persistdata = df
                .add_state(::std::cell::RefCell::new(::std::vec::Vec::new()));
            let sg_5v1_node_43v1_counterdata = df
                .add_state(::std::cell::RefCell::new(0..));
            df.add_subgraph_stratified(
                "Subgraph GraphSubgraphId(5v1)",
                1,
                var_expr!(hoff_9v3_recv, hoff_16v3_recv),
                var_expr!(hoff_6v3_send, hoff_10v3_send, hoff_13v3_send),
                move |
                    context,
                    var_args!(hoff_9v3_recv, hoff_16v3_recv),
                    var_args!(hoff_6v3_send, hoff_10v3_send, hoff_13v3_send)|
                {
                    let mut hoff_9v3_recv = hoff_9v3_recv.borrow_mut_swap();
                    let hoff_9v3_recv = hoff_9v3_recv.drain(..);
                    let mut hoff_16v3_recv = hoff_16v3_recv.borrow_mut_swap();
                    let hoff_16v3_recv = hoff_16v3_recv.drain(..);
                    let hoff_6v3_send = hydroflow::pusherator::for_each::ForEach::new(|
                        v|
                    {
                        hoff_6v3_send.give(Some(v));
                    });
                    let hoff_10v3_send = hydroflow::pusherator::for_each::ForEach::new(|
                        v|
                    {
                        hoff_10v3_send.give(Some(v));
                    });
                    let hoff_13v3_send = hydroflow::pusherator::for_each::ForEach::new(|
                        v|
                    {
                        hoff_13v3_send.give(Some(v));
                    });
                    let op_41v1 = hoff_16v3_recv.map(|(g, a): ((_, _), _)| (g.0, g.1));
                    let op_41v1 = {
                        #[allow(non_snake_case)]
                        #[inline(always)]
                        pub fn op_41v1__map__loc_unknown_start_1_0_end_1_0<
                            Item,
                            Input: ::std::iter::Iterator<Item = Item>,
                        >(input: Input) -> impl ::std::iter::Iterator<Item = Item> {
                            struct Pull<
                                Item,
                                Input: ::std::iter::Iterator<Item = Item>,
                            > {
                                inner: Input,
                            }
                            impl<
                                Item,
                                Input: ::std::iter::Iterator<Item = Item>,
                            > Iterator for Pull<Item, Input> {
                                type Item = Item;
                                #[inline(always)]
                                fn next(&mut self) -> Option<Self::Item> {
                                    self.inner.next()
                                }
                                #[inline(always)]
                                fn size_hint(&self) -> (usize, Option<usize>) {
                                    self.inner.size_hint()
                                }
                            }
                            Pull { inner: input }
                        }
                        op_41v1__map__loc_unknown_start_1_0_end_1_0(op_41v1)
                    };
                    let op_22v1 = op_41v1
                        .filter(|item| {
                            let mut borrow = context
                                .state_ref(sg_5v1_node_22v1_uniquedata)
                                .borrow_mut();
                            let set = borrow
                                .get_mut_clear((
                                    context.current_tick(),
                                    context.current_stratum(),
                                ));
                            if !set.contains(item) {
                                set.insert(::std::clone::Clone::clone(item));
                                true
                            } else {
                                false
                            }
                        });
                    let op_22v1 = {
                        #[allow(non_snake_case)]
                        #[inline(always)]
                        pub fn op_22v1__unique__loc_unknown_start_15_21_end_15_35<
                            Item,
                            Input: ::std::iter::Iterator<Item = Item>,
                        >(input: Input) -> impl ::std::iter::Iterator<Item = Item> {
                            struct Pull<
                                Item,
                                Input: ::std::iter::Iterator<Item = Item>,
                            > {
                                inner: Input,
                            }
                            impl<
                                Item,
                                Input: ::std::iter::Iterator<Item = Item>,
                            > Iterator for Pull<Item, Input> {
                                type Item = Item;
                                #[inline(always)]
                                fn next(&mut self) -> Option<Self::Item> {
                                    self.inner.next()
                                }
                                #[inline(always)]
                                fn size_hint(&self) -> (usize, Option<usize>) {
                                    self.inner.size_hint()
                                }
                            }
                            Pull { inner: input }
                        }
                        op_22v1__unique__loc_unknown_start_15_21_end_15_35(op_22v1)
                    };
                    let mut sg_5v1_node_23v1_negset = context
                        .state_ref(sg_5v1_node_23v1_diffdata_handle)
                        .borrow_mut();
                    sg_5v1_node_23v1_negset.extend(hoff_9v3_recv);
                    let op_23v1 = op_22v1
                        .filter(move |x| !sg_5v1_node_23v1_negset.contains(x));
                    let op_23v1 = {
                        #[allow(non_snake_case)]
                        #[inline(always)]
                        pub fn op_23v1__difference__loc_unknown_start_15_21_end_15_35<
                            Item,
                            Input: ::std::iter::Iterator<Item = Item>,
                        >(input: Input) -> impl ::std::iter::Iterator<Item = Item> {
                            struct Pull<
                                Item,
                                Input: ::std::iter::Iterator<Item = Item>,
                            > {
                                inner: Input,
                            }
                            impl<
                                Item,
                                Input: ::std::iter::Iterator<Item = Item>,
                            > Iterator for Pull<Item, Input> {
                                type Item = Item;
                                #[inline(always)]
                                fn next(&mut self) -> Option<Self::Item> {
                                    self.inner.next()
                                }
                                #[inline(always)]
                                fn size_hint(&self) -> (usize, Option<usize>) {
                                    self.inner.size_hint()
                                }
                            }
                            Pull { inner: input }
                        }
                        op_23v1__difference__loc_unknown_start_15_21_end_15_35(op_23v1)
                    };
                    let op_50v1 = hydroflow::pusherator::map::Map::new(
                        |row: (_, _)| ((row.0, row.1), ()),
                        hoff_10v3_send,
                    );
                    let op_50v1 = {
                        #[allow(non_snake_case)]
                        #[inline(always)]
                        pub fn op_50v1__map__loc_unknown_start_1_0_end_1_0<
                            Item,
                            Input: hydroflow::pusherator::Pusherator<Item = Item>,
                        >(
                            input: Input,
                        ) -> impl hydroflow::pusherator::Pusherator<Item = Item> {
                            struct Push<
                                Item,
                                Input: hydroflow::pusherator::Pusherator<Item = Item>,
                            > {
                                inner: Input,
                            }
                            impl<
                                Item,
                                Input: hydroflow::pusherator::Pusherator<Item = Item>,
                            > hydroflow::pusherator::Pusherator for Push<Item, Input> {
                                type Item = Item;
                                #[inline(always)]
                                fn give(&mut self, item: Self::Item) {
                                    self.inner.give(item)
                                }
                            }
                            Push { inner: input }
                        }
                        op_50v1__map__loc_unknown_start_1_0_end_1_0(op_50v1)
                    };
                    let op_46v1 = hydroflow::pusherator::map::Map::new(
                        |row: (_, _)| ((row.0,), ((row.1),)),
                        hoff_13v3_send,
                    );
                    let op_46v1 = {
                        #[allow(non_snake_case)]
                        #[inline(always)]
                        pub fn op_46v1__map__loc_unknown_start_1_0_end_1_0<
                            Item,
                            Input: hydroflow::pusherator::Pusherator<Item = Item>,
                        >(
                            input: Input,
                        ) -> impl hydroflow::pusherator::Pusherator<Item = Item> {
                            struct Push<
                                Item,
                                Input: hydroflow::pusherator::Pusherator<Item = Item>,
                            > {
                                inner: Input,
                            }
                            impl<
                                Item,
                                Input: hydroflow::pusherator::Pusherator<Item = Item>,
                            > hydroflow::pusherator::Pusherator for Push<Item, Input> {
                                type Item = Item;
                                #[inline(always)]
                                fn give(&mut self, item: Self::Item) {
                                    self.inner.give(item)
                                }
                            }
                            Push { inner: input }
                        }
                        op_46v1__map__loc_unknown_start_1_0_end_1_0(op_46v1)
                    };
                    let op_29v1 = hydroflow::pusherator::for_each::ForEach::new(|v| {
                        result3.send(v).unwrap()
                    });
                    let op_29v1 = {
                        #[allow(non_snake_case)]
                        #[inline(always)]
                        pub fn op_29v1__for_each__loc_unknown_start_6_29_end_6_67<
                            Item,
                            Input: hydroflow::pusherator::Pusherator<Item = Item>,
                        >(
                            input: Input,
                        ) -> impl hydroflow::pusherator::Pusherator<Item = Item> {
                            struct Push<
                                Item,
                                Input: hydroflow::pusherator::Pusherator<Item = Item>,
                            > {
                                inner: Input,
                            }
                            impl<
                                Item,
                                Input: hydroflow::pusherator::Pusherator<Item = Item>,
                            > hydroflow::pusherator::Pusherator for Push<Item, Input> {
                                type Item = Item;
                                #[inline(always)]
                                fn give(&mut self, item: Self::Item) {
                                    self.inner.give(item)
                                }
                            }
                            Push { inner: input }
                        }
                        op_29v1__for_each__loc_unknown_start_6_29_end_6_67(op_29v1)
                    };
                    let op_11v1 = hydroflow::pusherator::filter::Filter::new(
                        |item| {
                            let mut borrow = context
                                .state_ref(sg_5v1_node_11v1_uniquedata)
                                .borrow_mut();
                            let set = borrow
                                .get_mut_clear((
                                    context.current_tick(),
                                    context.current_stratum(),
                                ));
                            if !set.contains(item) {
                                set.insert(::std::clone::Clone::clone(item));
                                true
                            } else {
                                false
                            }
                        },
                        op_29v1,
                    );
                    let op_11v1 = {
                        #[allow(non_snake_case)]
                        #[inline(always)]
                        pub fn op_11v1__unique__loc_unknown_start_6_20_end_6_27<
                            Item,
                            Input: hydroflow::pusherator::Pusherator<Item = Item>,
                        >(
                            input: Input,
                        ) -> impl hydroflow::pusherator::Pusherator<Item = Item> {
                            struct Push<
                                Item,
                                Input: hydroflow::pusherator::Pusherator<Item = Item>,
                            > {
                                inner: Input,
                            }
                            impl<
                                Item,
                                Input: hydroflow::pusherator::Pusherator<Item = Item>,
                            > hydroflow::pusherator::Pusherator for Push<Item, Input> {
                                type Item = Item;
                                #[inline(always)]
                                fn give(&mut self, item: Self::Item) {
                                    self.inner.give(item)
                                }
                            }
                            Push { inner: input }
                        }
                        op_11v1__unique__loc_unknown_start_6_20_end_6_27(op_11v1)
                    };
                    let mut sg_5v1_node_45v1_persistvec = context
                        .state_ref(sg_5v1_node_45v1_persistdata)
                        .borrow_mut();
                    let op_45v1 = {
                        fn constrain_types<'ctx, Push, Item>(
                            vec: &'ctx mut Vec<Item>,
                            mut output: Push,
                        ) -> impl 'ctx + hydroflow::pusherator::Pusherator<Item = Item>
                        where
                            Push: 'ctx + hydroflow::pusherator::Pusherator<Item = Item>,
                            Item: ::std::clone::Clone,
                        {
                            vec.iter()
                                .cloned()
                                .for_each(|item| {
                                    hydroflow::pusherator::Pusherator::give(&mut output, item);
                                });
                            hydroflow::pusherator::map::Map::new(
                                |item| {
                                    vec.push(item);
                                    vec.last().unwrap().clone()
                                },
                                output,
                            )
                        }
                        constrain_types(&mut *sg_5v1_node_45v1_persistvec, op_11v1)
                    };
                    let op_45v1 = {
                        #[allow(non_snake_case)]
                        #[inline(always)]
                        pub fn op_45v1__persist__loc_unknown_start_1_0_end_1_0<
                            Item,
                            Input: hydroflow::pusherator::Pusherator<Item = Item>,
                        >(
                            input: Input,
                        ) -> impl hydroflow::pusherator::Pusherator<Item = Item> {
                            struct Push<
                                Item,
                                Input: hydroflow::pusherator::Pusherator<Item = Item>,
                            > {
                                inner: Input,
                            }
                            impl<
                                Item,
                                Input: hydroflow::pusherator::Pusherator<Item = Item>,
                            > hydroflow::pusherator::Pusherator for Push<Item, Input> {
                                type Item = Item;
                                #[inline(always)]
                                fn give(&mut self, item: Self::Item) {
                                    self.inner.give(item)
                                }
                            }
                            Push { inner: input }
                        }
                        op_45v1__persist__loc_unknown_start_1_0_end_1_0(op_45v1)
                    };
                    let op_44v1 = hydroflow::pusherator::map::Map::new(
                        |(i, (g, a)): (_, ((_, _), _))| (g.0, g.1, i),
                        op_45v1,
                    );
                    let op_44v1 = {
                        #[allow(non_snake_case)]
                        #[inline(always)]
                        pub fn op_44v1__map__loc_unknown_start_1_0_end_1_0<
                            Item,
                            Input: hydroflow::pusherator::Pusherator<Item = Item>,
                        >(
                            input: Input,
                        ) -> impl hydroflow::pusherator::Pusherator<Item = Item> {
                            struct Push<
                                Item,
                                Input: hydroflow::pusherator::Pusherator<Item = Item>,
                            > {
                                inner: Input,
                            }
                            impl<
                                Item,
                                Input: hydroflow::pusherator::Pusherator<Item = Item>,
                            > hydroflow::pusherator::Pusherator for Push<Item, Input> {
                                type Item = Item;
                                #[inline(always)]
                                fn give(&mut self, item: Self::Item) {
                                    self.inner.give(item)
                                }
                            }
                            Push { inner: input }
                        }
                        op_44v1__map__loc_unknown_start_1_0_end_1_0(op_44v1)
                    };
                    let op_43v1 = hydroflow::pusherator::map::Map::new(
                        |item| {
                            let mut counter = context
                                .state_ref(sg_5v1_node_43v1_counterdata)
                                .borrow_mut();
                            (counter.next().unwrap(), item)
                        },
                        op_44v1,
                    );
                    let op_43v1 = {
                        #[allow(non_snake_case)]
                        #[inline(always)]
                        pub fn op_43v1__enumerate__loc_unknown_start_1_0_end_1_0<
                            Item,
                            Input: hydroflow::pusherator::Pusherator<Item = Item>,
                        >(
                            input: Input,
                        ) -> impl hydroflow::pusherator::Pusherator<Item = Item> {
                            struct Push<
                                Item,
                                Input: hydroflow::pusherator::Pusherator<Item = Item>,
                            > {
                                inner: Input,
                            }
                            impl<
                                Item,
                                Input: hydroflow::pusherator::Pusherator<Item = Item>,
                            > hydroflow::pusherator::Pusherator for Push<Item, Input> {
                                type Item = Item;
                                #[inline(always)]
                                fn give(&mut self, item: Self::Item) {
                                    self.inner.give(item)
                                }
                            }
                            Push { inner: input }
                        }
                        op_43v1__enumerate__loc_unknown_start_1_0_end_1_0(op_43v1)
                    };
                    let op_42v1 = hydroflow::pusherator::map::Map::new(
                        |row: (_, _)| ((row.0, row.1), ()),
                        op_43v1,
                    );
                    let op_42v1 = {
                        #[allow(non_snake_case)]
                        #[inline(always)]
                        pub fn op_42v1__map__loc_unknown_start_1_0_end_1_0<
                            Item,
                            Input: hydroflow::pusherator::Pusherator<Item = Item>,
                        >(
                            input: Input,
                        ) -> impl hydroflow::pusherator::Pusherator<Item = Item> {
                            struct Push<
                                Item,
                                Input: hydroflow::pusherator::Pusherator<Item = Item>,
                            > {
                                inner: Input,
                            }
                            impl<
                                Item,
                                Input: hydroflow::pusherator::Pusherator<Item = Item>,
                            > hydroflow::pusherator::Pusherator for Push<Item, Input> {
                                type Item = Item;
                                #[inline(always)]
                                fn give(&mut self, item: Self::Item) {
                                    self.inner.give(item)
                                }
                            }
                            Push { inner: input }
                        }
                        op_42v1__map__loc_unknown_start_1_0_end_1_0(op_42v1)
                    };
                    let op_24v1 = hydroflow::pusherator::tee::Tee::new(
                        op_42v1,
                        hydroflow::pusherator::tee::Tee::new(
                            op_46v1,
                            hydroflow::pusherator::tee::Tee::new(op_50v1, hoff_6v3_send),
                        ),
                    );
                    let op_24v1 = {
                        #[allow(non_snake_case)]
                        #[inline(always)]
                        pub fn op_24v1__tee__loc_unknown_start_15_21_end_15_35<
                            Item,
                            Input: hydroflow::pusherator::Pusherator<Item = Item>,
                        >(
                            input: Input,
                        ) -> impl hydroflow::pusherator::Pusherator<Item = Item> {
                            struct Push<
                                Item,
                                Input: hydroflow::pusherator::Pusherator<Item = Item>,
                            > {
                                inner: Input,
                            }
                            impl<
                                Item,
                                Input: hydroflow::pusherator::Pusherator<Item = Item>,
                            > hydroflow::pusherator::Pusherator for Push<Item, Input> {
                                type Item = Item;
                                #[inline(always)]
                                fn give(&mut self, item: Self::Item) {
                                    self.inner.give(item)
                                }
                            }
                            Push { inner: input }
                        }
                        op_24v1__tee__loc_unknown_start_15_21_end_15_35(op_24v1)
                    };
                    #[inline(always)]
                    fn check_pivot_run<
                        Pull: ::std::iter::Iterator<Item = Item>,
                        Push: hydroflow::pusherator::Pusherator<Item = Item>,
                        Item,
                    >(pull: Pull, push: Push) {
                        hydroflow::pusherator::pivot::Pivot::new(pull, push).run();
                    }
                    check_pivot_run(op_23v1, op_24v1);
                    context.schedule_subgraph(context.current_subgraph(), false);
                },
            );
            let sg_6v1_node_47v1_groupbydata = df
                .add_state(
                    ::std::cell::RefCell::new(
                        hydroflow::rustc_hash::FxHashMap::<(_,), (Option<_>,)>::default(),
                    ),
                );
            let sg_6v1_node_48v1_counterdata = df
                .add_state(
                    ::std::cell::RefCell::new(
                        hydroflow::lang::monotonic_map::MonotonicMap::new_init(0..),
                    ),
                );
            let sg_6v1_node_14v1_uniquedata = df
                .add_state(
                    ::std::cell::RefCell::new(
                        hydroflow::lang::monotonic_map::MonotonicMap::<
                            _,
                            hydroflow::rustc_hash::FxHashSet<_>,
                        >::default(),
                    ),
                );
            df.add_subgraph_stratified(
                "Subgraph GraphSubgraphId(6v1)",
                2,
                var_expr!(hoff_13v3_recv),
                var_expr!(),
                move |context, var_args!(hoff_13v3_recv), var_args!()| {
                    let mut hoff_13v3_recv = hoff_13v3_recv.borrow_mut_swap();
                    let hoff_13v3_recv = hoff_13v3_recv.drain(..);
                    let mut sg_6v1_node_47v1_hashtable = context
                        .state_ref(sg_6v1_node_47v1_groupbydata)
                        .borrow_mut();
                    {
                        #[inline(always)]
                        fn check_input<
                            Iter: ::std::iter::Iterator<Item = (A, B)>,
                            A: ::std::clone::Clone,
                            B: ::std::clone::Clone,
                        >(iter: Iter) -> impl ::std::iter::Iterator<Item = (A, B)> {
                            iter
                        }
                        for kv in check_input(hoff_13v3_recv) {
                            let entry = sg_6v1_node_47v1_hashtable
                                .entry(kv.0)
                                .or_insert_with(|| (None,));
                            #[allow(clippy::redundant_closure_call)]
                            (|old: &mut (Option<_>,), val: (_,)| {
                                old
                                    .0 = if let Some(prev) = old.0.take() {
                                    Some({
                                        let prev: (hydroflow::rustc_hash::FxHashSet<_>, _) = prev;
                                        let mut set: hydroflow::rustc_hash::FxHashSet<_> = prev.0;
                                        if set.insert(val.0) {
                                            (set, prev.1 + 1)
                                        } else {
                                            (set, prev.1)
                                        }
                                    })
                                } else {
                                    Some({
                                        let mut set = hydroflow::rustc_hash::FxHashSet::<
                                            _,
                                        >::default();
                                        set.insert(val.0);
                                        (set, 1)
                                    })
                                };
                            })(entry, kv.1);
                        }
                    }
                    let op_47v1 = sg_6v1_node_47v1_hashtable
                        .iter()
                        .map(
                            #[allow(suspicious_double_ref_op, clippy::clone_on_copy)]
                            |(k, v)| (k.clone(), v.clone()),
                        );
                    let op_47v1 = {
                        #[allow(non_snake_case)]
                        #[inline(always)]
                        pub fn op_47v1__group_by__loc_unknown_start_1_0_end_1_0<
                            Item,
                            Input: ::std::iter::Iterator<Item = Item>,
                        >(input: Input) -> impl ::std::iter::Iterator<Item = Item> {
                            struct Pull<
                                Item,
                                Input: ::std::iter::Iterator<Item = Item>,
                            > {
                                inner: Input,
                            }
                            impl<
                                Item,
                                Input: ::std::iter::Iterator<Item = Item>,
                            > Iterator for Pull<Item, Input> {
                                type Item = Item;
                                #[inline(always)]
                                fn next(&mut self) -> Option<Self::Item> {
                                    self.inner.next()
                                }
                                #[inline(always)]
                                fn size_hint(&self) -> (usize, Option<usize>) {
                                    self.inner.size_hint()
                                }
                            }
                            Pull { inner: input }
                        }
                        op_47v1__group_by__loc_unknown_start_1_0_end_1_0(op_47v1)
                    };
                    let op_48v1 = ::std::iter::Iterator::map(
                        op_47v1,
                        |item| {
                            let mut borrow = context
                                .state_ref(sg_6v1_node_48v1_counterdata)
                                .borrow_mut();
                            let counter = borrow
                                .get_mut_with(
                                    (context.current_tick(), context.current_stratum()),
                                    || 0..,
                                );
                            (counter.next().unwrap(), item)
                        },
                    );
                    let op_48v1 = {
                        #[allow(non_snake_case)]
                        #[inline(always)]
                        pub fn op_48v1__enumerate__loc_unknown_start_1_0_end_1_0<
                            Item,
                            Input: ::std::iter::Iterator<Item = Item>,
                        >(input: Input) -> impl ::std::iter::Iterator<Item = Item> {
                            struct Pull<
                                Item,
                                Input: ::std::iter::Iterator<Item = Item>,
                            > {
                                inner: Input,
                            }
                            impl<
                                Item,
                                Input: ::std::iter::Iterator<Item = Item>,
                            > Iterator for Pull<Item, Input> {
                                type Item = Item;
                                #[inline(always)]
                                fn next(&mut self) -> Option<Self::Item> {
                                    self.inner.next()
                                }
                                #[inline(always)]
                                fn size_hint(&self) -> (usize, Option<usize>) {
                                    self.inner.size_hint()
                                }
                            }
                            Pull { inner: input }
                        }
                        op_48v1__enumerate__loc_unknown_start_1_0_end_1_0(op_48v1)
                    };
                    let op_49v1 = op_48v1
                        .map(|(i, (g, a)): (_, ((_,), _))| (g.0, a.0.unwrap().1, i));
                    let op_49v1 = {
                        #[allow(non_snake_case)]
                        #[inline(always)]
                        pub fn op_49v1__map__loc_unknown_start_1_0_end_1_0<
                            Item,
                            Input: ::std::iter::Iterator<Item = Item>,
                        >(input: Input) -> impl ::std::iter::Iterator<Item = Item> {
                            struct Pull<
                                Item,
                                Input: ::std::iter::Iterator<Item = Item>,
                            > {
                                inner: Input,
                            }
                            impl<
                                Item,
                                Input: ::std::iter::Iterator<Item = Item>,
                            > Iterator for Pull<Item, Input> {
                                type Item = Item;
                                #[inline(always)]
                                fn next(&mut self) -> Option<Self::Item> {
                                    self.inner.next()
                                }
                                #[inline(always)]
                                fn size_hint(&self) -> (usize, Option<usize>) {
                                    self.inner.size_hint()
                                }
                            }
                            Pull { inner: input }
                        }
                        op_49v1__map__loc_unknown_start_1_0_end_1_0(op_49v1)
                    };
                    let op_14v1 = op_49v1
                        .filter(|item| {
                            let mut borrow = context
                                .state_ref(sg_6v1_node_14v1_uniquedata)
                                .borrow_mut();
                            let set = borrow
                                .get_mut_clear((
                                    context.current_tick(),
                                    context.current_stratum(),
                                ));
                            if !set.contains(item) {
                                set.insert(::std::clone::Clone::clone(item));
                                true
                            } else {
                                false
                            }
                        });
                    let op_14v1 = {
                        #[allow(non_snake_case)]
                        #[inline(always)]
                        pub fn op_14v1__unique__loc_unknown_start_7_20_end_7_27<
                            Item,
                            Input: ::std::iter::Iterator<Item = Item>,
                        >(input: Input) -> impl ::std::iter::Iterator<Item = Item> {
                            struct Pull<
                                Item,
                                Input: ::std::iter::Iterator<Item = Item>,
                            > {
                                inner: Input,
                            }
                            impl<
                                Item,
                                Input: ::std::iter::Iterator<Item = Item>,
                            > Iterator for Pull<Item, Input> {
                                type Item = Item;
                                #[inline(always)]
                                fn next(&mut self) -> Option<Self::Item> {
                                    self.inner.next()
                                }
                                #[inline(always)]
                                fn size_hint(&self) -> (usize, Option<usize>) {
                                    self.inner.size_hint()
                                }
                            }
                            Pull { inner: input }
                        }
                        op_14v1__unique__loc_unknown_start_7_20_end_7_27(op_14v1)
                    };
                    let op_30v1 = hydroflow::pusherator::for_each::ForEach::new(|v| {
                        result4.send(v).unwrap()
                    });
                    let op_30v1 = {
                        #[allow(non_snake_case)]
                        #[inline(always)]
                        pub fn op_30v1__for_each__loc_unknown_start_7_29_end_7_67<
                            Item,
                            Input: hydroflow::pusherator::Pusherator<Item = Item>,
                        >(
                            input: Input,
                        ) -> impl hydroflow::pusherator::Pusherator<Item = Item> {
                            struct Push<
                                Item,
                                Input: hydroflow::pusherator::Pusherator<Item = Item>,
                            > {
                                inner: Input,
                            }
                            impl<
                                Item,
                                Input: hydroflow::pusherator::Pusherator<Item = Item>,
                            > hydroflow::pusherator::Pusherator for Push<Item, Input> {
                                type Item = Item;
                                #[inline(always)]
                                fn give(&mut self, item: Self::Item) {
                                    self.inner.give(item)
                                }
                            }
                            Push { inner: input }
                        }
                        op_30v1__for_each__loc_unknown_start_7_29_end_7_67(op_30v1)
                    };
                    #[inline(always)]
                    fn check_pivot_run<
                        Pull: ::std::iter::Iterator<Item = Item>,
                        Push: hydroflow::pusherator::Pusherator<Item = Item>,
                        Item,
                    >(pull: Pull, push: Push) {
                        hydroflow::pusherator::pivot::Pivot::new(pull, push).run();
                    }
                    check_pivot_run(op_14v1, op_30v1);
                    context.schedule_subgraph(context.current_subgraph(), false);
                },
            );
            let sg_7v1_node_51v1_counterdata = df
                .add_state(::std::cell::RefCell::new(0..));
            let sg_7v1_node_17v1_uniquedata = df
                .add_state(
                    ::std::cell::RefCell::new(
                        hydroflow::lang::monotonic_map::MonotonicMap::<
                            _,
                            hydroflow::rustc_hash::FxHashSet<_>,
                        >::default(),
                    ),
                );
            let sg_7v1_node_18v1_diffdata_handle = df
                .add_state(
                    ::std::cell::RefCell::new(
                        hydroflow::rustc_hash::FxHashSet::default(),
                    ),
                );
            let sg_7v1_node_31v1_persistdata = df
                .add_state(::std::cell::RefCell::new(::std::vec::Vec::new()));
            df.add_subgraph_stratified(
                "Subgraph GraphSubgraphId(7v1)",
                1,
                var_expr!(hoff_10v3_recv, hoff_15v3_recv),
                var_expr!(hoff_12v3_send),
                move |
                    context,
                    var_args!(hoff_10v3_recv, hoff_15v3_recv),
                    var_args!(hoff_12v3_send)|
                {
                    let mut hoff_10v3_recv = hoff_10v3_recv.borrow_mut_swap();
                    let hoff_10v3_recv = hoff_10v3_recv.drain(..);
                    let mut hoff_15v3_recv = hoff_15v3_recv.borrow_mut_swap();
                    let hoff_15v3_recv = hoff_15v3_recv.drain(..);
                    let hoff_12v3_send = hydroflow::pusherator::for_each::ForEach::new(|
                        v|
                    {
                        hoff_12v3_send.give(Some(v));
                    });
                    let op_51v1 = ::std::iter::Iterator::map(
                        hoff_10v3_recv,
                        |item| {
                            let mut counter = context
                                .state_ref(sg_7v1_node_51v1_counterdata)
                                .borrow_mut();
                            (counter.next().unwrap(), item)
                        },
                    );
                    let op_51v1 = {
                        #[allow(non_snake_case)]
                        #[inline(always)]
                        pub fn op_51v1__enumerate__loc_unknown_start_1_0_end_1_0<
                            Item,
                            Input: ::std::iter::Iterator<Item = Item>,
                        >(input: Input) -> impl ::std::iter::Iterator<Item = Item> {
                            struct Pull<
                                Item,
                                Input: ::std::iter::Iterator<Item = Item>,
                            > {
                                inner: Input,
                            }
                            impl<
                                Item,
                                Input: ::std::iter::Iterator<Item = Item>,
                            > Iterator for Pull<Item, Input> {
                                type Item = Item;
                                #[inline(always)]
                                fn next(&mut self) -> Option<Self::Item> {
                                    self.inner.next()
                                }
                                #[inline(always)]
                                fn size_hint(&self) -> (usize, Option<usize>) {
                                    self.inner.size_hint()
                                }
                            }
                            Pull { inner: input }
                        }
                        op_51v1__enumerate__loc_unknown_start_1_0_end_1_0(op_51v1)
                    };
                    let op_52v1 = op_51v1
                        .map(|(i, (g, a)): (_, ((_, _), _))| (g.0, g.1, i));
                    let op_52v1 = {
                        #[allow(non_snake_case)]
                        #[inline(always)]
                        pub fn op_52v1__map__loc_unknown_start_1_0_end_1_0<
                            Item,
                            Input: ::std::iter::Iterator<Item = Item>,
                        >(input: Input) -> impl ::std::iter::Iterator<Item = Item> {
                            struct Pull<
                                Item,
                                Input: ::std::iter::Iterator<Item = Item>,
                            > {
                                inner: Input,
                            }
                            impl<
                                Item,
                                Input: ::std::iter::Iterator<Item = Item>,
                            > Iterator for Pull<Item, Input> {
                                type Item = Item;
                                #[inline(always)]
                                fn next(&mut self) -> Option<Self::Item> {
                                    self.inner.next()
                                }
                                #[inline(always)]
                                fn size_hint(&self) -> (usize, Option<usize>) {
                                    self.inner.size_hint()
                                }
                            }
                            Pull { inner: input }
                        }
                        op_52v1__map__loc_unknown_start_1_0_end_1_0(op_52v1)
                    };
                    let op_17v1 = op_52v1
                        .filter(|item| {
                            let mut borrow = context
                                .state_ref(sg_7v1_node_17v1_uniquedata)
                                .borrow_mut();
                            let set = borrow
                                .get_mut_clear((
                                    context.current_tick(),
                                    context.current_stratum(),
                                ));
                            if !set.contains(item) {
                                set.insert(::std::clone::Clone::clone(item));
                                true
                            } else {
                                false
                            }
                        });
                    let op_17v1 = {
                        #[allow(non_snake_case)]
                        #[inline(always)]
                        pub fn op_17v1__unique__loc_unknown_start_9_21_end_9_28<
                            Item,
                            Input: ::std::iter::Iterator<Item = Item>,
                        >(input: Input) -> impl ::std::iter::Iterator<Item = Item> {
                            struct Pull<
                                Item,
                                Input: ::std::iter::Iterator<Item = Item>,
                            > {
                                inner: Input,
                            }
                            impl<
                                Item,
                                Input: ::std::iter::Iterator<Item = Item>,
                            > Iterator for Pull<Item, Input> {
                                type Item = Item;
                                #[inline(always)]
                                fn next(&mut self) -> Option<Self::Item> {
                                    self.inner.next()
                                }
                                #[inline(always)]
                                fn size_hint(&self) -> (usize, Option<usize>) {
                                    self.inner.size_hint()
                                }
                            }
                            Pull { inner: input }
                        }
                        op_17v1__unique__loc_unknown_start_9_21_end_9_28(op_17v1)
                    };
                    let mut sg_7v1_node_18v1_negset = context
                        .state_ref(sg_7v1_node_18v1_diffdata_handle)
                        .borrow_mut();
                    sg_7v1_node_18v1_negset.extend(hoff_15v3_recv);
                    let op_18v1 = op_17v1
                        .filter(move |x| !sg_7v1_node_18v1_negset.contains(x));
                    let op_18v1 = {
                        #[allow(non_snake_case)]
                        #[inline(always)]
                        pub fn op_18v1__difference__loc_unknown_start_9_21_end_9_28<
                            Item,
                            Input: ::std::iter::Iterator<Item = Item>,
                        >(input: Input) -> impl ::std::iter::Iterator<Item = Item> {
                            struct Pull<
                                Item,
                                Input: ::std::iter::Iterator<Item = Item>,
                            > {
                                inner: Input,
                            }
                            impl<
                                Item,
                                Input: ::std::iter::Iterator<Item = Item>,
                            > Iterator for Pull<Item, Input> {
                                type Item = Item;
                                #[inline(always)]
                                fn next(&mut self) -> Option<Self::Item> {
                                    self.inner.next()
                                }
                                #[inline(always)]
                                fn size_hint(&self) -> (usize, Option<usize>) {
                                    self.inner.size_hint()
                                }
                            }
                            Pull { inner: input }
                        }
                        op_18v1__difference__loc_unknown_start_9_21_end_9_28(op_18v1)
                    };
                    let op_32v1 = hydroflow::pusherator::for_each::ForEach::new(|v| {
                        result5.send(v).unwrap()
                    });
                    let op_32v1 = {
                        #[allow(non_snake_case)]
                        #[inline(always)]
                        pub fn op_32v1__for_each__loc_unknown_start_10_29_end_10_67<
                            Item,
                            Input: hydroflow::pusherator::Pusherator<Item = Item>,
                        >(
                            input: Input,
                        ) -> impl hydroflow::pusherator::Pusherator<Item = Item> {
                            struct Push<
                                Item,
                                Input: hydroflow::pusherator::Pusherator<Item = Item>,
                            > {
                                inner: Input,
                            }
                            impl<
                                Item,
                                Input: hydroflow::pusherator::Pusherator<Item = Item>,
                            > hydroflow::pusherator::Pusherator for Push<Item, Input> {
                                type Item = Item;
                                #[inline(always)]
                                fn give(&mut self, item: Self::Item) {
                                    self.inner.give(item)
                                }
                            }
                            Push { inner: input }
                        }
                        op_32v1__for_each__loc_unknown_start_10_29_end_10_67(op_32v1)
                    };
                    let mut sg_7v1_node_31v1_persistvec = context
                        .state_ref(sg_7v1_node_31v1_persistdata)
                        .borrow_mut();
                    let op_31v1 = {
                        fn constrain_types<'ctx, Push, Item>(
                            vec: &'ctx mut Vec<Item>,
                            mut output: Push,
                        ) -> impl 'ctx + hydroflow::pusherator::Pusherator<Item = Item>
                        where
                            Push: 'ctx + hydroflow::pusherator::Pusherator<Item = Item>,
                            Item: ::std::clone::Clone,
                        {
                            vec.iter()
                                .cloned()
                                .for_each(|item| {
                                    hydroflow::pusherator::Pusherator::give(&mut output, item);
                                });
                            hydroflow::pusherator::map::Map::new(
                                |item| {
                                    vec.push(item);
                                    vec.last().unwrap().clone()
                                },
                                output,
                            )
                        }
                        constrain_types(&mut *sg_7v1_node_31v1_persistvec, op_32v1)
                    };
                    let op_31v1 = {
                        #[allow(non_snake_case)]
                        #[inline(always)]
                        pub fn op_31v1__persist__loc_unknown_start_10_20_end_10_27<
                            Item,
                            Input: hydroflow::pusherator::Pusherator<Item = Item>,
                        >(
                            input: Input,
                        ) -> impl hydroflow::pusherator::Pusherator<Item = Item> {
                            struct Push<
                                Item,
                                Input: hydroflow::pusherator::Pusherator<Item = Item>,
                            > {
                                inner: Input,
                            }
                            impl<
                                Item,
                                Input: hydroflow::pusherator::Pusherator<Item = Item>,
                            > hydroflow::pusherator::Pusherator for Push<Item, Input> {
                                type Item = Item;
                                #[inline(always)]
                                fn give(&mut self, item: Self::Item) {
                                    self.inner.give(item)
                                }
                            }
                            Push { inner: input }
                        }
                        op_31v1__persist__loc_unknown_start_10_20_end_10_27(op_31v1)
                    };
                    let op_19v1 = hydroflow::pusherator::tee::Tee::new(
                        op_31v1,
                        hoff_12v3_send,
                    );
                    let op_19v1 = {
                        #[allow(non_snake_case)]
                        #[inline(always)]
                        pub fn op_19v1__tee__loc_unknown_start_9_21_end_9_28<
                            Item,
                            Input: hydroflow::pusherator::Pusherator<Item = Item>,
                        >(
                            input: Input,
                        ) -> impl hydroflow::pusherator::Pusherator<Item = Item> {
                            struct Push<
                                Item,
                                Input: hydroflow::pusherator::Pusherator<Item = Item>,
                            > {
                                inner: Input,
                            }
                            impl<
                                Item,
                                Input: hydroflow::pusherator::Pusherator<Item = Item>,
                            > hydroflow::pusherator::Pusherator for Push<Item, Input> {
                                type Item = Item;
                                #[inline(always)]
                                fn give(&mut self, item: Self::Item) {
                                    self.inner.give(item)
                                }
                            }
                            Push { inner: input }
                        }
                        op_19v1__tee__loc_unknown_start_9_21_end_9_28(op_19v1)
                    };
                    #[inline(always)]
                    fn check_pivot_run<
                        Pull: ::std::iter::Iterator<Item = Item>,
                        Push: hydroflow::pusherator::Pusherator<Item = Item>,
                        Item,
                    >(pull: Pull, push: Push) {
                        hydroflow::pusherator::pivot::Pivot::new(pull, push).run();
                    }
                    check_pivot_run(op_18v1, op_19v1);
                    context.schedule_subgraph(context.current_subgraph(), false);
                },
            );
            df
        }
    }
}

