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

spate_core/ops/
builder.rs
