dfir_lang/graph/ops/
mod.rs

1//! DFIR's operators
2
3use std::collections::HashMap;
4use std::fmt::{Debug, Display};
5use std::ops::{Bound, RangeBounds};
6use std::sync::OnceLock;
7
8use documented::DocumentedVariants;
9use proc_macro2::{Ident, Literal, Span, TokenStream};
10use quote::quote_spanned;
11use serde::{Deserialize, Serialize};
12use slotmap::Key;
13use syn::punctuated::Punctuated;
14use syn::{Expr, Token, parse_quote_spanned};
15
16use super::{
17    AsCodeOptions, GraphLoopId, GraphNode, GraphNodeId, GraphSubgraphId, OpInstGenerics,
18    OperatorInstance, PortIndexValue,
19};
20use crate::diagnostic::{Diagnostic, Diagnostics, Level};
21use crate::parse::{Operator, PortIndex};
22
23/// The delay (soft barrier) type, for each input to an operator if needed.
24#[derive(Clone, Copy, PartialOrd, Ord, PartialEq, Eq, Debug, Serialize, Deserialize)]
25pub enum DelayType {
26    /// Input must be collected over the previous tick.
27    Tick,
28    /// Input must be collected over the previous tick but also not cause a new tick to occur.
29    TickLazy,
30    /// Input must be collected over the previous loop iteration. Causes the loop to re-fire.
31    Loop,
32    /// Input must be collected over the previous loop iteration. Does NOT cause the loop to re-fire.
33    LoopLazy,
34}
35
36/// Specification of the named (or unnamed) ports for an operator's inputs or outputs.
37pub enum PortListSpec {
38    /// Any number of unnamed (or optionally named) ports.
39    Variadic,
40    /// A specific number of named ports.
41    Fixed(Punctuated<PortIndex, Token![,]>),
42}
43
44/// An instance of this struct represents a single dfir operator.
45pub struct OperatorConstraints {
46    /// Operator's name.
47    pub name: &'static str,
48    /// Operator categories, for docs.
49    pub categories: &'static [OperatorCategory],
50
51    // TODO: generic argument ranges.
52    /// Input argument range required to not show an error.
53    pub hard_range_inn: &'static dyn RangeTrait<usize>,
54    /// Input argument range required to not show a warning.
55    pub soft_range_inn: &'static dyn RangeTrait<usize>,
56    /// Output argument range required to not show an error.
57    pub hard_range_out: &'static dyn RangeTrait<usize>,
58    /// Output argument range required to not show an warning.
59    pub soft_range_out: &'static dyn RangeTrait<usize>,
60    /// Number of arguments i.e. `operator(a, b, c)` has `num_args = 3`.
61    pub num_args: usize,
62    /// How many persistence lifetime arguments can be provided.
63    pub persistence_args: &'static dyn RangeTrait<usize>,
64    // /// How many (non-persistence) lifetime arguments can be provided.
65    // pub lifetime_args: &'static dyn RangeTrait<usize>,
66    /// How many generic type arguments can be provided.
67    pub type_args: &'static dyn RangeTrait<usize>,
68    /// If this operator receives external inputs and therefore must be in
69    /// stratum 0.
70    pub is_external_input: bool,
71    /// Flo semantics type.
72    pub flo_type: Option<FloType>,
73
74    /// What named or numbered input ports to expect?
75    pub ports_inn: Option<fn() -> PortListSpec>,
76    /// What named or numbered output ports to expect?
77    pub ports_out: Option<fn() -> PortListSpec>,
78
79    /// Determines if this input must be preceeded by a stratum barrier.
80    pub input_delaytype_fn: fn(&PortIndexValue) -> Option<DelayType>,
81    /// The operator's codegen. Returns code that is emited is several different locations. See [`OperatorWriteOutput`].
82    pub write_fn: WriteFn,
83}
84
85/// Type alias for [`OperatorConstraints::write_fn`]'s type.
86pub type WriteFn = fn(&WriteContextArgs<'_>, &mut Diagnostics) -> Result<OperatorWriteOutput, ()>;
87
88impl Debug for OperatorConstraints {
89    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
90        f.debug_struct("OperatorConstraints")
91            .field("name", &self.name)
92            .field("hard_range_inn", &self.hard_range_inn)
93            .field("soft_range_inn", &self.soft_range_inn)
94            .field("hard_range_out", &self.hard_range_out)
95            .field("soft_range_out", &self.soft_range_out)
96            .field("num_args", &self.num_args)
97            .field("persistence_args", &self.persistence_args)
98            .field("type_args", &self.type_args)
99            .field("is_external_input", &self.is_external_input)
100            .field("ports_inn", &self.ports_inn)
101            .field("ports_out", &self.ports_out)
102            // .field("input_delaytype_fn", &self.input_delaytype_fn)
103            // .field("flow_prop_fn", &self.flow_prop_fn)
104            // .field("write_fn", &self.write_fn)
105            .finish()
106    }
107}
108
109/// The code generated and returned by a [`OperatorConstraints::write_fn`].
110///
111/// This type intentionally does not implement [`Default`]. Constructing it with
112/// `..` or `..Default::default()` silently drops newly added fields (this once
113/// dropped `write_tick_end` from `lattice_reduce`). List every field explicitly
114/// when constructing or destructuring.
115pub struct OperatorWriteOutput {
116    /// Code which runs once outside any subgraphs to set up external state
117    /// (channels, network connections, etc.) to be used by the subgraph.
118    pub write_prologue: TokenStream,
119    /// Iterator (or pusherator) code inside the subgraphs. The code for each
120    /// operator is emitted in order.
121    ///
122    /// Emitted code should assign to [`WriteContextArgs::ident`] and use
123    /// [`WriteContextArgs::inputs`] (pull `Stream`s) or
124    /// [`WriteContextArgs::outputs`] (push `Sink`s).
125    pub write_iterator: TokenStream,
126    /// Code which runs after `Stream`s/`Sink`s have been run. Mainly for flushing IO.
127    pub write_iterator_after: TokenStream,
128    /// Code which runs at the end of each tick, after all subgraphs have run.
129    /// Used for resetting state with `'tick` persistence.
130    pub write_tick_end: TokenStream,
131}
132
133/// Convenience range: zero or more (any number).
134pub const RANGE_ANY: &'static dyn RangeTrait<usize> = &(0..);
135/// Convenience range: exactly zero.
136pub const RANGE_0: &'static dyn RangeTrait<usize> = &(0..=0);
137/// Convenience range: exactly one.
138pub const RANGE_1: &'static dyn RangeTrait<usize> = &(1..=1);
139
140/// Helper to write the `write_iterator` portion of [`OperatorConstraints::write_fn`] output for
141/// unary identity operators.
142///
143/// When no explicit type argument is given and
144/// [`AsCodeOptions::exclude_type_guards`](super::AsCodeOptions::exclude_type_guards) is set, the
145/// identity `check_input`/`check_output` type guard is omitted and the input/output is simply
146/// rebound.
147pub fn identity_write_iterator_fn(
148    &WriteContextArgs {
149        root,
150        op_span,
151        ident,
152        inputs,
153        outputs,
154        is_pull,
155        options,
156        op_inst:
157            OperatorInstance {
158                generics: OpInstGenerics { type_args, .. },
159                ..
160            },
161        ..
162    }: &WriteContextArgs<'_>,
163) -> TokenStream {
164    if type_args.is_empty() && options.exclude_type_guards {
165        // No user-provided type to pin down, so the guard would be a pure (verbose) no-op.
166        let src = if is_pull { &inputs[0] } else { &outputs[0] };
167        return quote_spanned! {op_span=>
168            let #ident = #src;
169        };
170    }
171
172    let generic_type = type_args
173        .first()
174        .map(quote::ToTokens::to_token_stream)
175        .unwrap_or_else(|| quote_spanned!(op_span=> _));
176
177    if is_pull {
178        let input = &inputs[0];
179        quote_spanned! {op_span=>
180            let #ident = {
181                fn check_input<Pull, Item>(pull: Pull) -> impl #root::dfir_pipes::pull::Pull<Item = Item, Meta = Pull::Meta, CanPend = Pull::CanPend, CanEnd = Pull::CanEnd>
182                where
183                    Pull: #root::dfir_pipes::pull::Pull<Item = Item>,
184                {
185                    pull
186                }
187                check_input::<_, #generic_type>(#input)
188            };
189        }
190    } else {
191        let output = &outputs[0];
192        quote_spanned! {op_span=>
193            let #ident = {
194                fn check_output<Psh, Item>(push: Psh) -> impl #root::dfir_pipes::push::Push<Item, (), CanPend = Psh::CanPend>
195                where
196                    Psh: #root::dfir_pipes::push::Push<Item, ()>,
197                {
198                    push
199                }
200                check_output::<_, #generic_type>(#output)
201            };
202        }
203    }
204}
205
206/// [`OperatorConstraints::write_fn`] for unary identity operators.
207pub const IDENTITY_WRITE_FN: WriteFn = |write_context_args, _| {
208    let write_iterator = identity_write_iterator_fn(write_context_args);
209    Ok(OperatorWriteOutput {
210        write_prologue: Default::default(),
211        write_iterator,
212        write_iterator_after: Default::default(),
213        write_tick_end: Default::default(),
214    })
215};
216
217/// Helper to write the `write_iterator` portion of [`OperatorConstraints::write_fn`] output for
218/// the null operator - an operator that ignores all inputs and produces no output.
219pub fn null_write_iterator_fn(
220    &WriteContextArgs {
221        root,
222        op_span,
223        ident,
224        inputs,
225        outputs,
226        is_pull,
227        op_inst:
228            OperatorInstance {
229                generics: OpInstGenerics { type_args, .. },
230                ..
231            },
232        ..
233    }: &WriteContextArgs<'_>,
234) -> TokenStream {
235    let default_type = parse_quote_spanned! {op_span=> _};
236    let iter_type = type_args.first().unwrap_or(&default_type);
237
238    if is_pull {
239        quote_spanned! {op_span=>
240            let #ident = #root::dfir_pipes::pull::poll_fn({
241                #(
242                    let mut #inputs = ::std::boxed::Box::pin(#inputs);
243                )*
244                move |_cx| {
245                    // Make sure to poll all `#inputs` to completion.
246                    // NOTE(mingwei): `null()` can only have 0 or 1 inputs, though.
247                    // TODO(mingwei): Do we actually need to poll to completion or can we short-circuit?
248                    #(
249                        let #inputs = #root::dfir_pipes::pull::Pull::pull(
250                            ::std::pin::Pin::as_mut(&mut #inputs),
251                            <_ as #root::dfir_pipes::Context>::from_task(_cx),
252                        );
253                    )*
254                    #(
255                        if let #root::dfir_pipes::pull::PullStep::Pending(_) = #inputs {
256                            return #root::dfir_pipes::pull::PullStep::Pending(#root::dfir_pipes::Yes);
257                        }
258                    )*
259                    #root::dfir_pipes::pull::PullStep::<_, _, #root::dfir_pipes::Yes, _>::Ended(#root::dfir_pipes::Yes)
260                }
261            });
262        }
263    } else {
264        quote_spanned! {op_span=>
265            #[allow(clippy::let_unit_value, clippy::ignored_unit_patterns)]
266            let _ = (#(#outputs),*);
267            let #ident = #root::dfir_pipes::push::for_each::<_, #iter_type>(::std::mem::drop::<#iter_type>);
268        }
269    }
270}
271
272/// [`OperatorConstraints::write_fn`] for the null operator - an operator that ignores all inputs
273/// and produces no output.
274pub const NULL_WRITE_FN: WriteFn = |write_context_args, _| {
275    let write_iterator = null_write_iterator_fn(write_context_args);
276    Ok(OperatorWriteOutput {
277        write_prologue: Default::default(),
278        write_iterator,
279        write_iterator_after: Default::default(),
280        write_tick_end: Default::default(),
281    })
282};
283
284macro_rules! declare_ops {
285    ( $( $mod:ident :: $op:ident, )* ) => {
286        $( pub(crate) mod $mod; )*
287        /// All DFIR operators.
288        pub const OPERATORS: &[OperatorConstraints] = &[
289            $( $mod :: $op, )*
290        ];
291    };
292}
293declare_ops![
294    all_iterations::ALL_ITERATIONS,
295    anti_join::ANTI_JOIN,
296    assert::ASSERT,
297    assert_eq::ASSERT_EQ,
298    batch::BATCH,
299    batch_eager::BATCH_EAGER,
300    batch_lazy::BATCH_LAZY,
301    chain::CHAIN,
302    chain_first_n::CHAIN_FIRST_N,
303    _counter::_COUNTER,
304    cross_join::CROSS_JOIN,
305    cross_join_multiset::CROSS_JOIN_MULTISET,
306    cross_singleton::CROSS_SINGLETON,
307    demux_enum::DEMUX_ENUM,
308    dest_file::DEST_FILE,
309    dest_sink::DEST_SINK,
310    dest_sink_serde::DEST_SINK_SERDE,
311    difference::DIFFERENCE,
312    enumerate::ENUMERATE,
313    filter::FILTER,
314    filter_map::FILTER_MAP,
315    flat_map::FLAT_MAP,
316    flat_map_stream_blocking::FLAT_MAP_STREAM_BLOCKING,
317    flatten::FLATTEN,
318    flatten_stream_blocking::FLATTEN_STREAM_BLOCKING,
319    fold::FOLD,
320    fold_no_replay::FOLD_NO_REPLAY,
321    for_each::FOR_EACH,
322    identity::IDENTITY,
323    initialize::INITIALIZE,
324    inspect::INSPECT,
325    iter_ref::ITER_REF,
326    join::JOIN,
327    join_fused::JOIN_FUSED,
328    join_fused_lhs::JOIN_FUSED_LHS,
329    join_fused_rhs::JOIN_FUSED_RHS,
330    join_multiset::JOIN_MULTISET,
331    join_multiset_half::JOIN_MULTISET_HALF,
332    fold_keyed::FOLD_KEYED,
333    reduce_keyed::REDUCE_KEYED,
334    lattice_bimorphism::LATTICE_BIMORPHISM,
335    _lattice_fold_batch::_LATTICE_FOLD_BATCH,
336    lattice_fold::LATTICE_FOLD,
337    _lattice_join_fused_join::_LATTICE_JOIN_FUSED_JOIN,
338    lattice_reduce::LATTICE_REDUCE,
339    map::MAP,
340    union::UNION,
341    multiset_delta::MULTISET_DELTA,
342    defer_signal::DEFER_SIGNAL,
343    defer_tick::DEFER_TICK,
344    defer_tick_lazy::DEFER_TICK_LAZY,
345    null::NULL,
346    partition::PARTITION,
347    persist::PERSIST,
348    resolve_futures::RESOLVE_FUTURES,
349    resolve_futures_blocking::RESOLVE_FUTURES_BLOCKING,
350    resolve_futures_blocking_ordered::RESOLVE_FUTURES_BLOCKING_ORDERED,
351    resolve_futures_ordered::RESOLVE_FUTURES_ORDERED,
352    reduce::REDUCE,
353    reduce_no_replay::REDUCE_NO_REPLAY,
354    scan::SCAN,
355    scan_async_blocking::SCAN_ASYNC_BLOCKING,
356    spin::SPIN,
357    sort::SORT,
358    sort_by_key::SORT_BY_KEY,
359    source_file::SOURCE_FILE,
360    source_interval::SOURCE_INTERVAL,
361    source_iter::SOURCE_ITER,
362    source_json::SOURCE_JSON,
363    source_stdin::SOURCE_STDIN,
364    source_stream::SOURCE_STREAM,
365    source_stream_serde::SOURCE_STREAM_SERDE,
366    state::STATE,
367    state_by::STATE_BY,
368    tee::TEE,
369    unique::UNIQUE,
370    unzip::UNZIP,
371    zip::ZIP,
372    zip_longest::ZIP_LONGEST,
373];
374
375/// Get the operator lookup table, generating it if needed.
376pub fn operator_lookup() -> &'static HashMap<&'static str, &'static OperatorConstraints> {
377    pub static OPERATOR_LOOKUP: OnceLock<HashMap<&'static str, &'static OperatorConstraints>> =
378        OnceLock::new();
379    OPERATOR_LOOKUP.get_or_init(|| OPERATORS.iter().map(|op| (op.name, op)).collect())
380}
381/// Find an operator by [`GraphNode`].
382pub fn find_node_op_constraints(node: &GraphNode) -> Option<&'static OperatorConstraints> {
383    if let GraphNode::Operator(operator) = node {
384        find_op_op_constraints(operator)
385    } else {
386        None
387    }
388}
389/// Find an operator by an AST [`Operator`].
390pub fn find_op_op_constraints(operator: &Operator) -> Option<&'static OperatorConstraints> {
391    let name = &*operator.name_string();
392    operator_lookup().get(name).copied()
393}
394
395/// Context arguments provided to [`OperatorConstraints::write_fn`].
396#[derive(Clone)]
397pub struct WriteContextArgs<'a> {
398    /// `dfir` crate name for `use #root::something`.
399    pub root: &'a TokenStream,
400    /// `context` ident, the name of the provided
401    /// [`dfir_rs::scheduled::Context`](https://hydro.run/rustdoc/dfir_rs/scheduled/context/struct.Context.html).
402    pub context: &'a Ident,
403    /// `df` ident, the name of the
404    /// [`dfir_rs::scheduled::graph::Dfir`](https://hydro.run/rustdoc/dfir_rs/scheduled/graph/struct.Dfir.html)
405    /// instance.
406    pub df_ident: &'a Ident,
407    /// Subgraph ID in which this operator is contained.
408    pub subgraph_id: GraphSubgraphId,
409    /// Node ID identifying this operator in the flat or partitioned graph meta-datastructure.
410    pub node_id: GraphNodeId,
411    /// Loop ID in which this operator is contained, or `None` if not in a loop.
412    pub loop_id: Option<GraphLoopId>,
413    /// The source span of this operator.
414    pub op_span: Span,
415    /// Tag for this operator appended to the generated identifier.
416    pub op_tag: Option<String>,
417    /// Identifier for a function to call when doing work outside the stream.
418    pub work_fn: &'a Ident,
419    /// Identifier for a function to wrap futures when doing work outside the stream.
420    pub work_fn_async: &'a Ident,
421
422    /// Ident the `Stream` or `Sink` should be assigned to.
423    pub ident: &'a Ident,
424    /// If a pull `Stream` (true) or push `Sink` (false) should be used.
425    pub is_pull: bool,
426    /// Input `Stream` operator idents (or ref idents; used for pull).
427    pub inputs: &'a [Ident],
428    /// Output `Sink` operator idents (or ref idents; used for push).
429    pub outputs: &'a [Ident],
430
431    /// Operator name.
432    pub op_name: &'static str,
433    /// Operator instance arguments object.
434    pub op_inst: &'a OperatorInstance,
435    /// Arguments provided by the user into the operator as arguments.
436    /// I.e. the `a, b, c` in `-> my_op(a, b, c) -> `.
437    ///
438    /// These arguments include singleton postprocessing codegen, with
439    /// [`std::cell::RefCell::borrow_mut`] code pre-generated.
440    pub arguments: &'a Punctuated<Expr, Token![,]>,
441
442    /// The codegen options passed to
443    /// [`DfirGraph::as_code_with_options`](super::DfirGraph::as_code_with_options).
444    ///
445    /// Operators may consult these, e.g. to omit their own type-guard helper functions when
446    /// [`AsCodeOptions::exclude_type_guards`] is set.
447    pub options: &'a AsCodeOptions,
448}
449impl WriteContextArgs<'_> {
450    /// Generate a (almost certainly) unique identifier with the given suffix.
451    ///
452    /// Includes the subgraph and node IDs in the generated identifier.
453    ///
454    /// This will always return the same identifier for a given `suffix`.
455    pub fn make_ident(&self, suffix: impl AsRef<str>) -> Ident {
456        Ident::new(
457            &format!(
458                "sg_{:?}_node_{:?}_{}",
459                self.subgraph_id.data(),
460                self.node_id.data(),
461                suffix.as_ref(),
462            ),
463            self.op_span,
464        )
465    }
466
467    /// Returns the given number of persistence arguments, defaulting to `'tick` when unspecified.
468    pub fn persistence_args<const N: usize>(
469        &self,
470        diagnostics: &mut Diagnostics,
471    ) -> [Persistence; N] {
472        let len = self.op_inst.generics.persistence_args.len();
473        if 0 != len && 1 != len && N != len {
474            diagnostics.push(Diagnostic::spanned(
475                self.op_span,
476                Level::Error,
477                format!(
478                    "The operator `{}` only accepts 0, 1, or {} persistence arguments",
479                    self.op_name, N
480                ),
481            ));
482        }
483
484        let mut out = [Persistence::Tick; N];
485        self.op_inst
486            .generics
487            .persistence_args
488            .iter()
489            .copied()
490            .cycle() // Re-use the first element for both persistences.
491            .take(N)
492            .enumerate()
493            .for_each(|(i, p)| {
494                out[i] = p;
495            });
496        out
497    }
498}
499
500/// An object-safe version of [`RangeBounds`].
501pub trait RangeTrait<T>: Send + Sync + Debug
502where
503    T: ?Sized,
504{
505    /// Start (lower) bound.
506    fn start_bound(&self) -> Bound<&T>;
507    /// End (upper) bound.
508    fn end_bound(&self) -> Bound<&T>;
509    /// Returns if `item` is contained in this range.
510    fn contains(&self, item: &T) -> bool
511    where
512        T: PartialOrd<T>;
513
514    /// Turn this range into a human-readable string.
515    fn human_string(&self) -> String
516    where
517        T: Display + PartialEq,
518    {
519        match (self.start_bound(), self.end_bound()) {
520            (Bound::Unbounded, Bound::Unbounded) => "any number of".to_owned(),
521
522            (Bound::Included(n), Bound::Included(x)) if n == x => {
523                format!("exactly {}", n)
524            }
525            (Bound::Included(n), Bound::Included(x)) => {
526                format!("at least {} and at most {}", n, x)
527            }
528            (Bound::Included(n), Bound::Excluded(x)) => {
529                format!("at least {} and less than {}", n, x)
530            }
531            (Bound::Included(n), Bound::Unbounded) => format!("at least {}", n),
532            (Bound::Excluded(n), Bound::Included(x)) => {
533                format!("more than {} and at most {}", n, x)
534            }
535            (Bound::Excluded(n), Bound::Excluded(x)) => {
536                format!("more than {} and less than {}", n, x)
537            }
538            (Bound::Excluded(n), Bound::Unbounded) => format!("more than {}", n),
539            (Bound::Unbounded, Bound::Included(x)) => format!("at most {}", x),
540            (Bound::Unbounded, Bound::Excluded(x)) => format!("less than {}", x),
541        }
542    }
543}
544
545impl<R, T> RangeTrait<T> for R
546where
547    R: RangeBounds<T> + Send + Sync + Debug,
548{
549    fn start_bound(&self) -> Bound<&T> {
550        self.start_bound()
551    }
552
553    fn end_bound(&self) -> Bound<&T> {
554        self.end_bound()
555    }
556
557    fn contains(&self, item: &T) -> bool
558    where
559        T: PartialOrd<T>,
560    {
561        self.contains(item)
562    }
563}
564
565/// Persistence lifetimes: `'none`, `'loop`, `'tick`, or `'static`.
566#[derive(Clone, Copy, PartialOrd, Ord, PartialEq, Eq, Debug, Serialize, Deserialize)]
567pub enum Persistence {
568    /// Persistence for one "tick" (or loop iteration), i.e. no persistance at all
569    Tick,
570    /// Persistence across the entire program execution.
571    Static,
572}
573impl Persistence {
574    /// Returns a lowercase string for the persistence type.
575    pub fn to_str_lowercase(self) -> &'static str {
576        match self {
577            Persistence::Tick => "tick",
578            Persistence::Static => "static",
579        }
580    }
581}
582
583/// Helper which creates a error message string literal for when the Tokio runtime is not found.
584fn make_missing_runtime_msg(op_name: &str) -> Literal {
585    Literal::string(&format!(
586        "`{}()` must be used within a Tokio runtime. For example, use `#[dfir_rs::main]` on your main method.",
587        op_name
588    ))
589}
590
591/// Operator categories, for docs.
592#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, DocumentedVariants)]
593pub enum OperatorCategory {
594    /// Maps: Simple one-in-one-out operators.
595    Map,
596    /// Filters: One-in zero-or-one-out operators.
597    Filter,
598    /// Flattens: One-in multiple-out operators.
599    Flatten,
600    /// Folds: Operators which accumulate elements together.
601    Fold,
602    /// Keyed Folds: Operators which accumulate elements together by key.
603    KeyedFold,
604    /// Lattice Folds: Folds based on lattice-merge.
605    LatticeFold,
606    /// Persistent Operators: Persistent (stateful) operators.
607    Persistence,
608    /// Multi-Input Operators: Operators with multiple inputs.
609    MultiIn,
610    /// Multi-Output Operators: Operators with multiple outputs.
611    MultiOut,
612    /// Sources: Operators which produce output elements (and consume no inputs).
613    Source,
614    /// Sinks: Operators which consume input elements (and produce no outputs).
615    Sink,
616    /// Control Flow Operators: Operators which affect control flow/scheduling.
617    Control,
618    /// Compiler Fusion Operators: Operators which are necessary to implement certain optimizations and rewrite rules.
619    CompilerFusionOperator,
620    /// Windowing Operators: Operators for windowing `loop` inputs.
621    Windowing,
622    /// Un-Windowing Operators: Operators for collecting `loop` outputs.
623    Unwindowing,
624}
625impl OperatorCategory {
626    /// Human-readible heading name, for docs.
627    pub fn name(self) -> &'static str {
628        self.get_variant_docs().split_once(':').unwrap().0
629    }
630    /// Human description, for docs.
631    pub fn description(self) -> &'static str {
632        self.get_variant_docs().split_once(':').unwrap().1
633    }
634}
635
636/// Operator type for Flo semantics.
637#[derive(Clone, Copy, PartialOrd, Ord, PartialEq, Eq, Debug)]
638pub enum FloType {
639    /// A source operator, which must be at the top level.
640    Source,
641    /// A windowing operator, for moving data into a loop context.
642    Windowing,
643    /// A lazy windowing operator — moves data into a loop context but does not trigger the loop.
644    /// Data is dropped if the loop does not fire that tick.
645    WindowingLazy,
646    /// An eager windowing operator — moves data into a loop context and always triggers the loop,
647    /// even when the windowed input is empty. Only valid at the entry of a root-level loop (it is
648    /// disallowed in nested loops, where it would prevent the fixpoint iteration from terminating).
649    WindowingEager,
650    /// An un-windowing operator, for moving data out of a loop context.
651    Unwindowing,
652}
dfir_lang/graph/ops/mod.rs

dfir_lang/graph/ops/
mod.rs
