etl_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//!
8//! # Owned vs borrowed record families
9//!
10//! For owned families ([`Owned<T>`](crate::deser::Owned)) the builder
11//! offers [`ChainBuilder::map`] / [`ChainBuilder::try_map`] with plain
12//! closure bounds — bare closures infer. For **borrowing** families a
13//! `rustc` limitation (E0582: a higher-ranked lifetime may not appear only
14//! in associated-type positions) rules out `FnMut`-with-projection-output
15//! bounds at the definition site; use [`ChainBuilder::map_rec`] /
16//! [`ChainBuilder::try_map_rec`] and pass **`fn` items**, which are
17//! naturally higher-ranked:
18//!
19//! ```ignore
20//! fn shrink<'a>(e: LogEvent<'a>) -> Compact<'a> { /* ... */ }
21//! chain(log_deser).map_rec::<CompactF, _>(shrink)
22//! ```
23//!
24//! [`ChainBuilder::filter`], [`ChainBuilder::inspect`], and
25//! [`ChainBuilder::flat_map`] have no output binding, so a single generic
26//! method serves both kinds of family.
27//!
28//! ```
29//! use etl_core::backpressure::InflightBudget;
30//! use etl_core::deser::{BytesPassthrough, Owned};
31//! use etl_core::error::ErrorPolicy;
32//! use etl_core::ops::{ChunkConfig, chain};
33//! use etl_core::record::Record;
34//! use etl_core::sink::{KeyHashRouter, RowEncoder, shard_queues};
35//! use std::sync::Arc;
36//!
37//! // A trivial encoder writing `<u32 len><bytes>` rows.
38//! #[derive(Clone)]
39//! struct LenPrefix;
40//! impl RowEncoder<Owned<Vec<u8>>> for LenPrefix {
41//!     fn encode<'buf>(
42//!         &mut self,
43//!         rec: &Record<Vec<u8>>,
44//!         buf: &mut bytes::BytesMut,
45//!     ) -> Result<(), etl_core::error::SinkError> {
46//!         buf.extend_from_slice(&(rec.payload.len() as u32).to_le_bytes());
47//!         buf.extend_from_slice(&rec.payload);
48//!         Ok(())
49//!     }
50//! }
51//!
52//! let (queues, _rx) = shard_queues(2, 64);
53//! let budget = Arc::new(InflightBudget::new());
54//!
55//! let mut pipeline_chain = chain(BytesPassthrough)
56//!     .map(|mut bytes: Vec<u8>| {
57//!         bytes.make_ascii_uppercase();
58//!         bytes
59//!     })
60//!     .filter(|bytes: &Vec<u8>| !bytes.is_empty())
61//!     .try_map(
62//!         |bytes: Vec<u8>| String::from_utf8(bytes).map(String::into_bytes),
63//!         ErrorPolicy::Skip,
64//!     )
65//!     .sink(LenPrefix, KeyHashRouter, ChunkConfig::default(), queues, budget)
66//!     .build();
67//! # let _ = &mut pipeline_chain;
68//! ```
69
70use super::chain::{
71    FatalSlot, Filter, FlatMap, Inspect, Map, OpMeter, OpMeterSlot, StageLifecycle, TryMap,
72    TypedChain,
73};
74use super::handoff::{ChunkConfig, SinkHandoff};
75use super::{Collector, Emitter, RunnableChain};
76use crate::backpressure::InflightBudget;
77use crate::deser::{Deserializer, Owned, RecFamily};
78use crate::error::ErrorPolicy;
79use crate::metrics::{ComponentLabels, DeserMetrics, OperatorMetrics};
80use crate::sink::{RowEncoder, ShardQueues, ShardRouter};
81use std::marker::PhantomData;
82use std::sync::Arc;
83
84/// A record-to-record transform between families. Implemented for every
85/// `FnMut(In) -> Out`; expressed as an independent two-parameter trait so
86/// higher-ranked builder bounds stay legal for borrowing families (see the
87/// module docs on E0582). `fn` items satisfy it at every lifetime.
88pub trait MapFn<In, Out>: FnMut(In) -> Out {}
89impl<G, In, Out> MapFn<In, Out> for G where G: FnMut(In) -> Out {}
90
91/// Fallible variant of [`MapFn`].
92pub trait TryMapFn<In, Out, Err>: FnMut(In) -> Result<Out, Err> {}
93impl<G, In, Out, Err> TryMapFn<In, Out, Err> for G where G: FnMut(In) -> Result<Out, Err> {}
94
95/// Assembles recorded parts into the concrete collector stack, given the
96/// terminal stage. Takes `&self` so one set of parts can assemble many
97/// identical chains — stage closures must be `Clone` (plain closures and
98/// closures over `Clone`/`Arc` state are).
99pub trait Assemble<Term> {
100    /// The assembled collector stack.
101    type Out;
102    /// Build the stack around `term`.
103    fn assemble(&self, term: Term) -> Self::Out;
104}
105
106/// The empty stage list.
107#[derive(Clone, Copy, Debug, Default)]
108pub struct Root;
109
110impl<T> Assemble<T> for Root {
111    type Out = T;
112    fn assemble(&self, term: T) -> T {
113        term
114    }
115}
116
117/// Recorded `map`/`map_rec` stage.
118#[derive(Clone, Debug)]
119pub struct MapPart<Prev, G> {
120    prev: Prev,
121    f: G,
122    meter: OpMeterSlot,
123}
124
125impl<Prev, G: Clone, Term> Assemble<Term> for MapPart<Prev, G>
126where
127    Prev: Assemble<Map<G, Term>>,
128{
129    type Out = Prev::Out;
130    fn assemble(&self, term: Term) -> Self::Out {
131        self.prev.assemble(Map {
132            f: self.f.clone(),
133            next: term,
134            meter: self.meter.clone(),
135        })
136    }
137}
138
139/// Recorded `filter` stage.
140#[derive(Clone, Debug)]
141pub struct FilterPart<Prev, P> {
142    prev: Prev,
143    p: P,
144    meter: OpMeterSlot,
145}
146
147impl<Prev, P: Clone, Term> Assemble<Term> for FilterPart<Prev, P>
148where
149    Prev: Assemble<Filter<P, Term>>,
150{
151    type Out = Prev::Out;
152    fn assemble(&self, term: Term) -> Self::Out {
153        self.prev.assemble(Filter {
154            p: self.p.clone(),
155            next: term,
156            meter: self.meter.clone(),
157        })
158    }
159}
160
161/// Recorded `inspect` stage.
162#[derive(Clone, Debug)]
163pub struct InspectPart<Prev, G> {
164    prev: Prev,
165    f: G,
166}
167
168impl<Prev, G: Clone, Term> Assemble<Term> for InspectPart<Prev, G>
169where
170    Prev: Assemble<Inspect<G, Term>>,
171{
172    type Out = Prev::Out;
173    fn assemble(&self, term: Term) -> Self::Out {
174        self.prev.assemble(Inspect {
175            f: self.f.clone(),
176            next: term,
177        })
178    }
179}
180
181/// Recorded `try_map`/`try_map_rec` stage.
182#[derive(Clone, Debug)]
183pub struct TryMapPart<Prev, G> {
184    prev: Prev,
185    f: G,
186    policy: ErrorPolicy,
187    component: Arc<str>,
188    meter: OpMeterSlot,
189}
190
191impl<Prev, G: Clone, Term> Assemble<Term> for TryMapPart<Prev, G>
192where
193    Prev: Assemble<TryMap<G, Term>>,
194{
195    type Out = Prev::Out;
196    fn assemble(&self, term: Term) -> Self::Out {
197        self.prev.assemble(TryMap {
198            f: self.f.clone(),
199            next: term,
200            policy: self.policy,
201            component: Arc::clone(&self.component),
202            meter: self.meter.clone(),
203            fatal: FatalSlot(None),
204        })
205    }
206}
207
208/// Recorded `flat_map` stage.
209#[derive(Clone, Debug)]
210pub struct FlatMapPart<OutF: RecFamily, Prev, G> {
211    prev: Prev,
212    g: G,
213    meter: OpMeterSlot,
214    _out: PhantomData<fn() -> OutF>,
215}
216
217impl<OutF: RecFamily, Prev, G: Clone, Term> Assemble<Term> for FlatMapPart<OutF, Prev, G>
218where
219    Prev: Assemble<FlatMap<OutF, G, Term>>,
220{
221    type Out = Prev::Out;
222    fn assemble(&self, term: Term) -> Self::Out {
223        self.prev.assemble(FlatMap {
224            g: self.g.clone(),
225            next: term,
226            meter: self.meter.clone(),
227            _out: PhantomData,
228        })
229    }
230}
231
232#[derive(Clone, Debug)]
233struct MetricsSpec {
234    pipeline: String,
235    component: String,
236    deser: Arc<DeserMetrics>,
237}
238
239impl MetricsSpec {
240    fn op_handle(&self, idx: usize, kind: &'static str) -> Arc<OperatorMetrics> {
241        let labels = ComponentLabels::new(
242            self.pipeline.clone(),
243            format!("{}.{idx}_{kind}", self.component),
244            kind,
245        );
246        Arc::new(OperatorMetrics::new(&labels))
247    }
248}
249
250fn meter_for(metrics: &Option<MetricsSpec>, idx: usize, kind: &'static str) -> OpMeterSlot {
251    OpMeterSlot(OpMeter::new(
252        metrics.as_ref().map(|m| m.op_handle(idx, kind)),
253    ))
254}
255
256/// Fluent builder for one pipeline's operator chain. `DF` is the
257/// deserializer's record family; `CurF` the family at the current end of
258/// the chain (changed by `map_rec` and `flat_map`, and by `map` for owned
259/// payloads).
260#[derive(Clone, Debug)]
261pub struct ChainBuilder<DF: RecFamily, CurF: RecFamily, D, P> {
262    deser: D,
263    parts: P,
264    deser_policy: ErrorPolicy,
265    metrics: Option<MetricsSpec>,
266    stage_idx: usize,
267    _fam: PhantomData<fn() -> (DF, CurF)>,
268}
269
270/// Start a chain from a deserializer producing family `F`.
271pub fn chain<F: RecFamily, D: Deserializer<F>>(deser: D) -> ChainBuilder<F, F, D, Root> {
272    ChainBuilder {
273        deser,
274        parts: Root,
275        deser_policy: ErrorPolicy::Skip,
276        metrics: None,
277        stage_idx: 0,
278        _fam: PhantomData,
279    }
280}
281
282/// Start a chain from a deserializer producing owned records `T`.
283pub fn chain_owned<T, D>(deser: D) -> ChainBuilder<Owned<T>, Owned<T>, D, Root>
284where
285    T: Send + 'static,
286    D: Deserializer<Owned<T>>,
287{
288    chain(deser)
289}
290
291impl<DF: RecFamily, CurF: RecFamily, D, P> ChainBuilder<DF, CurF, D, P> {
292    /// Enable framework metrics for every stage of this chain. Must be
293    /// called before any stage is added so all stages get handles.
294    ///
295    /// # Panics
296    ///
297    /// Panics if stages were already added.
298    #[must_use]
299    pub fn with_metrics(
300        mut self,
301        pipeline: impl Into<String>,
302        component: impl Into<String>,
303    ) -> Self {
304        assert_eq!(
305            self.stage_idx, 0,
306            "with_metrics must be called before stages are added"
307        );
308        let pipeline = pipeline.into();
309        let component = component.into();
310        let deser_labels = ComponentLabels::new(
311            pipeline.clone(),
312            format!("{component}.deserializer"),
313            "deserializer",
314        );
315        self.metrics = Some(MetricsSpec {
316            pipeline,
317            component,
318            deser: Arc::new(DeserMetrics::new(&deser_labels)),
319        });
320        self
321    }
322
323    /// Error policy for the deserializer stage (default: `Skip`).
324    #[must_use]
325    pub fn deser_error_policy(mut self, policy: ErrorPolicy) -> Self {
326        self.deser_policy = policy;
327        self
328    }
329
330    /// Transform each record into family `NF`. For borrowing families pass
331    /// a `fn` item (see the module docs); for owned payloads
332    /// [`ChainBuilder::map`] is more ergonomic.
333    #[must_use]
334    pub fn map_rec<NF, G>(self, f: G) -> ChainBuilder<DF, NF, D, MapPart<P, G>>
335    where
336        NF: RecFamily,
337        G: for<'buf> MapFn<CurF::Rec<'buf>, NF::Rec<'buf>>,
338    {
339        let Self {
340            deser,
341            parts,
342            deser_policy,
343            metrics,
344            stage_idx,
345            _fam,
346        } = self;
347        let meter = meter_for(&metrics, stage_idx, "map");
348        ChainBuilder {
349            deser,
350            parts: MapPart {
351                prev: parts,
352                f,
353                meter,
354            },
355            deser_policy,
356            metrics,
357            stage_idx: stage_idx + 1,
358            _fam: PhantomData,
359        }
360    }
361
362    /// Fallibly transform each record into family `NF` with a per-stage
363    /// [`ErrorPolicy`]. For borrowing families pass a `fn` item.
364    #[must_use]
365    pub fn try_map_rec<NF, G, E>(
366        self,
367        f: G,
368        policy: ErrorPolicy,
369    ) -> ChainBuilder<DF, NF, D, TryMapPart<P, G>>
370    where
371        NF: RecFamily,
372        G: for<'buf> TryMapFn<CurF::Rec<'buf>, NF::Rec<'buf>, E>,
373        E: std::fmt::Display,
374    {
375        let Self {
376            deser,
377            parts,
378            deser_policy,
379            metrics,
380            stage_idx,
381            _fam,
382        } = self;
383        let meter = meter_for(&metrics, stage_idx, "try_map");
384        ChainBuilder {
385            deser,
386            parts: TryMapPart {
387                prev: parts,
388                f,
389                policy,
390                component: Arc::from(format!("try_map_{stage_idx}")),
391                meter,
392            },
393            deser_policy,
394            metrics,
395            stage_idx: stage_idx + 1,
396            _fam: PhantomData,
397        }
398    }
399
400    /// Keep only records whose payload satisfies the predicate.
401    #[must_use]
402    pub fn filter<Pr>(self, p: Pr) -> ChainBuilder<DF, CurF, D, FilterPart<P, Pr>>
403    where
404        Pr: for<'buf> FnMut(&CurF::Rec<'buf>) -> bool,
405    {
406        let Self {
407            deser,
408            parts,
409            deser_policy,
410            metrics,
411            stage_idx,
412            _fam,
413        } = self;
414        let meter = meter_for(&metrics, stage_idx, "filter");
415        ChainBuilder {
416            deser,
417            parts: FilterPart {
418                prev: parts,
419                p,
420                meter,
421            },
422            deser_policy,
423            metrics,
424            stage_idx: stage_idx + 1,
425            _fam: PhantomData,
426        }
427    }
428
429    /// Observe each record's payload without transforming it.
430    #[must_use]
431    pub fn inspect<G>(self, f: G) -> ChainBuilder<DF, CurF, D, InspectPart<P, G>>
432    where
433        G: for<'buf> FnMut(&CurF::Rec<'buf>),
434    {
435        let Self {
436            deser,
437            parts,
438            deser_policy,
439            metrics,
440            stage_idx,
441            _fam,
442        } = self;
443        ChainBuilder {
444            deser,
445            parts: InspectPart { prev: parts, f },
446            deser_policy,
447            metrics,
448            stage_idx: stage_idx + 1,
449            _fam: PhantomData,
450        }
451    }
452
453    /// Expand each record into 0..N records of family `OutF` through a
454    /// stack-borrowed [`Emitter`].
455    #[must_use]
456    pub fn flat_map<OutF, G>(self, g: G) -> ChainBuilder<DF, OutF, D, FlatMapPart<OutF, P, G>>
457    where
458        OutF: RecFamily,
459        G: for<'buf> FnMut(CurF::Rec<'buf>, &mut Emitter<'_, OutF>),
460    {
461        let Self {
462            deser,
463            parts,
464            deser_policy,
465            metrics,
466            stage_idx,
467            _fam,
468        } = self;
469        let meter = meter_for(&metrics, stage_idx, "flat_map");
470        ChainBuilder {
471            deser,
472            parts: FlatMapPart {
473                prev: parts,
474                g,
475                meter,
476                _out: PhantomData,
477            },
478            deser_policy,
479            metrics,
480            stage_idx: stage_idx + 1,
481            _fam: PhantomData,
482        }
483    }
484
485    /// Terminate the chain into a sink: records are routed by `router`,
486    /// encoded by `encoder` on the pipeline thread, and handed to the sink
487    /// workers through `queues`.
488    #[must_use]
489    pub fn sink<E, R>(
490        self,
491        encoder: E,
492        router: R,
493        cfg: ChunkConfig,
494        queues: ShardQueues,
495        budget: Arc<InflightBudget>,
496    ) -> SinkedChain<DF, CurF, D, P, E, R> {
497        let handoff_meter = meter_for(&self.metrics, self.stage_idx, "sink_handoff");
498        SinkedChain {
499            builder: self,
500            encoder,
501            router,
502            cfg,
503            queues,
504            budget,
505            handoff_meter,
506        }
507    }
508}
509
510/// Closure-friendly transforms for chains whose current records are owned.
511impl<DF: RecFamily, T: Send + 'static, D, P> ChainBuilder<DF, Owned<T>, D, P> {
512    /// Transform each record's payload. Bare closures infer; the output
513    /// type may differ (the chain's family becomes `Owned<U>`).
514    #[must_use]
515    pub fn map<U, G>(self, f: G) -> ChainBuilder<DF, Owned<U>, D, MapPart<P, G>>
516    where
517        U: Send + 'static,
518        G: FnMut(T) -> U,
519    {
520        let Self {
521            deser,
522            parts,
523            deser_policy,
524            metrics,
525            stage_idx,
526            _fam,
527        } = self;
528        let meter = meter_for(&metrics, stage_idx, "map");
529        ChainBuilder {
530            deser,
531            parts: MapPart {
532                prev: parts,
533                f,
534                meter,
535            },
536            deser_policy,
537            metrics,
538            stage_idx: stage_idx + 1,
539            _fam: PhantomData,
540        }
541    }
542
543    /// Fallibly transform each record's payload with a per-stage
544    /// [`ErrorPolicy`].
545    #[must_use]
546    pub fn try_map<U, G, E>(
547        self,
548        f: G,
549        policy: ErrorPolicy,
550    ) -> ChainBuilder<DF, Owned<U>, D, TryMapPart<P, G>>
551    where
552        U: Send + 'static,
553        G: FnMut(T) -> Result<U, E>,
554        E: std::fmt::Display,
555    {
556        let Self {
557            deser,
558            parts,
559            deser_policy,
560            metrics,
561            stage_idx,
562            _fam,
563        } = self;
564        let meter = meter_for(&metrics, stage_idx, "try_map");
565        ChainBuilder {
566            deser,
567            parts: TryMapPart {
568                prev: parts,
569                f,
570                policy,
571                component: Arc::from(format!("try_map_{stage_idx}")),
572                meter,
573            },
574            deser_policy,
575            metrics,
576            stage_idx: stage_idx + 1,
577            _fam: PhantomData,
578        }
579    }
580}
581
582/// A fully specified chain, ready to build — or to stamp out one instance
583/// per pipeline thread via [`SinkedChain::build_factory`].
584#[derive(Clone, Debug)]
585pub struct SinkedChain<DF: RecFamily, CurF: RecFamily, D, P, E, R> {
586    builder: ChainBuilder<DF, CurF, D, P>,
587    encoder: E,
588    router: R,
589    cfg: ChunkConfig,
590    queues: ShardQueues,
591    budget: Arc<InflightBudget>,
592    handoff_meter: OpMeterSlot,
593}
594
595impl<DF, CurF, D, P, E, R> SinkedChain<DF, CurF, D, P, E, R>
596where
597    DF: RecFamily,
598    CurF: RecFamily,
599    D: Deserializer<DF> + 'static,
600    P: Assemble<SinkHandoff<CurF, E, R>>,
601    P::Out: for<'buf> Collector<<DF as RecFamily>::Rec<'buf>> + StageLifecycle + Send + 'static,
602    E: RowEncoder<CurF> + Clone + 'static,
603    R: ShardRouter + Send + 'static,
604{
605    /// Build one chain instance. Consumes the specification; the deserializer
606    /// and router need no `Clone`, but the encoder does — the terminal stage
607    /// mints one encoder per shard (columnar encoders hold per-block state
608    /// that cannot be shared), and every in-tree encoder is `Clone`.
609    #[must_use]
610    pub fn build(self) -> Box<dyn RunnableChain> {
611        let SinkedChain {
612            builder,
613            encoder,
614            router,
615            cfg,
616            queues,
617            budget,
618            handoff_meter,
619        } = self;
620        let term = SinkHandoff::new(
621            encoder,
622            router,
623            queues,
624            budget,
625            cfg,
626            handoff_meter,
627            Arc::from("sink_handoff"),
628        );
629        let ops = builder.parts.assemble(term);
630        Box::new(TypedChain::<DF, D, _>::new(
631            builder.deser,
632            ops,
633            builder.deser_policy,
634            builder.metrics.as_ref().map(|m| Arc::clone(&m.deser)),
635        ))
636    }
637
638    /// Turn the specification into a factory producing one identical chain
639    /// per pipeline thread.
640    #[must_use]
641    pub fn build_factory(self) -> ChainFactory<DF, CurF, D, P, E, R>
642    where
643        D: Clone,
644        E: Clone,
645        R: Clone,
646    {
647        ChainFactory { spec: self }
648    }
649}
650
651/// Stamps out identical chains — one per pipeline thread. `Send + Sync`
652/// when the deserializer, stage closures, encoder, and router are.
653#[derive(Clone, Debug)]
654pub struct ChainFactory<DF: RecFamily, CurF: RecFamily, D, P, E, R> {
655    spec: SinkedChain<DF, CurF, D, P, E, R>,
656}
657
658impl<DF, CurF, D, P, E, R> ChainFactory<DF, CurF, D, P, E, R>
659where
660    DF: RecFamily,
661    CurF: RecFamily,
662    D: Deserializer<DF> + Clone + 'static,
663    P: Assemble<SinkHandoff<CurF, E, R>>,
664    P::Out: for<'buf> Collector<<DF as RecFamily>::Rec<'buf>> + StageLifecycle + Send + 'static,
665    E: RowEncoder<CurF> + Clone + 'static,
666    R: ShardRouter + Clone + Send + 'static,
667{
668    /// Build one more identical chain.
669    #[must_use]
670    pub fn make(&self) -> Box<dyn RunnableChain> {
671        let spec = &self.spec;
672        let term = SinkHandoff::new(
673            spec.encoder.clone(),
674            spec.router.clone(),
675            spec.queues.clone(),
676            Arc::clone(&spec.budget),
677            spec.cfg,
678            spec.handoff_meter.clone(),
679            Arc::from("sink_handoff"),
680        );
681        let ops = spec.builder.parts.assemble(term);
682        Box::new(TypedChain::<DF, D, _>::new(
683            spec.builder.deser.clone(),
684            ops,
685            spec.builder.deser_policy,
686            spec.builder.metrics.as_ref().map(|m| Arc::clone(&m.deser)),
687        ))
688    }
689}
etl_core/ops/builder.rs

etl_core/ops/
builder.rs
