spate_core/ops/
builder.rs

1//! The fluent, type-safe chain builder.
2//!
3//! Stages are recorded as lightweight *parts* and assembled into the
4//! statically composed collector stack when the chain is built. The same
5//! parts can assemble any number of identical chains (one per pipeline
6//! thread) through [`ChainFactory`].
7
8use super::chain::{
9    FatalSlot, Filter, FlatMap, Inspect, Map, OpMeter, OpMeterSlot, StageLifecycle, TryMap,
10    TypedChain,
11};
12use super::handoff::{ChunkConfig, SinkHandoff};
13use super::split::{ErasedBranch, Sink, SinkCtx, SplitEmitter, SplitTerminal, new_branch};
14use super::{Collector, Emitter, RunnableChain};
15use crate::backpressure::InflightBudget;
16use crate::deser::{Deserializer, Owned, RecFamily};
17use crate::error::ErrorPolicy;
18use crate::metrics::{ComponentLabels, DeserMetrics, OperatorMetrics};
19use crate::sink::{RecordRouter, RowEncoder, ShardQueues};
20use std::marker::PhantomData;
21use std::sync::Arc;
22
23/// A record-to-record transform between families. Implemented for every
24/// `FnMut(In) -> Out`; expressed as an independent two-parameter trait so
25/// higher-ranked builder bounds stay legal for borrowing families (see
26/// [the module docs](crate::ops) on E0582). `fn` items satisfy it at
27/// every lifetime.
28///
29/// Name the bound to write a helper generic over one stage.
30///
31/// ```
32/// use spate_core::deser::RecFamily;
33/// use spate_core::ops::MapFn;
34///
35/// struct LogEvent<'buf> {
36///     line: &'buf str,
37/// }
38/// struct LogF;
39/// impl RecFamily for LogF {
40///     type Rec<'buf> = LogEvent<'buf>;
41/// }
42///
43/// fn first_word<'a>(ev: LogEvent<'a>) -> LogEvent<'a> {
44///     LogEvent {
45///         line: ev.line.split(' ').next().unwrap_or(""),
46///     }
47/// }
48///
49/// fn stage<F, G>(g: G) -> G
50/// where
51///     F: RecFamily,
52///     G: for<'buf> MapFn<F::Rec<'buf>, F::Rec<'buf>>,
53/// {
54///     g
55/// }
56///
57/// let _ = stage::<LogF, _>(first_word);
58/// ```
59pub trait MapFn<In, Out>: FnMut(In) -> Out {}
60impl<G, In, Out> MapFn<In, Out> for G where G: FnMut(In) -> Out {}
61
62/// Fallible variant of [`MapFn`], with the error type as a third parameter.
63///
64/// ```
65/// use spate_core::deser::RecFamily;
66/// use spate_core::ops::TryMapFn;
67/// use std::num::ParseIntError;
68///
69/// struct LogEvent<'buf> {
70///     line: &'buf str,
71/// }
72/// struct LogF;
73/// impl RecFamily for LogF {
74///     type Rec<'buf> = LogEvent<'buf>;
75/// }
76///
77/// fn numeric_only<'a>(ev: LogEvent<'a>) -> Result<LogEvent<'a>, ParseIntError> {
78///     ev.line.parse::<u32>()?;
79///     Ok(ev)
80/// }
81///
82/// fn try_stage<F, G, E>(g: G) -> G
83/// where
84///     F: RecFamily,
85///     G: for<'buf> TryMapFn<F::Rec<'buf>, F::Rec<'buf>, E>,
86///     E: std::fmt::Display,
87/// {
88///     g
89/// }
90///
91/// let _ = try_stage::<LogF, _, ParseIntError>(numeric_only);
92/// ```
93pub trait TryMapFn<In, Out, Err>: FnMut(In) -> Result<Out, Err> {}
94impl<G, In, Out, Err> TryMapFn<In, Out, Err> for G where G: FnMut(In) -> Result<Out, Err> {}
95
96/// Assembles recorded parts into the concrete collector stack, given the
97/// terminal stage. Takes `&self` so one set of parts can assemble many
98/// identical chains. Stage closures must therefore be `Clone` (plain closures
99/// and closures over `Clone`/`Arc` state are).
100pub trait Assemble<Term> {
101    /// The assembled collector stack.
102    type Out;
103    /// Build the stack around `term`.
104    fn assemble(&self, term: Term) -> Self::Out;
105}
106
107/// The empty stage list.
108#[derive(Clone, Copy, Debug, Default)]
109pub struct Root;
110
111impl<T> Assemble<T> for Root {
112    type Out = T;
113    fn assemble(&self, term: T) -> T {
114        term
115    }
116}
117
118/// Recorded `map`/`map_rec` stage.
119#[derive(Clone, Debug)]
120pub struct MapPart<Prev, G> {
121    prev: Prev,
122    f: G,
123    meter: OpMeterSlot,
124}
125
126impl<Prev, G: Clone, Term> Assemble<Term> for MapPart<Prev, G>
127where
128    Prev: Assemble<Map<G, Term>>,
129{
130    type Out = Prev::Out;
131    fn assemble(&self, term: Term) -> Self::Out {
132        self.prev.assemble(Map {
133            f: self.f.clone(),
134            next: term,
135            meter: self.meter.clone(),
136        })
137    }
138}
139
140/// Recorded `filter` stage.
141#[derive(Clone, Debug)]
142pub struct FilterPart<Prev, P> {
143    prev: Prev,
144    p: P,
145    meter: OpMeterSlot,
146}
147
148impl<Prev, P: Clone, Term> Assemble<Term> for FilterPart<Prev, P>
149where
150    Prev: Assemble<Filter<P, Term>>,
151{
152    type Out = Prev::Out;
153    fn assemble(&self, term: Term) -> Self::Out {
154        self.prev.assemble(Filter {
155            p: self.p.clone(),
156            next: term,
157            meter: self.meter.clone(),
158        })
159    }
160}
161
162/// Recorded `inspect` stage.
163#[derive(Clone, Debug)]
164pub struct InspectPart<Prev, G> {
165    prev: Prev,
166    f: G,
167}
168
169impl<Prev, G: Clone, Term> Assemble<Term> for InspectPart<Prev, G>
170where
171    Prev: Assemble<Inspect<G, Term>>,
172{
173    type Out = Prev::Out;
174    fn assemble(&self, term: Term) -> Self::Out {
175        self.prev.assemble(Inspect {
176            f: self.f.clone(),
177            next: term,
178        })
179    }
180}
181
182/// Recorded `try_map`/`try_map_rec` stage.
183#[derive(Clone, Debug)]
184pub struct TryMapPart<Prev, G> {
185    prev: Prev,
186    f: G,
187    policy: ErrorPolicy,
188    component: Arc<str>,
189    meter: OpMeterSlot,
190}
191
192impl<Prev, G: Clone, Term> Assemble<Term> for TryMapPart<Prev, G>
193where
194    Prev: Assemble<TryMap<G, Term>>,
195{
196    type Out = Prev::Out;
197    fn assemble(&self, term: Term) -> Self::Out {
198        self.prev.assemble(TryMap {
199            f: self.f.clone(),
200            next: term,
201            policy: self.policy,
202            component: Arc::clone(&self.component),
203            meter: self.meter.clone(),
204            fatal: FatalSlot(None),
205        })
206    }
207}
208
209/// Recorded `flat_map` stage.
210#[derive(Clone, Debug)]
211pub struct FlatMapPart<OutF: RecFamily, Prev, G> {
212    prev: Prev,
213    g: G,
214    meter: OpMeterSlot,
215    _out: PhantomData<fn() -> OutF>,
216}
217
218impl<OutF: RecFamily, Prev, G: Clone, Term> Assemble<Term> for FlatMapPart<OutF, Prev, G>
219where
220    Prev: Assemble<FlatMap<OutF, G, Term>>,
221{
222    type Out = Prev::Out;
223    fn assemble(&self, term: Term) -> Self::Out {
224        self.prev.assemble(FlatMap {
225            g: self.g.clone(),
226            next: term,
227            meter: self.meter.clone(),
228            _out: PhantomData,
229        })
230    }
231}
232
233#[derive(Clone, Debug)]
234struct MetricsSpec {
235    pipeline: String,
236    component: String,
237    deser: Arc<DeserMetrics>,
238}
239
240impl MetricsSpec {
241    fn op_handle(&self, idx: usize, kind: &'static str) -> Arc<OperatorMetrics> {
242        let labels = ComponentLabels::new(
243            self.pipeline.clone(),
244            format!("{}.{idx}_{kind}", self.component),
245            kind,
246        );
247        Arc::new(OperatorMetrics::new(&labels))
248    }
249}
250
251fn meter_for(metrics: &Option<MetricsSpec>, idx: usize, kind: &'static str) -> OpMeterSlot {
252    OpMeterSlot(OpMeter::new(
253        metrics.as_ref().map(|m| m.op_handle(idx, kind)),
254    ))
255}
256
257/// Fluent builder for one pipeline's operator chain. `DF` is the
258/// deserializer's record family; `CurF` the family at the current end of
259/// the chain (changed by `map_rec` and `flat_map`, and by `map` for owned
260/// payloads).
261#[derive(Clone, Debug)]
262pub struct ChainBuilder<DF: RecFamily, CurF: RecFamily, D, P> {
263    deser: D,
264    parts: P,
265    deser_policy: ErrorPolicy,
266    metrics: Option<MetricsSpec>,
267    stage_idx: usize,
268    _fam: PhantomData<fn() -> (DF, CurF)>,
269}
270
271/// Start a chain from a deserializer producing family `F`.
272pub fn chain<F: RecFamily, D: Deserializer<F>>(deser: D) -> ChainBuilder<F, F, D, Root> {
273    ChainBuilder {
274        deser,
275        parts: Root,
276        deser_policy: ErrorPolicy::Skip,
277        metrics: None,
278        stage_idx: 0,
279        _fam: PhantomData,
280    }
281}
282
283/// Start a chain from a deserializer producing owned records `T`.
284pub fn chain_owned<T, D>(deser: D) -> ChainBuilder<Owned<T>, Owned<T>, D, Root>
285where
286    T: Send + 'static,
287    D: Deserializer<Owned<T>>,
288{
289    chain(deser)
290}
291
292impl<DF: RecFamily, CurF: RecFamily, D, P> ChainBuilder<DF, CurF, D, P> {
293    /// Enable framework metrics for every stage of this chain. Must be
294    /// called before any stage is added so all stages get handles.
295    ///
296    /// # Panics
297    ///
298    /// Panics if stages were already added.
299    #[must_use]
300    pub fn with_metrics(
301        mut self,
302        pipeline: impl Into<String>,
303        component: impl Into<String>,
304    ) -> Self {
305        assert_eq!(
306            self.stage_idx, 0,
307            "with_metrics must be called before stages are added"
308        );
309        let pipeline = pipeline.into();
310        let component = component.into();
311        let deser_labels = ComponentLabels::new(
312            pipeline.clone(),
313            format!("{component}.deserializer"),
314            "deserializer",
315        );
316        self.metrics = Some(MetricsSpec {
317            pipeline,
318            component,
319            deser: Arc::new(DeserMetrics::new(&deser_labels)),
320        });
321        self
322    }
323
324    /// Error policy for the deserializer stage (default: `Skip`).
325    #[must_use]
326    pub fn deser_error_policy(mut self, policy: ErrorPolicy) -> Self {
327        self.deser_policy = policy;
328        self
329    }
330
331    /// Transform each record into family `NF`. For borrowing families pass
332    /// a `fn` item (see [the module docs](crate::ops)); for owned payloads
333    /// [`ChainBuilder::map`] is more ergonomic.
334    #[must_use]
335    pub fn map_rec<NF, G>(self, f: G) -> ChainBuilder<DF, NF, D, MapPart<P, G>>
336    where
337        NF: RecFamily,
338        G: for<'buf> MapFn<CurF::Rec<'buf>, NF::Rec<'buf>>,
339    {
340        let Self {
341            deser,
342            parts,
343            deser_policy,
344            metrics,
345            stage_idx,
346            _fam,
347        } = self;
348        let meter = meter_for(&metrics, stage_idx, "map");
349        ChainBuilder {
350            deser,
351            parts: MapPart {
352                prev: parts,
353                f,
354                meter,
355            },
356            deser_policy,
357            metrics,
358            stage_idx: stage_idx + 1,
359            _fam: PhantomData,
360        }
361    }
362
363    /// Fallibly transform each record into family `NF` with a per-stage
364    /// [`ErrorPolicy`]. For borrowing families pass a `fn` item.
365    #[must_use]
366    pub fn try_map_rec<NF, G, E>(
367        self,
368        f: G,
369        policy: ErrorPolicy,
370    ) -> ChainBuilder<DF, NF, D, TryMapPart<P, G>>
371    where
372        NF: RecFamily,
373        G: for<'buf> TryMapFn<CurF::Rec<'buf>, NF::Rec<'buf>, E>,
374        E: std::fmt::Display,
375    {
376        let Self {
377            deser,
378            parts,
379            deser_policy,
380            metrics,
381            stage_idx,
382            _fam,
383        } = self;
384        let meter = meter_for(&metrics, stage_idx, "try_map");
385        ChainBuilder {
386            deser,
387            parts: TryMapPart {
388                prev: parts,
389                f,
390                policy,
391                component: Arc::from(format!("try_map_{stage_idx}")),
392                meter,
393            },
394            deser_policy,
395            metrics,
396            stage_idx: stage_idx + 1,
397            _fam: PhantomData,
398        }
399    }
400
401    /// Keep only records whose payload satisfies the predicate.
402    #[must_use]
403    pub fn filter<Pr>(self, p: Pr) -> ChainBuilder<DF, CurF, D, FilterPart<P, Pr>>
404    where
405        Pr: for<'buf> FnMut(&CurF::Rec<'buf>) -> bool,
406    {
407        let Self {
408            deser,
409            parts,
410            deser_policy,
411            metrics,
412            stage_idx,
413            _fam,
414        } = self;
415        let meter = meter_for(&metrics, stage_idx, "filter");
416        ChainBuilder {
417            deser,
418            parts: FilterPart {
419                prev: parts,
420                p,
421                meter,
422            },
423            deser_policy,
424            metrics,
425            stage_idx: stage_idx + 1,
426            _fam: PhantomData,
427        }
428    }
429
430    /// Observe each record's payload without transforming it.
431    #[must_use]
432    pub fn inspect<G>(self, f: G) -> ChainBuilder<DF, CurF, D, InspectPart<P, G>>
433    where
434        G: for<'buf> FnMut(&CurF::Rec<'buf>),
435    {
436        let Self {
437            deser,
438            parts,
439            deser_policy,
440            metrics,
441            stage_idx,
442            _fam,
443        } = self;
444        ChainBuilder {
445            deser,
446            parts: InspectPart { prev: parts, f },
447            deser_policy,
448            metrics,
449            stage_idx: stage_idx + 1,
450            _fam: PhantomData,
451        }
452    }
453
454    /// Expand each record into 0..N records of family `OutF` through a
455    /// stack-borrowed [`Emitter`].
456    #[must_use]
457    pub fn flat_map<OutF, G>(self, g: G) -> ChainBuilder<DF, OutF, D, FlatMapPart<OutF, P, G>>
458    where
459        OutF: RecFamily,
460        G: for<'buf> FnMut(CurF::Rec<'buf>, &mut Emitter<'_, OutF>),
461    {
462        let Self {
463            deser,
464            parts,
465            deser_policy,
466            metrics,
467            stage_idx,
468            _fam,
469        } = self;
470        let meter = meter_for(&metrics, stage_idx, "flat_map");
471        ChainBuilder {
472            deser,
473            parts: FlatMapPart {
474                prev: parts,
475                g,
476                meter,
477                _out: PhantomData,
478            },
479            deser_policy,
480            metrics,
481            stage_idx: stage_idx + 1,
482            _fam: PhantomData,
483        }
484    }
485
486    /// Terminate the chain into a sink. Records are routed by `router`,
487    /// encoded by `encoder` on the pipeline thread, and handed to the sink
488    /// workers through `queues`. `router` may be any meta-only
489    /// [`ShardRouter`](crate::sink::ShardRouter) (bridged automatically,
490    /// [`KeyHashRouter`](crate::sink::KeyHashRouter) being the default
491    /// choice) or a record-aware
492    /// [`RecordRouter`](crate::sink::RecordRouter).
493    #[must_use]
494    pub fn sink<E, R>(
495        self,
496        encoder: E,
497        router: R,
498        cfg: ChunkConfig,
499        queues: ShardQueues,
500        budget: Arc<InflightBudget>,
501    ) -> SinkedChain<DF, CurF, D, P, E, R> {
502        let handoff_meter = meter_for(&self.metrics, self.stage_idx, "sink_handoff");
503        SinkedChain {
504            builder: self,
505            encoder,
506            router,
507            cfg,
508            queues,
509            budget,
510            handoff_meter,
511        }
512    }
513
514    /// Terminate the chain into a **split sink**. Each record is routed to
515    /// exactly one of several typed sink branches (each its own
516    /// table/schema/encoder), declared with [`SplitBuilder::add`] and
517    /// dispatched by the closure passed to [`SplitBuilder::route`]. Each
518    /// branch's chunking comes from its own [`SinkCtx`](crate::ops::SinkCtx)
519    /// (resolved from that sink's per-name YAML `chunk:` block, or
520    /// `SinkOptions::with_chunk`, or the default), so a split tunes each
521    /// destination independently. `unmatched` is the policy for a record that
522    /// reaches no branch. [`ErrorPolicy::Fail`] (the operator default) stops
523    /// the pipeline; [`ErrorPolicy::Skip`] drops it and counts
524    /// `spate_operator_records_dropped_total{reason="unrouted"}`.
525    #[must_use]
526    pub fn split(self, unmatched: ErrorPolicy) -> SplitBuilder<DF, CurF, D, P> {
527        SplitBuilder {
528            builder: self,
529            unmatched,
530            branches: Vec::new(),
531            next_idx: 0,
532        }
533    }
534}
535
536/// Closure-friendly transforms for chains whose current records are owned.
537impl<DF: RecFamily, T: Send + 'static, D, P> ChainBuilder<DF, Owned<T>, D, P> {
538    /// Transform each record's payload. Bare closures infer; the output
539    /// type may differ (the chain's family becomes `Owned<U>`).
540    #[must_use]
541    pub fn map<U, G>(self, f: G) -> ChainBuilder<DF, Owned<U>, D, MapPart<P, G>>
542    where
543        U: Send + 'static,
544        G: FnMut(T) -> U,
545    {
546        let Self {
547            deser,
548            parts,
549            deser_policy,
550            metrics,
551            stage_idx,
552            _fam,
553        } = self;
554        let meter = meter_for(&metrics, stage_idx, "map");
555        ChainBuilder {
556            deser,
557            parts: MapPart {
558                prev: parts,
559                f,
560                meter,
561            },
562            deser_policy,
563            metrics,
564            stage_idx: stage_idx + 1,
565            _fam: PhantomData,
566        }
567    }
568
569    /// Fallibly transform each record's payload with a per-stage
570    /// [`ErrorPolicy`].
571    #[must_use]
572    pub fn try_map<U, G, E>(
573        self,
574        f: G,
575        policy: ErrorPolicy,
576    ) -> ChainBuilder<DF, Owned<U>, D, TryMapPart<P, G>>
577    where
578        U: Send + 'static,
579        G: FnMut(T) -> Result<U, E>,
580        E: std::fmt::Display,
581    {
582        let Self {
583            deser,
584            parts,
585            deser_policy,
586            metrics,
587            stage_idx,
588            _fam,
589        } = self;
590        let meter = meter_for(&metrics, stage_idx, "try_map");
591        ChainBuilder {
592            deser,
593            parts: TryMapPart {
594                prev: parts,
595                f,
596                policy,
597                component: Arc::from(format!("try_map_{stage_idx}")),
598                meter,
599            },
600            deser_policy,
601            metrics,
602            stage_idx: stage_idx + 1,
603            _fam: PhantomData,
604        }
605    }
606}
607
608/// A fully specified chain, ready to build, or to stamp out one instance per
609/// pipeline thread via [`SinkedChain::build_factory`].
610#[derive(Clone, Debug)]
611pub struct SinkedChain<DF: RecFamily, CurF: RecFamily, D, P, E, R> {
612    builder: ChainBuilder<DF, CurF, D, P>,
613    encoder: E,
614    router: R,
615    cfg: ChunkConfig,
616    queues: ShardQueues,
617    budget: Arc<InflightBudget>,
618    handoff_meter: OpMeterSlot,
619}
620
621impl<DF, CurF, D, P, E, R> SinkedChain<DF, CurF, D, P, E, R>
622where
623    DF: RecFamily,
624    CurF: RecFamily,
625    D: Deserializer<DF> + 'static,
626    P: Assemble<SinkHandoff<CurF, E, R>>,
627    P::Out: for<'buf> Collector<<DF as RecFamily>::Rec<'buf>> + StageLifecycle + Send + 'static,
628    E: RowEncoder<CurF> + Clone + 'static,
629    R: RecordRouter<CurF> + Send + 'static,
630{
631    /// Build one chain instance. Consumes the specification; the deserializer
632    /// and router need no `Clone`, but the encoder does. The terminal stage
633    /// mints one encoder per shard (columnar encoders hold per-block state
634    /// that cannot be shared), and every in-tree encoder is `Clone`.
635    #[must_use]
636    pub fn build(self) -> Box<dyn RunnableChain> {
637        let SinkedChain {
638            builder,
639            encoder,
640            router,
641            cfg,
642            queues,
643            budget,
644            handoff_meter,
645        } = self;
646        let term = SinkHandoff::new(
647            encoder,
648            router,
649            queues,
650            budget,
651            cfg,
652            handoff_meter,
653            Arc::from("sink_handoff"),
654        );
655        let ops = builder.parts.assemble(term);
656        Box::new(TypedChain::<DF, D, _>::new(
657            builder.deser,
658            ops,
659            builder.deser_policy,
660            builder.metrics.as_ref().map(|m| Arc::clone(&m.deser)),
661        ))
662    }
663
664    /// Turn the specification into a factory producing one identical chain
665    /// per pipeline thread.
666    #[must_use]
667    pub fn build_factory(self) -> ChainFactory<DF, CurF, D, P, E, R>
668    where
669        D: Clone,
670        E: Clone,
671        R: Clone,
672    {
673        ChainFactory { spec: self }
674    }
675}
676
677/// Stamps out identical chains, one per pipeline thread. `Send + Sync`
678/// when the deserializer, stage closures, encoder, and router are.
679#[derive(Clone, Debug)]
680pub struct ChainFactory<DF: RecFamily, CurF: RecFamily, D, P, E, R> {
681    spec: SinkedChain<DF, CurF, D, P, E, R>,
682}
683
684impl<DF, CurF, D, P, E, R> ChainFactory<DF, CurF, D, P, E, R>
685where
686    DF: RecFamily,
687    CurF: RecFamily,
688    D: Deserializer<DF> + Clone + 'static,
689    P: Assemble<SinkHandoff<CurF, E, R>>,
690    P::Out: for<'buf> Collector<<DF as RecFamily>::Rec<'buf>> + StageLifecycle + Send + 'static,
691    E: RowEncoder<CurF> + Clone + 'static,
692    R: RecordRouter<CurF> + Clone + Send + 'static,
693{
694    /// Build one more identical chain.
695    #[must_use]
696    pub fn make(&self) -> Box<dyn RunnableChain> {
697        let spec = &self.spec;
698        let term = SinkHandoff::new(
699            spec.encoder.clone(),
700            spec.router.clone(),
701            spec.queues.clone(),
702            Arc::clone(&spec.budget),
703            spec.cfg,
704            spec.handoff_meter.clone(),
705            Arc::from("sink_handoff"),
706        );
707        let ops = spec.builder.parts.assemble(term);
708        Box::new(TypedChain::<DF, D, _>::new(
709            spec.builder.deser.clone(),
710            ops,
711            spec.builder.deser_policy,
712            spec.builder.metrics.as_ref().map(|m| Arc::clone(&m.deser)),
713        ))
714    }
715}
716
717/// Accumulates split-sink branches before the routing closure is supplied.
718/// Built by [`ChainBuilder::split`]; each [`add`](Self::add) declares one
719/// destination and hands back a typed handle, then [`route`](Self::route)
720/// takes the closure that dispatches to them.
721pub struct SplitBuilder<DF: RecFamily, CurF: RecFamily, D, P> {
722    builder: ChainBuilder<DF, CurF, D, P>,
723    unmatched: ErrorPolicy,
724    branches: Vec<Box<dyn ErasedBranch>>,
725    next_idx: usize,
726}
727
728impl<DF: RecFamily, CurF: RecFamily, D, P> std::fmt::Debug for SplitBuilder<DF, CurF, D, P> {
729    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
730        f.debug_struct("SplitBuilder")
731            .field("branches", &self.branches.len())
732            .field("unmatched", &self.unmatched)
733            .finish_non_exhaustive()
734    }
735}
736
737impl<DF: RecFamily, CurF: RecFamily, D, P> SplitBuilder<DF, CurF, D, P> {
738    /// Declare one destination branch of family `F` and return its typed,
739    /// `Copy` [`Sink<F>`] handle. `sink` comes from
740    /// [`ChainCtx::sink`](crate::pipeline::ChainCtx::sink); `encoder`/`router`
741    /// are that table's, exactly as for [`ChainBuilder::sink`]. The declaration
742    /// order fixes the handle indices; call once per destination, then
743    /// [`route`](Self::route).
744    #[must_use = "a dropped Sink<F> handle leaves its branch permanently unreachable"]
745    pub fn add<F, E, R>(&mut self, encoder: E, router: R, sink: SinkCtx) -> Sink<F>
746    where
747        F: RecFamily + 'static,
748        E: RowEncoder<F> + Clone + Send + 'static,
749        R: RecordRouter<F> + 'static,
750    {
751        let idx = self.next_idx;
752        self.next_idx += 1;
753        // Per-branch operator meter, labeled by the sink name so each branch's
754        // `spate_operator_*` series is distinct.
755        let meter = OpMeterSlot(OpMeter::new(self.builder.metrics.as_ref().map(|m| {
756            let labels = ComponentLabels::new(
757                m.pipeline.clone(),
758                format!("{}.sink.{}", m.component, sink.name),
759                "sink_handoff",
760            );
761            Arc::new(OperatorMetrics::new(&labels))
762        })));
763        let component: Arc<str> = Arc::from(format!("sink.{}", sink.name));
764        let branch = new_branch::<F, E, R>(
765            encoder,
766            router,
767            sink.queues,
768            sink.budget,
769            sink.chunk,
770            meter,
771            component,
772        );
773        self.branches.push(branch);
774        Sink::new(idx)
775    }
776
777    /// Supply the routing closure (classify + extract per record in one
778    /// `match`) and finish the split. The closure emits each record to exactly
779    /// one branch via [`SplitEmitter::emit`]; emitting to none invokes the
780    /// `unmatched` policy from [`ChainBuilder::split`].
781    #[must_use]
782    pub fn route<G>(self, route: G) -> RoutedSplit<DF, CurF, D, P, G>
783    where
784        G: for<'buf> FnMut(CurF::Rec<'buf>, &mut SplitEmitter<'_>) + Send + 'static,
785    {
786        let handoff_meter = meter_for(&self.builder.metrics, self.builder.stage_idx, "split");
787        RoutedSplit {
788            builder: self.builder,
789            unmatched: self.unmatched,
790            branches: self.branches,
791            route,
792            handoff_meter,
793        }
794    }
795}
796
797/// A split-terminated chain, ready to [`build`](Self::build). Not `Clone`; the
798/// branches are built once. Stamp identical per-thread chains by re-running
799/// the `chains` factory closure (as the pipeline builder does).
800pub struct RoutedSplit<DF: RecFamily, CurF: RecFamily, D, P, G> {
801    builder: ChainBuilder<DF, CurF, D, P>,
802    unmatched: ErrorPolicy,
803    branches: Vec<Box<dyn ErasedBranch>>,
804    route: G,
805    handoff_meter: OpMeterSlot,
806}
807
808impl<DF: RecFamily, CurF: RecFamily, D, P, G> std::fmt::Debug for RoutedSplit<DF, CurF, D, P, G> {
809    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
810        f.debug_struct("RoutedSplit")
811            .field("branches", &self.branches.len())
812            .field("unmatched", &self.unmatched)
813            .finish_non_exhaustive()
814    }
815}
816
817impl<DF, CurF, D, P, G> RoutedSplit<DF, CurF, D, P, G>
818where
819    DF: RecFamily,
820    CurF: RecFamily,
821    D: Deserializer<DF> + 'static,
822    G: for<'buf> FnMut(CurF::Rec<'buf>, &mut SplitEmitter<'_>) + Send + 'static,
823    P: Assemble<SplitTerminal<CurF, G>>,
824    P::Out: for<'buf> Collector<<DF as RecFamily>::Rec<'buf>> + StageLifecycle + Send + 'static,
825{
826    /// Build one chain instance whose terminal is the split sink.
827    #[must_use]
828    pub fn build(self) -> Box<dyn RunnableChain> {
829        let RoutedSplit {
830            builder,
831            unmatched,
832            branches,
833            route,
834            handoff_meter,
835        } = self;
836        let term = SplitTerminal::new(
837            route,
838            branches,
839            unmatched,
840            handoff_meter,
841            Arc::from("split"),
842        );
843        let ops = builder.parts.assemble(term);
844        Box::new(TypedChain::<DF, D, _>::new(
845            builder.deser,
846            ops,
847            builder.deser_policy,
848            builder.metrics.as_ref().map(|m| Arc::clone(&m.deser)),
849        ))
850    }
851}
spate_core/ops/builder.rs

spate_core/ops/
builder.rs
