refactor(hydroflow_plus): move rewrites to a submodule

hydroflow_plus/src/builder/built.rs

@@ -73,7 +73,7 @@ impl<'a> BuiltFlow<'a> {
     }
 
     pub fn with_default_optimize(self) -> BuiltFlow<'a> {
-        self.optimize_with(crate::persist_pullup::persist_pullup)
+        self.optimize_with(crate::rewrites::persist_pullup::persist_pullup)
     }
 
     fn into_deploy<D: LocalDeploy<'a>>(mut self) -> DeployFlow<'a, D> {

hydroflow_plus/src/lib.rs

@@ -34,10 +34,7 @@ pub use builder::FlowBuilder;
 
 pub mod ir;
 
-pub mod persist_pullup;
-pub mod profiler;
-
-pub mod properties;
+pub mod rewrites;
 
 mod staging_util;
 

hydroflow_plus/src/rewrites/mod.rs

@@ -0,0 +1,3 @@
+pub mod persist_pullup;
+pub mod profiler;
+pub mod properties;

hydroflow_plus/src/profiler.rs → hydroflow_plus/src/rewrites/profiler.rs

@@ -3,8 +3,9 @@ use std::cell::RefCell;
 use hydroflow::futures::channel::mpsc::UnboundedSender;
 use stageleft::*;
 
+use super::profiler as myself; // TODO(shadaj): stageleft does not support `self::...`
 use crate::ir::*;
-use crate::{profiler as myself, RuntimeContext};
+use crate::RuntimeContext;
 
 pub fn increment_counter(count: &mut u64) {
     *count += 1;

hydroflow_plus/src/snapshots/hydroflow_plus__persist_pullup__tests__persist_pullup_behind_tee-2.snap → hydroflow_plus/src/rewrites/snapshots/hydroflow_plus__rewrites__persist_pullup__tests__persist_pullup_behind_tee-2.snap

@@ -1,17 +1,17 @@
 ---
-source: hydroflow_plus/src/persist_pullup.rs
+source: hydroflow_plus/src/rewrites/persist_pullup.rs
 expression: optimized.ir()
 ---
 [
     ForEach {
-        f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , () > ({ use crate :: __staged :: persist_pullup :: tests :: * ; | n | println ! ("{}" , n) }),
+        f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , () > ({ use crate :: __staged :: rewrites :: persist_pullup :: tests :: * ; | n | println ! ("{}" , n) }),
         input: Persist(
             Map {
-                f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , i32 > ({ use crate :: __staged :: persist_pullup :: tests :: * ; | v | v + 1 }),
+                f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , i32 > ({ use crate :: __staged :: rewrites :: persist_pullup :: tests :: * ; | v | v + 1 }),
                 input: Tee {
                     inner: <tee>: Source {
                         source: Iter(
-                            { use crate :: __staged :: persist_pullup :: tests :: * ; 0 .. 10 },
+                            { use crate :: __staged :: rewrites :: persist_pullup :: tests :: * ; 0 .. 10 },
                         ),
                         location_kind: Process(
                             0,
@@ -22,14 +22,14 @@ expression: optimized.ir()
         ),
     },
     ForEach {
-        f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , () > ({ use crate :: __staged :: persist_pullup :: tests :: * ; | n | println ! ("{}" , n) }),
+        f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , () > ({ use crate :: __staged :: rewrites :: persist_pullup :: tests :: * ; | n | println ! ("{}" , n) }),
         input: Persist(
             Map {
-                f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , i32 > ({ use crate :: __staged :: persist_pullup :: tests :: * ; | v | v + 1 }),
+                f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , i32 > ({ use crate :: __staged :: rewrites :: persist_pullup :: tests :: * ; | v | v + 1 }),
                 input: Tee {
                     inner: <tee>: Source {
                         source: Iter(
-                            { use crate :: __staged :: persist_pullup :: tests :: * ; 0 .. 10 },
+                            { use crate :: __staged :: rewrites :: persist_pullup :: tests :: * ; 0 .. 10 },
                         ),
                         location_kind: Process(
                             0,

hydroflow_plus/src/snapshots/hydroflow_plus__persist_pullup__tests__persist_pullup_behind_tee.snap → hydroflow_plus/src/rewrites/snapshots/hydroflow_plus__rewrites__persist_pullup__tests__persist_pullup_behind_tee.snap

@@ -1,21 +1,21 @@
 ---
-source: hydroflow_plus/src/persist_pullup.rs
+source: hydroflow_plus/src/rewrites/persist_pullup.rs
 expression: built.ir()
 ---
 [
     ForEach {
-        f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , () > ({ use crate :: __staged :: persist_pullup :: tests :: * ; | n | println ! ("{}" , n) }),
+        f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , () > ({ use crate :: __staged :: rewrites :: persist_pullup :: tests :: * ; | n | println ! ("{}" , n) }),
         input: Unpersist(
             Persist(
                 Map {
-                    f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , i32 > ({ use crate :: __staged :: persist_pullup :: tests :: * ; | v | v + 1 }),
+                    f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , i32 > ({ use crate :: __staged :: rewrites :: persist_pullup :: tests :: * ; | v | v + 1 }),
                     input: Tee {
                         inner: <tee>: Persist(
                             Unpersist(
                                 Persist(
                                     Source {
                                         source: Iter(
-                                            { use crate :: __staged :: persist_pullup :: tests :: * ; 0 .. 10 },
+                                            { use crate :: __staged :: rewrites :: persist_pullup :: tests :: * ; 0 .. 10 },
                                         ),
                                         location_kind: Process(
                                             0,
@@ -30,18 +30,18 @@ expression: built.ir()
         ),
     },
     ForEach {
-        f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , () > ({ use crate :: __staged :: persist_pullup :: tests :: * ; | n | println ! ("{}" , n) }),
+        f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , () > ({ use crate :: __staged :: rewrites :: persist_pullup :: tests :: * ; | n | println ! ("{}" , n) }),
         input: Unpersist(
             Persist(
                 Map {
-                    f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , i32 > ({ use crate :: __staged :: persist_pullup :: tests :: * ; | v | v + 1 }),
+                    f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , i32 > ({ use crate :: __staged :: rewrites :: persist_pullup :: tests :: * ; | v | v + 1 }),
                     input: Tee {
                         inner: <tee>: Persist(
                             Unpersist(
                                 Persist(
                                     Source {
                                         source: Iter(
-                                            { use crate :: __staged :: persist_pullup :: tests :: * ; 0 .. 10 },
+                                            { use crate :: __staged :: rewrites :: persist_pullup :: tests :: * ; 0 .. 10 },
                                         ),
                                         location_kind: Process(
                                             0,

hydroflow_plus/src/rewrites/snapshots/hydroflow_plus__rewrites__persist_pullup__tests__persist_pullup_behind_tee@surface_graph_0.snap

@@ -0,0 +1,20 @@
+---
+source: hydroflow_plus/src/rewrites/persist_pullup.rs
+expression: graph.surface_syntax_string()
+---
+1v1 = source_iter ({ use crate :: __staged :: rewrites :: persist_pullup :: tests :: * ; 0 .. 10 });
+2v1 = tee ();
+3v1 = map (stageleft :: runtime_support :: fn1_type_hint :: < i32 , i32 > ({ use crate :: __staged :: rewrites :: persist_pullup :: tests :: * ; | v | v + 1 }));
+4v1 = persist :: < 'static > ();
+5v1 = for_each (stageleft :: runtime_support :: fn1_type_hint :: < i32 , () > ({ use crate :: __staged :: rewrites :: persist_pullup :: tests :: * ; | n | println ! ("{}" , n) }));
+6v1 = map (stageleft :: runtime_support :: fn1_type_hint :: < i32 , i32 > ({ use crate :: __staged :: rewrites :: persist_pullup :: tests :: * ; | v | v + 1 }));
+7v1 = persist :: < 'static > ();
+8v1 = for_each (stageleft :: runtime_support :: fn1_type_hint :: < i32 , () > ({ use crate :: __staged :: rewrites :: persist_pullup :: tests :: * ; | n | println ! ("{}" , n) }));
+
+1v1 -> 2v1;
+2v1 -> 3v1;
+3v1 -> 4v1;
+4v1 -> 5v1;
+2v1 -> 6v1;
+6v1 -> 7v1;
+7v1 -> 8v1;

hydroflow_plus/src/snapshots/hydroflow_plus__persist_pullup__tests__persist_pullup_through_map-2.snap → hydroflow_plus/src/rewrites/snapshots/hydroflow_plus__rewrites__persist_pullup__tests__persist_pullup_through_map-2.snap

@@ -1,15 +1,15 @@
 ---
-source: hydroflow_plus/src/persist_pullup.rs
+source: hydroflow_plus/src/rewrites/persist_pullup.rs
 expression: optimized.ir()
 ---
 [
     ForEach {
-        f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , () > ({ use crate :: __staged :: persist_pullup :: tests :: * ; | n | println ! ("{}" , n) }),
+        f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , () > ({ use crate :: __staged :: rewrites :: persist_pullup :: tests :: * ; | n | println ! ("{}" , n) }),
         input: Map {
-            f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , i32 > ({ use crate :: __staged :: persist_pullup :: tests :: * ; | v | v + 1 }),
+            f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , i32 > ({ use crate :: __staged :: rewrites :: persist_pullup :: tests :: * ; | v | v + 1 }),
             input: Source {
                 source: Iter(
-                    { use crate :: __staged :: persist_pullup :: tests :: * ; 0 .. 10 },
+                    { use crate :: __staged :: rewrites :: persist_pullup :: tests :: * ; 0 .. 10 },
                 ),
                 location_kind: Process(
                     0,

hydroflow_plus/src/snapshots/hydroflow_plus__persist_pullup__tests__persist_pullup_through_map.snap → hydroflow_plus/src/rewrites/snapshots/hydroflow_plus__rewrites__persist_pullup__tests__persist_pullup_through_map.snap

@@ -1,17 +1,17 @@
 ---
-source: hydroflow_plus/src/persist_pullup.rs
+source: hydroflow_plus/src/rewrites/persist_pullup.rs
 expression: built.ir()
 ---
 [
     ForEach {
-        f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , () > ({ use crate :: __staged :: persist_pullup :: tests :: * ; | n | println ! ("{}" , n) }),
+        f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , () > ({ use crate :: __staged :: rewrites :: persist_pullup :: tests :: * ; | n | println ! ("{}" , n) }),
         input: Unpersist(
             Map {
-                f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , i32 > ({ use crate :: __staged :: persist_pullup :: tests :: * ; | v | v + 1 }),
+                f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , i32 > ({ use crate :: __staged :: rewrites :: persist_pullup :: tests :: * ; | v | v + 1 }),
                 input: Persist(
                     Source {
                         source: Iter(
-                            { use crate :: __staged :: persist_pullup :: tests :: * ; 0 .. 10 },
+                            { use crate :: __staged :: rewrites :: persist_pullup :: tests :: * ; 0 .. 10 },
                         ),
                         location_kind: Process(
                             0,

hydroflow_plus/src/rewrites/snapshots/hydroflow_plus__rewrites__persist_pullup__tests__persist_pullup_through_map@surface_graph_0.snap

@@ -0,0 +1,10 @@
+---
+source: hydroflow_plus/src/rewrites/persist_pullup.rs
+expression: graph.surface_syntax_string()
+---
+1v1 = source_iter ({ use crate :: __staged :: rewrites :: persist_pullup :: tests :: * ; 0 .. 10 });
+2v1 = map (stageleft :: runtime_support :: fn1_type_hint :: < i32 , i32 > ({ use crate :: __staged :: rewrites :: persist_pullup :: tests :: * ; | v | v + 1 }));
+3v1 = for_each (stageleft :: runtime_support :: fn1_type_hint :: < i32 , () > ({ use crate :: __staged :: rewrites :: persist_pullup :: tests :: * ; | n | println ! ("{}" , n) }));
+
+1v1 -> 2v1;
+2v1 -> 3v1;

hydroflow_plus/src/rewrites/snapshots/hydroflow_plus__rewrites__profiler__tests__profiler_wrapping_all_operators-2.snap

@@ -0,0 +1,26 @@
+---
+source: hydroflow_plus/src/rewrites/profiler.rs
+expression: "&pushed_down.ir()"
+---
+[
+    ForEach {
+        f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , () > ({ use crate :: __staged :: rewrites :: profiler :: tests :: * ; | n | println ! ("{}" , n) }),
+        input: Inspect {
+            f: { use crate :: __staged :: rewrites :: profiler :: * ; let counter_queue = Fake ; let counters = Fake ; let my_id = 0u32 ; { counter_queue . borrow () . unbounded_send ((my_id as usize , counters . borrow () [my_id as usize])) . unwrap () ; counters . borrow_mut () [my_id as usize] = 0 ; move | _ | { myself :: increment_counter (& mut counters . borrow_mut () [my_id as usize]) ; } } },
+            input: Map {
+                f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , i32 > ({ use crate :: __staged :: rewrites :: profiler :: tests :: * ; | v | v + 1 }),
+                input: Inspect {
+                    f: { use crate :: __staged :: rewrites :: profiler :: * ; let counter_queue = Fake ; let counters = Fake ; let my_id = 1u32 ; { counter_queue . borrow () . unbounded_send ((my_id as usize , counters . borrow () [my_id as usize])) . unwrap () ; counters . borrow_mut () [my_id as usize] = 0 ; move | _ | { myself :: increment_counter (& mut counters . borrow_mut () [my_id as usize]) ; } } },
+                    input: Source {
+                        source: Iter(
+                            { use crate :: __staged :: rewrites :: profiler :: tests :: * ; 0 .. 10 },
+                        ),
+                        location_kind: Process(
+                            0,
+                        ),
+                    },
+                },
+            },
+        },
+    },
+]

hydroflow_plus/src/snapshots/hydroflow_plus__profiler__tests__profiler_wrapping_all_operators.snap → hydroflow_plus/src/rewrites/snapshots/hydroflow_plus__rewrites__profiler__tests__profiler_wrapping_all_operators.snap

@@ -1,17 +1,17 @@
 ---
-source: hydroflow_plus/src/profiler.rs
+source: hydroflow_plus/src/rewrites/profiler.rs
 expression: "&built.ir()"
 ---
 [
     ForEach {
-        f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , () > ({ use crate :: __staged :: profiler :: tests :: * ; | n | println ! ("{}" , n) }),
+        f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , () > ({ use crate :: __staged :: rewrites :: profiler :: tests :: * ; | n | println ! ("{}" , n) }),
         input: Unpersist(
             Map {
-                f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , i32 > ({ use crate :: __staged :: profiler :: tests :: * ; | v | v + 1 }),
+                f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , i32 > ({ use crate :: __staged :: rewrites :: profiler :: tests :: * ; | v | v + 1 }),
                 input: Persist(
                     Source {
                         source: Iter(
-                            { use crate :: __staged :: profiler :: tests :: * ; 0 .. 10 },
+                            { use crate :: __staged :: rewrites :: profiler :: tests :: * ; 0 .. 10 },
                         ),
                         location_kind: Process(
                             0,

hydroflow_plus/src/snapshots/hydroflow_plus__properties__tests__property_optimized.snap → hydroflow_plus/src/rewrites/snapshots/hydroflow_plus__rewrites__properties__tests__property_optimized.snap

@@ -1,18 +1,18 @@
 ---
-source: hydroflow_plus/src/properties.rs
+source: hydroflow_plus/src/rewrites/properties.rs
 expression: built.ir()
 ---
 [
     ForEach {
-        f: stageleft :: runtime_support :: fn1_type_hint :: < (std :: string :: String , i32) , () > ({ use crate :: __staged :: properties :: tests :: * ; | (string , count) | println ! ("{}: {}" , string , count) }),
+        f: stageleft :: runtime_support :: fn1_type_hint :: < (std :: string :: String , i32) , () > ({ use crate :: __staged :: rewrites :: properties :: tests :: * ; | (string , count) | println ! ("{}: {}" , string , count) }),
         input: FoldKeyed {
-            init: stageleft :: runtime_support :: fn0_type_hint :: < i32 > ({ use crate :: __staged :: properties :: tests :: * ; | | 0 }),
-            acc: stageleft :: runtime_support :: fn2_borrow_mut_type_hint :: < i32 , () , () > ({ use crate :: __staged :: properties :: tests :: * ; | count : & mut i32 , _ | * count += 1 }),
+            init: stageleft :: runtime_support :: fn0_type_hint :: < i32 > ({ use crate :: __staged :: rewrites :: properties :: tests :: * ; | | 0 }),
+            acc: stageleft :: runtime_support :: fn2_borrow_mut_type_hint :: < i32 , () , () > ({ use crate :: __staged :: rewrites :: properties :: tests :: * ; | count : & mut i32 , _ | * count += 1 }),
             input: Map {
-                f: stageleft :: runtime_support :: fn1_type_hint :: < std :: string :: String , (std :: string :: String , ()) > ({ use crate :: __staged :: properties :: tests :: * ; | string : String | (string , ()) }),
+                f: stageleft :: runtime_support :: fn1_type_hint :: < std :: string :: String , (std :: string :: String , ()) > ({ use crate :: __staged :: rewrites :: properties :: tests :: * ; | string : String | (string , ()) }),
                 input: Source {
                     source: Iter(
-                        { use crate :: __staged :: properties :: tests :: * ; vec ! [] },
+                        { use crate :: __staged :: rewrites :: properties :: tests :: * ; vec ! [] },
                     ),
                     location_kind: Process(
                         0,