feat(hydroflow_lang): Add FloTypes to operators, for source check

hydro-project · Nov 26, 2024 · 76726ab · 76726ab
1 parent 31be16a
commit 76726ab
Show file tree

Hide file tree

Showing 70 changed files with 102 additions and 17 deletions.
diff --git a/hydroflow_lang/src/graph/flat_graph_builder.rs b/hydroflow_lang/src/graph/flat_graph_builder.rs
@@ -14,6 +14,7 @@ use syn::{Error, Ident, ItemUse};
 use super::ops::defer_tick::DEFER_TICK;
 use super::ops::source_iter::SOURCE_ITER;
 use super::ops::source_stream::SOURCE_STREAM;
+use super::ops::FloType;
 use super::{GraphEdgeId, GraphLoopId, GraphNode, GraphNodeId, HydroflowGraph, PortIndexValue};
 use crate::diagnostic::{Diagnostic, Level};
 use crate::graph::graph_algorithms;
@@ -937,10 +938,9 @@ impl FlatGraphBuilder {
             let Some(_loop_id) = self.flat_graph.node_loop(node_id) else {
                 continue;
             };
-            // TODO(mingwei): don't hardcode source names, add prop to `op_constraints`.
-            if SOURCE_STREAM.name == op_inst.op_constraints.name
-                || SOURCE_ITER.name == op_inst.op_constraints.name
-            {
+
+            // Source operators must be at the top level.
+            if Some(FloType::Source) == op_inst.op_constraints.flo_type {
                 self.diagnostics.push(Diagnostic::spanned(
                     node.span(),
                     Level::Error,

diff --git a/hydroflow_lang/src/graph/ops/_lattice_fold_batch.rs b/hydroflow_lang/src/graph/ops/_lattice_fold_batch.rs
@@ -42,6 +42,7 @@ pub const _LATTICE_FOLD_BATCH: OperatorConstraints = OperatorConstraints {
     num_args: 0,
     is_external_input: false,
     has_singleton_output: false,
+    flo_type: None,
     ports_inn: Some(|| PortListSpec::Fixed(parse_quote! { input, signal })),
     ports_out: None,
     input_delaytype_fn: |_| Some(DelayType::MonotoneAccum),

diff --git a/hydroflow_lang/src/graph/ops/_lattice_join_fused_join.rs b/hydroflow_lang/src/graph/ops/_lattice_join_fused_join.rs
@@ -86,6 +86,7 @@ pub const _LATTICE_JOIN_FUSED_JOIN: OperatorConstraints = OperatorConstraints {
     type_args: &(2..=2),
     is_external_input: false,
     has_singleton_output: false,
+    flo_type: None,
     ports_inn: Some(|| super::PortListSpec::Fixed(parse_quote! { 0, 1 })),
     ports_out: None,
     input_delaytype_fn: |_| Some(DelayType::MonotoneAccum),

diff --git a/hydroflow_lang/src/graph/ops/anti_join.rs b/hydroflow_lang/src/graph/ops/anti_join.rs
@@ -37,6 +37,7 @@ pub const ANTI_JOIN: OperatorConstraints = OperatorConstraints {
     // to prevent reading uncleared data if this subgraph doesn't run.
     // https://github.com/hydro-project/hydroflow/issues/1298
     has_singleton_output: false,
+    flo_type: None,
     ports_inn: Some(|| super::PortListSpec::Fixed(parse_quote! { pos, neg })),
     ports_out: None,
     input_delaytype_fn: |idx| match idx {

diff --git a/hydroflow_lang/src/graph/ops/anti_join_multiset.rs b/hydroflow_lang/src/graph/ops/anti_join_multiset.rs
@@ -37,6 +37,7 @@ pub const ANTI_JOIN_MULTISET: OperatorConstraints = OperatorConstraints {
     // to prevent reading uncleared data if this subgraph doesn't run.
     // https://github.com/hydro-project/hydroflow/issues/1298
     has_singleton_output: false,
+    flo_type: None,
     ports_inn: Some(|| super::PortListSpec::Fixed(parse_quote! { pos, neg })),
     ports_out: None,
     input_delaytype_fn: |idx| match idx {

diff --git a/hydroflow_lang/src/graph/ops/assert.rs b/hydroflow_lang/src/graph/ops/assert.rs
@@ -26,6 +26,7 @@ pub const ASSERT: OperatorConstraints = OperatorConstraints {
     type_args: RANGE_0,
     is_external_input: false,
     has_singleton_output: false,
+    flo_type: None,
     ports_inn: None,
     ports_out: None,
     input_delaytype_fn: |_| None,

diff --git a/hydroflow_lang/src/graph/ops/assert_eq.rs b/hydroflow_lang/src/graph/ops/assert_eq.rs
@@ -37,6 +37,7 @@ pub const ASSERT_EQ: OperatorConstraints = OperatorConstraints {
     type_args: RANGE_0,
     is_external_input: false,
     has_singleton_output: false,
+    flo_type: None,
     ports_inn: None,
     ports_out: None,
     input_delaytype_fn: |_| None,

diff --git a/hydroflow_lang/src/graph/ops/chain.rs b/hydroflow_lang/src/graph/ops/chain.rs
@@ -30,6 +30,7 @@ pub const CHAIN: OperatorConstraints = OperatorConstraints {
     num_args: 0,
     is_external_input: false,
     has_singleton_output: false,
+    flo_type: None,
     ports_inn: None,
     ports_out: None,
     input_delaytype_fn: |idx| match idx {

diff --git a/hydroflow_lang/src/graph/ops/cross_join.rs b/hydroflow_lang/src/graph/ops/cross_join.rs
@@ -45,6 +45,7 @@ pub const CROSS_JOIN: OperatorConstraints = OperatorConstraints {
     type_args: &(0..=1),
     is_external_input: false,
     has_singleton_output: false,
+    flo_type: None,
     ports_inn: Some(|| super::PortListSpec::Fixed(parse_quote! { 0, 1 })),
     ports_out: None,
     input_delaytype_fn: |_| None,

diff --git a/hydroflow_lang/src/graph/ops/cross_join_multiset.rs b/hydroflow_lang/src/graph/ops/cross_join_multiset.rs
@@ -34,6 +34,7 @@ pub const CROSS_JOIN_MULTISET: OperatorConstraints = OperatorConstraints {
     type_args: &(0..=1),
     is_external_input: false,
     has_singleton_output: false,
+    flo_type: None,
     ports_inn: Some(|| super::PortListSpec::Fixed(parse_quote! { 0, 1 })),
     ports_out: None,
     input_delaytype_fn: |_| None,

diff --git a/hydroflow_lang/src/graph/ops/cross_singleton.rs b/hydroflow_lang/src/graph/ops/cross_singleton.rs
@@ -39,6 +39,7 @@ pub const CROSS_SINGLETON: OperatorConstraints = OperatorConstraints {
     num_args: 0,
     is_external_input: false,
     has_singleton_output: false,
+    flo_type: None,
     ports_inn: Some(|| super::PortListSpec::Fixed(parse_quote! { input, single })),
     ports_out: None,
     input_delaytype_fn: |idx| match idx {

diff --git a/hydroflow_lang/src/graph/ops/defer_signal.rs b/hydroflow_lang/src/graph/ops/defer_signal.rs
@@ -34,6 +34,7 @@ pub const DEFER_SIGNAL: OperatorConstraints = OperatorConstraints {
     num_args: 0,
     is_external_input: false,
     has_singleton_output: false,
+    flo_type: None,
     ports_inn: Some(|| super::PortListSpec::Fixed(parse_quote! { input, signal })),
     ports_out: None,
     input_delaytype_fn: |_| Some(DelayType::Stratum),

diff --git a/hydroflow_lang/src/graph/ops/defer_tick.rs b/hydroflow_lang/src/graph/ops/defer_tick.rs
@@ -72,6 +72,7 @@ pub const DEFER_TICK: OperatorConstraints = OperatorConstraints {
     type_args: &(0..=1),
     is_external_input: false,
     has_singleton_output: false,
+    flo_type: None,
     ports_inn: None,
     ports_out: None,
     input_delaytype_fn: |_| Some(DelayType::Tick),

diff --git a/hydroflow_lang/src/graph/ops/defer_tick_lazy.rs b/hydroflow_lang/src/graph/ops/defer_tick_lazy.rs
@@ -17,6 +17,7 @@ pub const DEFER_TICK_LAZY: OperatorConstraints = OperatorConstraints {
     type_args: RANGE_0,
     is_external_input: false,
     has_singleton_output: false,
+    flo_type: None,
     ports_inn: None,
     ports_out: None,
     input_delaytype_fn: |_| Some(DelayType::TickLazy),

diff --git a/hydroflow_lang/src/graph/ops/demux.rs b/hydroflow_lang/src/graph/ops/demux.rs
@@ -53,6 +53,7 @@ pub const DEMUX: OperatorConstraints = OperatorConstraints {
     type_args: RANGE_0,
     is_external_input: false,
     has_singleton_output: false,
+    flo_type: None,
     ports_inn: None,
     ports_out: Some(|| PortListSpec::Variadic),
     input_delaytype_fn: |_| None,

diff --git a/hydroflow_lang/src/graph/ops/demux_enum.rs b/hydroflow_lang/src/graph/ops/demux_enum.rs
@@ -52,6 +52,7 @@ pub const DEMUX_ENUM: OperatorConstraints = OperatorConstraints {
     type_args: RANGE_1,
     is_external_input: false,
     has_singleton_output: false,
+    flo_type: None,
     ports_inn: None,
     ports_out: Some(|| PortListSpec::Variadic),
     input_delaytype_fn: |_| None,

diff --git a/hydroflow_lang/src/graph/ops/dest_file.rs b/hydroflow_lang/src/graph/ops/dest_file.rs
@@ -32,6 +32,7 @@ pub const DEST_FILE: OperatorConstraints = OperatorConstraints {
     type_args: RANGE_0,
     is_external_input: false,
     has_singleton_output: false,
+    flo_type: None,
     ports_inn: None,
     ports_out: None,
     input_delaytype_fn: |_| None,

diff --git a/hydroflow_lang/src/graph/ops/dest_sink.rs b/hydroflow_lang/src/graph/ops/dest_sink.rs
@@ -93,6 +93,7 @@ pub const DEST_SINK: OperatorConstraints = OperatorConstraints {
     type_args: RANGE_0,
     is_external_input: false,
     has_singleton_output: false,
+    flo_type: None,
     ports_inn: None,
     ports_out: None,
     input_delaytype_fn: |_| None,

diff --git a/hydroflow_lang/src/graph/ops/dest_sink_serde.rs b/hydroflow_lang/src/graph/ops/dest_sink_serde.rs
@@ -35,6 +35,7 @@ pub const DEST_SINK_SERDE: OperatorConstraints = OperatorConstraints {
     type_args: RANGE_0,
     is_external_input: false,
     has_singleton_output: false,
+    flo_type: None,
     ports_inn: None,
     ports_out: None,
     input_delaytype_fn: |_| None,

diff --git a/hydroflow_lang/src/graph/ops/difference.rs b/hydroflow_lang/src/graph/ops/difference.rs
@@ -34,6 +34,7 @@ pub const DIFFERENCE: OperatorConstraints = OperatorConstraints {
     type_args: RANGE_0,
     is_external_input: false,
     has_singleton_output: false,
+    flo_type: None,
     ports_inn: Some(|| super::PortListSpec::Fixed(parse_quote! { pos, neg })),
     ports_out: None,
     input_delaytype_fn: |idx| match idx {

diff --git a/hydroflow_lang/src/graph/ops/difference_multiset.rs b/hydroflow_lang/src/graph/ops/difference_multiset.rs
@@ -35,6 +35,7 @@ pub const DIFFERENCE_MULTISET: OperatorConstraints = OperatorConstraints {
     type_args: RANGE_0,
     is_external_input: false,
     has_singleton_output: false,
+    flo_type: None,
     ports_inn: Some(|| super::PortListSpec::Fixed(parse_quote! { pos, neg })),
     ports_out: None,
     input_delaytype_fn: |idx| match idx {

diff --git a/hydroflow_lang/src/graph/ops/enumerate.rs b/hydroflow_lang/src/graph/ops/enumerate.rs
@@ -32,6 +32,7 @@ pub const ENUMERATE: OperatorConstraints = OperatorConstraints {
     type_args: RANGE_0,
     is_external_input: false,
     has_singleton_output: false,
+    flo_type: None,
     ports_inn: None,
     ports_out: None,
     input_delaytype_fn: |_| None,

diff --git a/hydroflow_lang/src/graph/ops/filter.rs b/hydroflow_lang/src/graph/ops/filter.rs
@@ -30,6 +30,7 @@ pub const FILTER: OperatorConstraints = OperatorConstraints {
     type_args: RANGE_0,
     is_external_input: false,
     has_singleton_output: false,
+    flo_type: None,
     ports_inn: None,
     ports_out: None,
     input_delaytype_fn: |_| None,

diff --git a/hydroflow_lang/src/graph/ops/filter_map.rs b/hydroflow_lang/src/graph/ops/filter_map.rs
@@ -28,6 +28,7 @@ pub const FILTER_MAP: OperatorConstraints = OperatorConstraints {
     type_args: RANGE_0,
     is_external_input: false,
     has_singleton_output: false,
+    flo_type: None,
     ports_inn: None,
     ports_out: None,
     input_delaytype_fn: |_| None,

diff --git a/hydroflow_lang/src/graph/ops/flat_map.rs b/hydroflow_lang/src/graph/ops/flat_map.rs
@@ -32,6 +32,7 @@ pub const FLAT_MAP: OperatorConstraints = OperatorConstraints {
     type_args: RANGE_0,
     is_external_input: false,
     has_singleton_output: false,
+    flo_type: None,
     ports_inn: None,
     ports_out: None,
     input_delaytype_fn: |_| None,

diff --git a/hydroflow_lang/src/graph/ops/flatten.rs b/hydroflow_lang/src/graph/ops/flatten.rs
@@ -27,6 +27,7 @@ pub const FLATTEN: OperatorConstraints = OperatorConstraints {
     type_args: RANGE_0,
     is_external_input: false,
     has_singleton_output: false,
+    flo_type: None,
     ports_inn: None,
     ports_out: None,
     input_delaytype_fn: |_| None,

diff --git a/hydroflow_lang/src/graph/ops/fold.rs b/hydroflow_lang/src/graph/ops/fold.rs
@@ -45,6 +45,7 @@ pub const FOLD: OperatorConstraints = OperatorConstraints {
     type_args: RANGE_0,
     is_external_input: false,
     has_singleton_output: true,
+    flo_type: None,
     ports_inn: None,
     ports_out: None,
     input_delaytype_fn: |_| Some(DelayType::Stratum),

diff --git a/hydroflow_lang/src/graph/ops/fold_keyed.rs b/hydroflow_lang/src/graph/ops/fold_keyed.rs
@@ -80,6 +80,7 @@ pub const FOLD_KEYED: OperatorConstraints = OperatorConstraints {
     // to prevent reading uncleared data if this subgraph doesn't run.
     // https://github.com/hydro-project/hydroflow/issues/1298
     has_singleton_output: false,
+    flo_type: None,
     ports_inn: None,
     ports_out: None,
     input_delaytype_fn: |_| Some(DelayType::Stratum),

diff --git a/hydroflow_lang/src/graph/ops/for_each.rs b/hydroflow_lang/src/graph/ops/for_each.rs
@@ -30,6 +30,7 @@ pub const FOR_EACH: OperatorConstraints = OperatorConstraints {
     type_args: RANGE_0,
     is_external_input: false,
     has_singleton_output: false,
+    flo_type: None,
     ports_inn: None,
     ports_out: None,
     input_delaytype_fn: |_| None,

diff --git a/hydroflow_lang/src/graph/ops/identity.rs b/hydroflow_lang/src/graph/ops/identity.rs
@@ -33,6 +33,7 @@ pub const IDENTITY: OperatorConstraints = OperatorConstraints {
     type_args: &(0..=1),
     is_external_input: false,
     has_singleton_output: false,
+    flo_type: None,
     ports_inn: None,
     ports_out: None,
     input_delaytype_fn: |_| None,

diff --git a/hydroflow_lang/src/graph/ops/initialize.rs b/hydroflow_lang/src/graph/ops/initialize.rs
@@ -26,6 +26,7 @@ pub const INITIALIZE: OperatorConstraints = OperatorConstraints {
     type_args: RANGE_0,
     is_external_input: false,
     has_singleton_output: false,
+    flo_type: None,
     ports_inn: None,
     ports_out: None,
     input_delaytype_fn: |_| None,

diff --git a/hydroflow_lang/src/graph/ops/inspect.rs b/hydroflow_lang/src/graph/ops/inspect.rs
@@ -31,6 +31,7 @@ pub const INSPECT: OperatorConstraints = OperatorConstraints {
     type_args: RANGE_0,
     is_external_input: false,
     has_singleton_output: false,
+    flo_type: None,
     ports_inn: None,
     ports_out: None,
     input_delaytype_fn: |_| None,

diff --git a/hydroflow_lang/src/graph/ops/join.rs b/hydroflow_lang/src/graph/ops/join.rs
@@ -91,6 +91,7 @@ pub const JOIN: OperatorConstraints = OperatorConstraints {
     type_args: &(0..=1),
     is_external_input: false,
     has_singleton_output: false,
+    flo_type: None,
     ports_inn: Some(|| super::PortListSpec::Fixed(parse_quote! { 0, 1 })),
     ports_out: None,
     input_delaytype_fn: |_| None,

diff --git a/hydroflow_lang/src/graph/ops/join_fused.rs b/hydroflow_lang/src/graph/ops/join_fused.rs
@@ -100,6 +100,7 @@ pub const JOIN_FUSED: OperatorConstraints = OperatorConstraints {
     type_args: RANGE_0,
     is_external_input: false,
     has_singleton_output: false,
+    flo_type: None,
     ports_inn: Some(|| super::PortListSpec::Fixed(parse_quote! { 0, 1 })),
     ports_out: None,
     input_delaytype_fn: |_| Some(DelayType::Stratum),

diff --git a/hydroflow_lang/src/graph/ops/join_fused_lhs.rs b/hydroflow_lang/src/graph/ops/join_fused_lhs.rs
@@ -34,6 +34,7 @@ pub const JOIN_FUSED_LHS: OperatorConstraints = OperatorConstraints {
     type_args: RANGE_0,
     is_external_input: false,
     has_singleton_output: false,
+    flo_type: None,
     ports_inn: Some(|| super::PortListSpec::Fixed(parse_quote! { 0, 1 })),
     ports_out: None,
     input_delaytype_fn: |idx| match idx {

diff --git a/hydroflow_lang/src/graph/ops/join_fused_rhs.rs b/hydroflow_lang/src/graph/ops/join_fused_rhs.rs
@@ -21,6 +21,7 @@ pub const JOIN_FUSED_RHS: OperatorConstraints = OperatorConstraints {
     type_args: RANGE_0,
     is_external_input: false,
     has_singleton_output: false,
+    flo_type: None,
     ports_inn: Some(|| super::PortListSpec::Fixed(parse_quote! { 0, 1 })),
     ports_out: None,
     input_delaytype_fn: |idx| match idx {

diff --git a/hydroflow_lang/src/graph/ops/join_multiset.rs b/hydroflow_lang/src/graph/ops/join_multiset.rs
@@ -37,6 +37,7 @@ pub const JOIN_MULTISET: OperatorConstraints = OperatorConstraints {
     type_args: RANGE_0,
     is_external_input: false,
     has_singleton_output: false,
+    flo_type: None,
     ports_inn: Some(|| super::PortListSpec::Fixed(parse_quote! { 0, 1 })),
     ports_out: None,
     input_delaytype_fn: |_| None,

diff --git a/hydroflow_lang/src/graph/ops/lattice_bimorphism.rs b/hydroflow_lang/src/graph/ops/lattice_bimorphism.rs
@@ -52,6 +52,7 @@ pub const LATTICE_BIMORPHISM: OperatorConstraints = OperatorConstraints {
     type_args: RANGE_0,
     is_external_input: false,
     has_singleton_output: false,
+    flo_type: None,
     ports_inn: Some(|| super::PortListSpec::Fixed(parse_quote! { 0, 1 })),
     ports_out: None,
     input_delaytype_fn: |_| None,

diff --git a/hydroflow_lang/src/graph/ops/lattice_fold.rs b/hydroflow_lang/src/graph/ops/lattice_fold.rs
@@ -39,6 +39,7 @@ pub const LATTICE_FOLD: OperatorConstraints = OperatorConstraints {
     type_args: RANGE_0,
     is_external_input: false,
     has_singleton_output: false,
+    flo_type: None,
     ports_inn: None,
     ports_out: None,
     input_delaytype_fn: |_| Some(DelayType::MonotoneAccum),

diff --git a/hydroflow_lang/src/graph/ops/lattice_reduce.rs b/hydroflow_lang/src/graph/ops/lattice_reduce.rs
@@ -40,6 +40,7 @@ pub const LATTICE_REDUCE: OperatorConstraints = OperatorConstraints {
     type_args: RANGE_0,
     is_external_input: false,
     has_singleton_output: false,
+    flo_type: None,
     ports_inn: None,
     ports_out: None,
     input_delaytype_fn: |_| Some(DelayType::MonotoneAccum),

diff --git a/hydroflow_lang/src/graph/ops/map.rs b/hydroflow_lang/src/graph/ops/map.rs
@@ -32,6 +32,7 @@ pub const MAP: OperatorConstraints = OperatorConstraints {
     type_args: RANGE_0,
     is_external_input: false,
     has_singleton_output: false,
+    flo_type: None,
     ports_inn: None,
     ports_out: None,
     input_delaytype_fn: |_| None,

diff --git a/hydroflow_lang/src/graph/ops/mod.rs b/hydroflow_lang/src/graph/ops/mod.rs
@@ -70,6 +70,8 @@ pub struct OperatorConstraints {
     /// If true, [`WriteContextArgs::singleton_output_ident`] will be set to a meaningful value in
     /// the [`Self::write_fn`] invocation.
     pub has_singleton_output: bool,
+    /// Flo semantics type.
+    pub flo_type: Option<FloType>,
 
     /// What named or numbered input ports to expect?
     pub ports_inn: Option<fn() -> PortListSpec>,
@@ -541,3 +543,19 @@ impl OperatorCategory {
         }
     }
 }
+
+/// Operator type for Flo semantics.
+#[derive(Clone, Copy, PartialOrd, Ord, PartialEq, Eq, Debug)]
+pub enum FloType {
+<<<<<<< HEAD
+    Source,
+    Windowing,
+=======
+    /// A source operator, which must be at the top level.
+    Source,
+    /// A windowing operator, for moving data into a loop context.
+    Windowing,
+    /// An un-windowing operator, for moving data out of a loop context.
+>>>>>>> a44ec015fac (feat(hydroflow_lang): Add `FloType`s to operators, for source check)
+    Unwindowing,
+}
diff --git a/hydroflow_lang/src/graph/ops/multiset_delta.rs b/hydroflow_lang/src/graph/ops/multiset_delta.rs
@@ -46,6 +46,7 @@ pub const MULTISET_DELTA: OperatorConstraints = OperatorConstraints {
     // https://github.com/hydro-project/hydroflow/issues/1298
     // If `'tick` lifetimes are added.
     has_singleton_output: false,
+    flo_type: None,
     ports_inn: None,
     ports_out: None,
     input_delaytype_fn: |_| None,

diff --git a/hydroflow_lang/src/graph/ops/next_stratum.rs b/hydroflow_lang/src/graph/ops/next_stratum.rs
@@ -20,6 +20,7 @@ pub const NEXT_STRATUM: OperatorConstraints = OperatorConstraints {
     type_args: RANGE_0,
     is_external_input: false,
     has_singleton_output: false,
+    flo_type: None,
     ports_inn: None,
     ports_out: None,
     input_delaytype_fn: |_| Some(DelayType::Stratum),

diff --git a/hydroflow_lang/src/graph/ops/null.rs b/hydroflow_lang/src/graph/ops/null.rs
@@ -26,6 +26,7 @@ pub const NULL: OperatorConstraints = OperatorConstraints {
     type_args: &(0..=1),
     is_external_input: false,
     has_singleton_output: false,
+    flo_type: None,
     ports_inn: None,
     ports_out: None,
     input_delaytype_fn: |_| None,