datum/graph/
builder.rs

1//! Graph construction: the builder, the `GraphDsl` entry points, and the
2//! immutable `GraphBlueprint` they produce.
3//!
4//! Building a graph is side-effect-free (blueprint vs. materialization): `add`
5//! registers a stage's ports, `connect`/`wire` records an edge after
6//! `validate_connection` checks port kind, element `TypeId`, and single-use, and
7//! `finish` verifies the returned [`Shape`] matches the open ports. Execution
8//! starts only when a `run_*` method (defined in the `executor` module) is
9//! called on the resulting [`GraphBlueprint`].
10
11use super::*;
12use crate::{Attribute, Attributes};
13
14type PartialGraphBuilder<S> = dyn Fn(&mut GraphBuilder) -> StreamResult<S> + Send + Sync;
15
16#[derive(Clone, Debug)]
17struct PortRecord {
18    kind: PortKind,
19    type_id: TypeId,
20    type_name: &'static str,
21    name: Arc<str>,
22}
23
24#[derive(Clone, Debug)]
25pub(super) struct Edge {
26    pub(super) outlet: PortId,
27    pub(super) inlet: PortId,
28}
29
30#[derive(Clone)]
31pub(super) struct StageRecord {
32    pub(super) spec: StageSpec,
33    pub(super) logic_factory: Option<Arc<dyn Fn() -> GraphStageLogic + Send + Sync>>,
34}
35
36impl std::fmt::Debug for StageRecord {
37    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
38        f.debug_struct("StageRecord")
39            .field("spec", &self.spec)
40            .field("has_logic", &self.logic_factory.is_some())
41            .finish()
42    }
43}
44
45/// Mutable scratch state for assembling a dynamic graph inside a `GraphDsl`
46/// closure.
47///
48/// Tracks registered stages, their ports, the edges wired between them, and any
49/// deferred wiring errors (raised by `wire`, which collects rather than
50/// short-circuits). `finish` consumes it into a validated [`GraphBlueprint`].
51/// Use this builder for cycles, runtime-data-dependent topology, erased
52/// interop, `connect_any`, and method-based `wire` construction. For static
53/// typed graphs where double-wiring should be rejected by Rust moves, use
54/// [`TypedGraphBuilder`](super::TypedGraphBuilder).
55#[derive(Debug, Default)]
56pub struct GraphBuilder {
57    allocator: PortAllocator,
58    ports: HashMap<PortId, PortRecord>,
59    stages: Vec<StageRecord>,
60    edges: Vec<Edge>,
61    errors: Vec<StreamError>,
62}
63
64impl GraphBuilder {
65    #[must_use]
66    pub fn add<G: GraphStage>(&mut self, stage: G) -> G::Shape {
67        self.add_with_attributes(stage, Attributes::default())
68    }
69
70    #[must_use]
71    pub fn add_with_attributes<G: GraphStage>(
72        &mut self,
73        stage: G,
74        attributes: Attributes,
75    ) -> G::Shape {
76        let shape = stage.allocate_shape(&mut self.allocator);
77        let inlets = shape.inlets();
78        let outlets = shape.outlets();
79        self.ports.reserve(inlets.len() + outlets.len());
80
81        for inlet in &inlets {
82            self.register_inlet(inlet);
83        }
84        for outlet in &outlets {
85            self.register_outlet(outlet);
86        }
87        let spec = stage
88            .stage_spec_with_ports(&shape, inlets, outlets)
89            .add_attributes(attributes);
90        let logic_factory = if matches!(spec.kind, StageKind::Opaque) {
91            let shape_clone = shape.clone();
92            Some(Arc::new(move || stage.create_logic(&shape_clone))
93                as Arc<dyn Fn() -> GraphStageLogic + Send + Sync>)
94        } else {
95            None
96        };
97        self.stages.push(StageRecord {
98            spec,
99            logic_factory,
100        });
101        shape
102    }
103
104    #[must_use]
105    pub fn add_named<G: GraphStage>(&mut self, stage: G, name: impl Into<String>) -> G::Shape {
106        self.add_with_attributes(stage, Attributes::named(name))
107    }
108
109    pub fn connect<T: 'static>(&mut self, outlet: Outlet<T>, inlet: Inlet<T>) -> StreamResult<()> {
110        self.connect_any(outlet.erase(), inlet.erase())
111    }
112
113    /// Connect type-erased ports for dynamic graph construction and interop.
114    ///
115    /// This remains an explicit compatibility surface for cases where the
116    /// element type is only known at runtime. Prefer [`connect`](Self::connect)
117    /// or [`TypedGraphBuilder`](super::TypedGraphBuilder) for static Rust
118    /// graphs.
119    pub fn connect_any(&mut self, outlet: AnyOutlet, inlet: AnyInlet) -> StreamResult<()> {
120        match self.connect_any_unrecorded(outlet, inlet) {
121            Ok(()) => Ok(()),
122            Err(error) => self.record_error(error),
123        }
124    }
125
126    pub fn import<S: Shape>(&mut self, graph: &PartialGraph<S>) -> StreamResult<S> {
127        graph.build(self)
128    }
129
130    /// Apply a dynamic/method-based wiring spec, deferring errors to `finish`.
131    ///
132    /// `wire` is useful for graph DSL ergonomics and interop-style topology
133    /// construction. Static Rust graphs can use [`TypedGraphBuilder`] when
134    /// compile-time double-wire rejection is desired.
135    pub fn wire<W: WireSpec>(&mut self, spec: W) -> &mut Self {
136        if let Err(error) = spec.apply(self) {
137            let _ = self.record_error(error);
138        }
139        self
140    }
141
142    /// Apply a dynamic/method-based wiring spec, returning wiring errors
143    /// immediately.
144    ///
145    /// This is the fallible counterpart to [`wire`](Self::wire) and remains a
146    /// runtime-validated dynamic/interop surface.
147    pub fn try_wire<W: WireSpec>(&mut self, spec: W) -> StreamResult<&mut Self> {
148        match spec.apply(self) {
149            Ok(()) => Ok(self),
150            Err(error) => {
151                self.errors.push(error.clone());
152                Err(error)
153            }
154        }
155    }
156
157    pub(super) fn connect_any_unrecorded(
158        &mut self,
159        outlet: AnyOutlet,
160        inlet: AnyInlet,
161    ) -> StreamResult<()> {
162        self.validate_connection(&outlet, &inlet)?;
163        self.edges.push(Edge {
164            outlet: outlet.id(),
165            inlet: inlet.id(),
166        });
167        Ok(())
168    }
169
170    pub(super) fn is_outlet_connected(&self, outlet: &AnyOutlet) -> bool {
171        self.edges.iter().any(|edge| edge.outlet == outlet.id())
172    }
173
174    pub(super) fn is_inlet_connected(&self, inlet: &AnyInlet) -> bool {
175        self.edges.iter().any(|edge| edge.inlet == inlet.id())
176    }
177
178    fn record_error(&mut self, error: StreamError) -> StreamResult<()> {
179        self.errors.push(error.clone());
180        Err(error)
181    }
182
183    fn register_inlet(&mut self, inlet: &AnyInlet) {
184        self.ports.insert(
185            inlet.id(),
186            PortRecord {
187                kind: PortKind::Inlet,
188                type_id: inlet.type_id(),
189                type_name: inlet.type_name(),
190                name: Arc::clone(&inlet.name),
191            },
192        );
193    }
194
195    fn register_outlet(&mut self, outlet: &AnyOutlet) {
196        self.ports.insert(
197            outlet.id(),
198            PortRecord {
199                kind: PortKind::Outlet,
200                type_id: outlet.type_id(),
201                type_name: outlet.type_name(),
202                name: Arc::clone(&outlet.name),
203            },
204        );
205    }
206
207    fn validate_connection(&self, outlet: &AnyOutlet, inlet: &AnyInlet) -> StreamResult<()> {
208        let outlet_record = self.ports.get(&outlet.id()).ok_or_else(|| {
209            StreamError::GraphValidation(format!("unknown outlet {}", outlet.name()))
210        })?;
211        let inlet_record = self.ports.get(&inlet.id()).ok_or_else(|| {
212            StreamError::GraphValidation(format!("unknown inlet {}", inlet.name()))
213        })?;
214
215        if outlet_record.kind != PortKind::Outlet {
216            return Err(StreamError::GraphValidation(format!(
217                "{} is not an outlet",
218                outlet_record.name
219            )));
220        }
221        if inlet_record.kind != PortKind::Inlet {
222            return Err(StreamError::GraphValidation(format!(
223                "{} is not an inlet",
224                inlet_record.name
225            )));
226        }
227        if outlet_record.type_id != inlet_record.type_id {
228            return Err(StreamError::GraphValidation(format!(
229                "cannot connect outlet {} ({}) to inlet {} ({})",
230                outlet_record.name,
231                outlet_record.type_name,
232                inlet_record.name,
233                inlet_record.type_name
234            )));
235        }
236        if self.edges.iter().any(|edge| edge.outlet == outlet.id()) {
237            return Err(StreamError::GraphValidation(format!(
238                "outlet {} is already connected",
239                outlet_record.name
240            )));
241        }
242        if self.edges.iter().any(|edge| edge.inlet == inlet.id()) {
243            return Err(StreamError::GraphValidation(format!(
244                "inlet {} is already connected",
245                inlet_record.name
246            )));
247        }
248
249        Ok(())
250    }
251
252    pub(super) fn finish<S: Shape>(self, shape: S) -> StreamResult<GraphBlueprint<S>> {
253        let mut errors = self.errors;
254        let connected_inlets: HashSet<PortId> = self.edges.iter().map(|edge| edge.inlet).collect();
255        let connected_outlets: HashSet<PortId> =
256            self.edges.iter().map(|edge| edge.outlet).collect();
257
258        let open_inlets: HashSet<PortId> = self
259            .ports
260            .iter()
261            .filter_map(|(id, port)| {
262                (port.kind == PortKind::Inlet && !connected_inlets.contains(id)).then_some(*id)
263            })
264            .collect();
265        let open_outlets: HashSet<PortId> = self
266            .ports
267            .iter()
268            .filter_map(|(id, port)| {
269                (port.kind == PortKind::Outlet && !connected_outlets.contains(id)).then_some(*id)
270            })
271            .collect();
272
273        let result_inlets: HashSet<PortId> = shape.inlets().iter().map(AnyInlet::id).collect();
274        let result_outlets: HashSet<PortId> = shape.outlets().iter().map(AnyOutlet::id).collect();
275
276        for inlet in shape.inlets() {
277            match self.ports.get(&inlet.id()) {
278                Some(port)
279                    if port.kind == PortKind::Inlet
280                        && port.type_id == inlet.type_id()
281                        && port.name.as_ref() == inlet.name() => {}
282                Some(port) if port.kind == PortKind::Inlet => {
283                    errors.push(StreamError::GraphValidation(format!(
284                        "result shape inlet {} does not match registered inlet {} ({})",
285                        inlet.name(),
286                        port.name,
287                        port.type_name
288                    )));
289                }
290                Some(port) => errors.push(StreamError::GraphValidation(format!(
291                    "result shape references non-inlet port {}",
292                    port.name
293                ))),
294                None => errors.push(StreamError::GraphValidation(format!(
295                    "result shape references unknown inlet {}",
296                    inlet.name()
297                ))),
298            }
299        }
300        for outlet in shape.outlets() {
301            match self.ports.get(&outlet.id()) {
302                Some(port)
303                    if port.kind == PortKind::Outlet
304                        && port.type_id == outlet.type_id()
305                        && port.name.as_ref() == outlet.name() => {}
306                Some(port) if port.kind == PortKind::Outlet => {
307                    errors.push(StreamError::GraphValidation(format!(
308                        "result shape outlet {} does not match registered outlet {} ({})",
309                        outlet.name(),
310                        port.name,
311                        port.type_name
312                    )));
313                }
314                Some(port) => errors.push(StreamError::GraphValidation(format!(
315                    "result shape references non-outlet port {}",
316                    port.name
317                ))),
318                None => errors.push(StreamError::GraphValidation(format!(
319                    "result shape references unknown outlet {}",
320                    outlet.name()
321                ))),
322            }
323        }
324
325        if open_inlets != result_inlets {
326            errors.push(StreamError::GraphValidation(format!(
327                "result shape inlets do not match open inlets: open={:?}, result={:?}",
328                describe_ports(&self.ports, &open_inlets),
329                describe_ports(&self.ports, &result_inlets)
330            )));
331        }
332        if open_outlets != result_outlets {
333            errors.push(StreamError::GraphValidation(format!(
334                "result shape outlets do not match open outlets: open={:?}, result={:?}",
335                describe_ports(&self.ports, &open_outlets),
336                describe_ports(&self.ports, &result_outlets)
337            )));
338        }
339
340        if !errors.is_empty() {
341            return Err(StreamError::GraphValidation(
342                errors
343                    .into_iter()
344                    .map(|error| error.to_string())
345                    .collect::<Vec<_>>()
346                    .join("; "),
347            ));
348        }
349
350        let segments = compute_segments(&self.stages);
351        Ok(GraphBlueprint {
352            shape,
353            stages: self.stages,
354            edges: self.edges,
355            segments,
356            attributes: Attributes::default(),
357        })
358    }
359}
360
361fn describe_ports(ports: &HashMap<PortId, PortRecord>, ids: &HashSet<PortId>) -> Vec<String> {
362    let mut names = ids
363        .iter()
364        .map(|id| {
365            ports
366                .get(id)
367                .map(|port| port.name.as_ref().to_owned())
368                .unwrap_or_else(|| format!("unknown#{}", id.as_usize()))
369        })
370        .collect::<Vec<_>>();
371    names.sort();
372    names
373}
374
375fn compute_segments(stages: &[StageRecord]) -> Vec<FusedSegment> {
376    let mut segments = Vec::with_capacity(1);
377    let mut current = Vec::with_capacity(stages.len());
378
379    for (index, stage) in stages.iter().enumerate() {
380        if stage.spec.async_boundary && !current.is_empty() {
381            segments.push(FusedSegment {
382                stage_indices: std::mem::take(&mut current),
383            });
384        }
385        current.push(index);
386        if stage.spec.async_boundary {
387            segments.push(FusedSegment {
388                stage_indices: std::mem::take(&mut current),
389            });
390        }
391    }
392
393    if !current.is_empty() {
394        segments.push(FusedSegment {
395            stage_indices: current,
396        });
397    }
398
399    segments
400}
401
402/// Entry points for building graphs. `create` takes a closure returning the
403/// shape directly, `try_create` a closure returning `StreamResult<Shape>` (so
404/// `?` works on `connect`/`try_wire`), and `partial` builds a reusable
405/// [`PartialGraph`] fragment.
406pub struct GraphDsl;
407
408impl GraphDsl {
409    pub fn create<S, F>(build: F) -> StreamResult<GraphBlueprint<S>>
410    where
411        S: Shape,
412        F: FnOnce(&mut GraphBuilder) -> S,
413    {
414        let mut builder = GraphBuilder::default();
415        let shape = build(&mut builder);
416        builder.finish(shape)
417    }
418
419    pub fn try_create<S, F>(build: F) -> StreamResult<GraphBlueprint<S>>
420    where
421        S: Shape,
422        F: FnOnce(&mut GraphBuilder) -> StreamResult<S>,
423    {
424        let mut builder = GraphBuilder::default();
425        let shape = build(&mut builder)?;
426        builder.finish(shape)
427    }
428
429    pub fn partial<S, F>(build: F) -> PartialGraph<S>
430    where
431        S: Shape,
432        F: Fn(&mut GraphBuilder) -> StreamResult<S> + Send + Sync + 'static,
433    {
434        PartialGraph {
435            build: Arc::new(build),
436            attributes: Attributes::default(),
437        }
438    }
439}
440
441/// Anything exposing a graph [`Shape`] — implemented by [`GraphBlueprint`].
442pub trait Graph {
443    type Shape: Shape;
444
445    fn shape(&self) -> Self::Shape;
446}
447
448/// A maximal run of stages with no async boundary between them, executed
449/// together by the fused executor. Boundaries (`AsyncBoundary` stages) split a
450/// graph into consecutive segments.
451#[derive(Clone, Debug)]
452pub struct FusedSegment {
453    stage_indices: Vec<usize>,
454}
455
456impl FusedSegment {
457    #[must_use]
458    pub fn stage_indices(&self) -> &[usize] {
459        &self.stage_indices
460    }
461}
462
463/// An immutable, validated graph ready to run. Produced by `GraphDsl::create`/
464/// `try_create`. Carries the external [`Shape`], the stages, the wired edges,
465/// the precomputed fused segments, and graph-level [`Attributes`]. The `run_*`
466/// methods are defined in the `executor` module; running one never mutates the
467/// blueprint, so it can be reused and run concurrently.
468pub struct GraphBlueprint<S: Shape> {
469    pub(super) shape: S,
470    pub(super) stages: Vec<StageRecord>,
471    pub(super) edges: Vec<Edge>,
472    pub(super) segments: Vec<FusedSegment>,
473    pub(super) attributes: Attributes,
474}
475
476impl<S: Shape + Clone> Clone for GraphBlueprint<S> {
477    fn clone(&self) -> Self {
478        Self {
479            shape: self.shape.clone(),
480            stages: self.stages.clone(),
481            edges: self.edges.clone(),
482            segments: self.segments.clone(),
483            attributes: self.attributes.clone(),
484        }
485    }
486}
487
488impl<S: Shape + fmt::Debug> fmt::Debug for GraphBlueprint<S> {
489    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
490        f.debug_struct("GraphBlueprint")
491            .field("shape", &self.shape)
492            .field("stages", &self.stages)
493            .field("edges", &self.edges)
494            .field("segments", &self.segments)
495            .field("attributes", &self.attributes)
496            .finish()
497    }
498}
499
500impl<S: Shape> GraphBlueprint<S> {
501    #[must_use]
502    pub fn shape(&self) -> S {
503        self.shape.clone()
504    }
505
506    #[must_use]
507    pub fn stage_count(&self) -> usize {
508        self.stages.len()
509    }
510
511    #[must_use]
512    pub fn edge_count(&self) -> usize {
513        self.edges.len()
514    }
515
516    #[must_use]
517    pub fn segments(&self) -> &[FusedSegment] {
518        &self.segments
519    }
520
521    #[must_use]
522    pub fn attributes(&self) -> &Attributes {
523        &self.attributes
524    }
525
526    /// Effective post-merge attributes for every stage in execution order.
527    ///
528    /// Graph-level attributes are inherited by each stage and stage-level attributes override them
529    /// per key, matching Akka's closest-wins mental model.
530    #[must_use]
531    pub fn effective_stage_attributes(&self) -> Vec<FusedNodeAttributes> {
532        self.stage_attribute_reports()
533    }
534
535    #[must_use]
536    pub fn with_attributes(mut self, attributes: Attributes) -> Self {
537        self.attributes = attributes;
538        self
539    }
540
541    #[must_use]
542    pub fn add_attributes(mut self, attributes: Attributes) -> Self {
543        self.attributes = self.attributes.and(attributes);
544        self
545    }
546
547    #[must_use]
548    pub fn named(self, name: impl Into<String>) -> Self {
549        self.add_attributes(Attributes::named(name))
550    }
551
552    pub(super) fn effective_attributes_for_stage(&self, stage: &StageRecord) -> Attributes {
553        self.attributes.clone().and(stage.spec.attributes().clone())
554    }
555
556    pub(super) fn stage_attribute_reports(&self) -> Vec<FusedNodeAttributes> {
557        self.stages
558            .iter()
559            .enumerate()
560            .map(|(index, stage)| {
561                let attributes = self.effective_attributes_for_stage(stage);
562                FusedNodeAttributes {
563                    stage_index: index,
564                    stage_name: stage.spec.name().to_owned(),
565                    effective_name: attributes
566                        .name()
567                        .map(str::to_owned)
568                        .unwrap_or_else(|| stage.spec.name().to_owned()),
569                    attributes,
570                }
571            })
572            .collect()
573    }
574
575    pub(super) fn plan_report(
576        &self,
577        selected_tier: FusedExecutorTier,
578        tier_changes: Vec<FusedTierChange>,
579    ) -> FusedPlanReport {
580        FusedPlanReport {
581            selected_tier,
582            stage_attributes: self.stage_attribute_reports(),
583            tier_changes,
584        }
585    }
586}
587
588impl<S: Shape> Graph for GraphBlueprint<S> {
589    type Shape = S;
590
591    fn shape(&self) -> Self::Shape {
592        self.shape()
593    }
594}
595
596/// A reusable graph fragment: a builder closure plus its [`Shape`], importable
597/// into multiple parent graphs via [`GraphBuilder::import`]. Still a blueprint —
598/// the closure runs (allocating fresh ports) each time it is imported.
599/// Aliased as [`ImportedGraph`].
600#[derive(Clone)]
601pub struct PartialGraph<S: Shape> {
602    build: Arc<PartialGraphBuilder<S>>,
603    attributes: Attributes,
604}
605
606impl<S: Shape> PartialGraph<S> {
607    pub fn build(&self, builder: &mut GraphBuilder) -> StreamResult<S> {
608        (self.build)(builder)
609    }
610
611    #[must_use]
612    pub fn attributes(&self) -> &Attributes {
613        &self.attributes
614    }
615
616    #[must_use]
617    pub fn with_attributes(mut self, attributes: Attributes) -> Self {
618        self.attributes = attributes;
619        self
620    }
621
622    #[must_use]
623    pub fn add_attributes(mut self, attributes: Attributes) -> Self {
624        self.attributes = self.attributes.and(attributes);
625        self
626    }
627
628    #[must_use]
629    pub fn named(self, name: impl Into<String>) -> Self {
630        self.add_attributes(Attributes::named(name))
631    }
632}
633
634impl<S: Shape> std::fmt::Debug for PartialGraph<S> {
635    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
636        f.debug_struct("PartialGraph")
637            .field("attributes", &self.attributes)
638            .finish_non_exhaustive()
639    }
640}
641
642pub type ImportedGraph<S> = PartialGraph<S>;
643
644/// Execution bound for the fused executor. `event_limit` caps the number of
645/// push/pull events a single run may take, so an unproductive cycle surfaces
646/// [`StreamError::EventLimitExceeded`] instead of hanging. Defaults to 100M.
647#[derive(Clone, Copy, Debug, PartialEq, Eq)]
648pub struct FusedExecutionConfig {
649    pub event_limit: usize,
650}
651
652impl Default for FusedExecutionConfig {
653    fn default() -> Self {
654        Self {
655            event_limit: 100_000_000,
656        }
657    }
658}
659
660/// Execution settings for the current graph async-boundary benchmark path.
661///
662/// This path validates a typed-linear graph and uses Ractor-backed async
663/// regions with bounded handoff queues to measure real boundary crossing cost.
664#[derive(Clone, Copy, Debug, PartialEq, Eq)]
665pub struct AsyncBoundaryExecutionConfig {
666    pub fused: FusedExecutionConfig,
667    pub buffer_size: usize,
668}
669
670impl Default for AsyncBoundaryExecutionConfig {
671    fn default() -> Self {
672        Self {
673            fused: FusedExecutionConfig::default(),
674            buffer_size: 16,
675        }
676    }
677}
678
679/// Result of a collecting run: the `output` vector plus instrumentation
680/// (`events` processed, `async_boundary_crossings`) and the selected graph execution plan.
681/// Returned by the `*_with_input_report` methods.
682#[derive(Clone, Debug, PartialEq, Eq)]
683pub struct FusedExecutionReport<T> {
684    pub output: Vec<T>,
685    pub events: usize,
686    pub async_boundary_crossings: usize,
687    pub plan: FusedPlanReport,
688    pub node_metrics: Vec<FusedNodeMetrics>,
689}
690
691/// Result of a terminal (count/fold) run: the reduced `result` plus the same
692/// instrumentation and plan data as [`FusedExecutionReport`]. Returned by the
693/// `*_count_*_report` / `*_fold_*_report` methods.
694#[derive(Clone, Debug, PartialEq, Eq)]
695pub struct FusedTerminalReport<T> {
696    pub result: T,
697    pub events: usize,
698    pub async_boundary_crossings: usize,
699    pub plan: FusedPlanReport,
700    pub node_metrics: Vec<FusedNodeMetrics>,
701}
702
703impl<T> FusedExecutionReport<T> {
704    pub(super) fn new(
705        output: Vec<T>,
706        events: usize,
707        async_boundary_crossings: usize,
708        plan: FusedPlanReport,
709    ) -> Self {
710        Self {
711            output,
712            events,
713            async_boundary_crossings,
714            plan,
715            node_metrics: Vec::new(),
716        }
717    }
718}
719
720impl<T> FusedTerminalReport<T> {
721    pub(super) fn new(
722        result: T,
723        events: usize,
724        async_boundary_crossings: usize,
725        plan: FusedPlanReport,
726    ) -> Self {
727        Self {
728            result,
729            events,
730            async_boundary_crossings,
731            plan,
732            node_metrics: Vec::new(),
733        }
734    }
735}
736
737impl<T> FusedExecutionReport<T> {
738    pub(super) fn with_node_metrics(mut self, node_metrics: Vec<FusedNodeMetrics>) -> Self {
739        self.node_metrics = node_metrics;
740        self
741    }
742
743    /// Return an immutable point-in-time copy of this run's node metrics.
744    #[must_use]
745    pub fn metrics_snapshot(&self) -> Vec<FusedNodeMetrics> {
746        self.node_metrics.clone()
747    }
748}
749
750impl<T> FusedTerminalReport<T> {
751    pub(super) fn with_node_metrics(mut self, node_metrics: Vec<FusedNodeMetrics>) -> Self {
752        self.node_metrics = node_metrics;
753        self
754    }
755
756    /// Return an immutable point-in-time copy of this run's node metrics.
757    #[must_use]
758    pub fn metrics_snapshot(&self) -> Vec<FusedNodeMetrics> {
759        self.node_metrics.clone()
760    }
761}
762
763/// Executor tier selected for a fused graph run.
764#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
765pub enum FusedExecutorTier {
766    TypedLinear,
767    TypedAcyclicJunction,
768    TypedMergeSequence,
769    TypedMergeLatest,
770    TypedCyclicFeedback,
771    AsyncBoundary,
772    TypedFanIn,
773    TypedConcat,
774    TypedInterleave,
775    TypedMergePreferred,
776    Erased,
777}
778
779/// Plan metadata attached to fused execution reports.
780#[derive(Clone, Debug, PartialEq, Eq)]
781pub struct FusedPlanReport {
782    pub selected_tier: FusedExecutorTier,
783    pub stage_attributes: Vec<FusedNodeAttributes>,
784    pub tier_changes: Vec<FusedTierChange>,
785}
786
787impl FusedPlanReport {
788    pub(super) fn empty(selected_tier: FusedExecutorTier) -> Self {
789        Self {
790            selected_tier,
791            stage_attributes: Vec::new(),
792            tier_changes: Vec::new(),
793        }
794    }
795}
796
797/// Effective attributes for one graph stage.
798#[derive(Clone, Debug, PartialEq, Eq)]
799pub struct FusedNodeAttributes {
800    pub stage_index: usize,
801    pub stage_name: String,
802    pub effective_name: String,
803    pub attributes: Attributes,
804}
805
806/// Why a run selected a different tier than Auto would normally choose, or why Auto fell through
807/// from typed planning to erased execution.
808#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
809pub enum FusedTierChangeReason {
810    FusionErasedOnlyRequested,
811    FusionTypedOnlyConstrained,
812    AutoTypedUnsupportedFallback,
813}
814
815/// A tier selection event visible in an execution report.
816#[derive(Clone, Debug, PartialEq, Eq)]
817pub struct FusedTierChange {
818    pub requested_attribute: Option<Attribute>,
819    pub reason: FusedTierChangeReason,
820    pub selected_tier: FusedExecutorTier,
821    pub affected_nodes: Vec<String>,
822}
datum/graph/builder.rs

datum/graph/
builder.rs
