spate_core/ops/

split.rs

//! The split terminal: route each record to exactly one of N typed sink
//! branches, each with its own schema, encoder, router, and shard queues.
//!
//! Where [`SinkHandoff`](super::handoff::SinkHandoff) is one table, the split
//! terminal fans a heterogeneously-typed stream out across many. The user
//! writes a single `match` (classify + extract in the same arm) and dispatches
//! with [`SplitEmitter::emit`]; a record that reaches no branch follows the
//! configured [`ErrorPolicy`] (`Fail`, the default, stops the pipeline;
//! `Skip` drops it and counts `spate_operator_records_dropped_total{reason="unrouted"}`).
//!
//! # How the typed dispatch stays cheap and object-safe
//!
//! Each branch is a [`SinkHandoff<F, BoxedEncoder<F>, BoxedRouter<F>>`](super::handoff::SinkHandoff).
//! Its encoder and router are erased, so the branch's concrete type depends
//! only on the destination family `F`. A [`SplitBranch<F>`] handle (a plain
//! index plus `F`) therefore names the exact concrete type, so
//! [`SplitEmitter::emit`] recovers it with one `Any` downcast, then routes and
//! encodes through the branch's boxed router/encoder straight into that
//! branch's per-shard buffer.
//! That costs one virtual call each per record over the single-sink path's
//! concrete types.
//!
//! The at-least-once machinery is inherited unchanged. Each branch clones the
//! poll batch's [`AckRef`](crate::checkpoint::AckRef) into its own fail-on-drop
//! `AckSet`, so the source watermark holds until *every* branch that received a
//! derived record has durably written, and any branch's failure stalls it.

use super::Collector;
use super::chain::{FatalSlot, OpMeterSlot, StageLifecycle};
use super::handoff::{ChunkConfig, SinkHandoff};
use crate::backpressure::InflightBudget;
use crate::checkpoint::AckRef;
use crate::deser::RecFamily;
use crate::error::{ErrorPolicy, FatalError, SinkError};
use crate::record::{Flow, Record, RecordMeta};
use crate::sink::{RecordRouter, RowEncoder, ShardQueues};
use bytes::BytesMut;
use std::any::Any;
use std::marker::PhantomData;
use std::sync::Arc;
use std::time::Duration;

/// The per-sink handles a split branch needs, resolved by name from the
/// chain factory's [`ChainCtx`](crate::pipeline::ChainCtx) via
/// [`ChainCtx::sink`](crate::pipeline::ChainCtx::sink). The bundle carries
/// the sink's name, so [`SplitBuilder::add`](super::SplitBuilder::add) does not
/// repeat it.
#[derive(Clone, Debug)]
#[non_exhaustive]
pub struct SplitTarget {
    pub(crate) name: String,
    pub(crate) queues: ShardQueues,
    pub(crate) budget: Arc<InflightBudget>,
    /// This branch's resolved terminal-stage chunking (its per-sink YAML
    /// `chunk:` block, `SinkOptions::with_chunk`, or the default), applied by
    /// [`SplitBuilder::add`](super::SplitBuilder::add).
    pub(crate) chunk: ChunkConfig,
}

impl SplitTarget {
    /// Bundle a named sink's queues and the shared in-flight budget. Chunking
    /// starts at [`ChunkConfig::default`]; override it with
    /// [`with_chunk`](Self::with_chunk). (Builder pipelines do not call this;
    /// [`ChainCtx::sink`](crate::pipeline::ChainCtx::sink) hands out a fully
    /// resolved `SplitTarget`.)
    #[must_use]
    pub fn new(name: String, queues: ShardQueues, budget: Arc<InflightBudget>) -> Self {
        SplitTarget {
            name,
            queues,
            budget,
            chunk: ChunkConfig::default(),
        }
    }

    /// Set this branch's terminal-stage chunking, the manual-assembly
    /// counterpart to the per-sink YAML `chunk:` block.
    #[must_use]
    pub fn with_chunk(mut self, chunk: ChunkConfig) -> Self {
        self.chunk = chunk;
        self
    }
}

/// A typed, `Copy` handle to one split branch, carrying a branch index plus
/// the destination family, so [`SplitEmitter::emit`] both type-checks the row
/// and recovers the branch with zero per-call lookup. Minted by
/// [`SplitBuilder::add`](super::SplitBuilder::add).
pub struct SplitBranch<F: RecFamily> {
    idx: usize,
    _f: PhantomData<fn() -> F>,
}

impl<F: RecFamily> SplitBranch<F> {
    pub(crate) fn new(idx: usize) -> Self {
        SplitBranch {
            idx,
            _f: PhantomData,
        }
    }
}

impl<F: RecFamily> Clone for SplitBranch<F> {
    fn clone(&self) -> Self {
        *self
    }
}

impl<F: RecFamily> Copy for SplitBranch<F> {}

impl<F: RecFamily> std::fmt::Debug for SplitBranch<F> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("SplitBranch")
            .field("idx", &self.idx)
            .finish()
    }
}

// ── Type-erased encoder/router so a branch's concrete type keys on `F` ──────

/// A [`RowEncoder`] that can clone itself into a box, so a boxed encoder is
/// still `Clone` (the terminal stage mints one encoder per shard).
trait EncoderClone<F: RecFamily>: RowEncoder<F> {
    /// Clone into a fresh box.
    fn clone_box(&self) -> Box<dyn EncoderClone<F>>;
}

impl<F: RecFamily, T> EncoderClone<F> for T
where
    T: RowEncoder<F> + Clone + 'static,
{
    fn clone_box(&self) -> Box<dyn EncoderClone<F>> {
        Box::new(self.clone())
    }
}

/// A branch's encoder, erased to depend only on the destination family.
type BoxedEncoder<F> = Box<dyn EncoderClone<F>>;

impl<F: RecFamily> Clone for BoxedEncoder<F> {
    fn clone(&self) -> Self {
        // Dispatch through the trait object to the *concrete* encoder's
        // `clone_box`. `self.clone_box()` would re-select the blanket impl
        // (which also covers `Box<dyn EncoderClone>`) and recurse into this
        // `clone` infinitely. `(**self)` pins it to the vtable.
        (**self).clone_box()
    }
}

impl<F: RecFamily> RowEncoder<F> for BoxedEncoder<F> {
    fn encode<'buf>(
        &mut self,
        rec: &Record<F::Rec<'buf>>,
        buf: &mut BytesMut,
    ) -> Result<(), SinkError> {
        (**self).encode(rec, buf)
    }

    fn buffered_bytes(&self) -> usize {
        (**self).buffered_bytes()
    }

    fn finish_chunk(&mut self, buf: &mut BytesMut) -> Result<(), SinkError> {
        (**self).finish_chunk(buf)
    }
}

/// A branch's router, erased to depend only on the destination family.
type BoxedRouter<F> = Box<dyn RecordRouter<F>>;

impl<F: RecFamily> RecordRouter<F> for BoxedRouter<F> {
    fn route_record<'buf>(&self, rec: &Record<F::Rec<'buf>>, num_shards: usize) -> usize {
        (**self).route_record(rec, num_shards)
    }
}

/// A branch's concrete type: a [`SinkHandoff`] over the erased encoder/router,
/// determined by the destination family alone.
type Branch<F> = SinkHandoff<F, BoxedEncoder<F>, BoxedRouter<F>>;

// ── Object-safe branch storage ──────────────────────────────────────────────

/// The lifecycle a split branch exposes to the terminal, plus `Any` recovery
/// for the typed emit path. Object-safe (no destination-family type appears in
/// the signatures), so branches of different families live in one `Vec`.
pub(crate) trait ErasedBranch: Send {
    fn relieve(&mut self) -> Flow;
    fn flush_terminal(&mut self) -> Flow;
    fn take_fatal(&mut self) -> Option<FatalError>;
    fn on_batch_end(&mut self, elapsed: Duration);
    fn as_any_mut(&mut self) -> &mut dyn Any;
}

impl<F, E, R> ErasedBranch for SinkHandoff<F, E, R>
where
    F: RecFamily + 'static,
    E: RowEncoder<F> + Clone + 'static,
    R: RecordRouter<F> + 'static,
{
    fn relieve(&mut self) -> Flow {
        StageLifecycle::relieve(self)
    }

    fn flush_terminal(&mut self) -> Flow {
        StageLifecycle::flush_terminal(self)
    }

    fn take_fatal(&mut self) -> Option<FatalError> {
        StageLifecycle::take_fatal(self)
    }

    fn on_batch_end(&mut self, elapsed: Duration) {
        StageLifecycle::on_batch_end(self, elapsed);
    }

    fn as_any_mut(&mut self) -> &mut dyn Any {
        self
    }
}

/// Build one erased branch from a concrete encoder/router pair.
pub(crate) fn new_branch<F, E, R>(
    encoder: E,
    router: R,
    queues: ShardQueues,
    budget: Arc<InflightBudget>,
    cfg: ChunkConfig,
    meter: OpMeterSlot,
    component: Arc<str>,
) -> Box<dyn ErasedBranch>
where
    F: RecFamily + 'static,
    E: RowEncoder<F> + Clone + Send + 'static,
    R: RecordRouter<F> + 'static,
{
    let encoder: BoxedEncoder<F> = Box::new(encoder);
    let router: BoxedRouter<F> = Box::new(router);
    let handoff: Branch<F> =
        SinkHandoff::new(encoder, router, queues, budget, cfg, meter, component);
    Box::new(handoff)
}

// ── The stack-borrowed emitter ──────────────────────────────────────────────

/// Stack-borrowed emitter handed to a [`route`](super::SplitBuilder::route)
/// closure.
/// [`emit`](Self::emit) routes one derived record to the branch named by a
/// [`SplitBranch<F>`] handle; a record that emits to no branch triggers the
/// split's `unmatched` policy. One `Any` downcast per emit, no per-call name
/// lookup.
pub struct SplitEmitter<'a> {
    branches: &'a mut [Box<dyn ErasedBranch>],
    meta: RecordMeta,
    ack: &'a AckRef,
    emitted: u32,
    flow: Flow,
    /// Whether an [`emit`](Self::emit) left a branch holding a fatal.
    latched: bool,
}

impl std::fmt::Debug for SplitEmitter<'_> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("SplitEmitter")
            .field("emitted", &self.emitted)
            .field("flow", &self.flow)
            .finish_non_exhaustive()
    }
}

impl SplitEmitter<'_> {
    /// Route one derived record to `handle`'s branch, inheriting the parent's
    /// metadata and acknowledgment handle. Emitting to no branch (returning
    /// from the closure without any `emit`) invokes the `unmatched` policy.
    /// Once a branch of this split holds a fatal, every later emit in the
    /// payload is dropped until the chain drains it.
    ///
    /// # Panics
    ///
    /// Panics if `handle`'s index lies past this split's branches, or its
    /// record family does not match the branch at that index. A handle minted
    /// by another split's builder whose index and family match a branch of
    /// this split routes to that branch without a panic.
    #[inline]
    pub fn emit<'buf, F: RecFamily + 'static>(
        &mut self,
        handle: SplitBranch<F>,
        row: F::Rec<'buf>,
    ) {
        let branch = self
            .branches
            .get_mut(handle.idx)
            .and_then(|b| b.as_any_mut().downcast_mut::<Branch<F>>())
            .expect(
                "split branch/handle mismatch: this SplitBranch<F> handle does not \
                 name a branch of this split (an index past its branches, or the \
                 wrong record family)",
            );
        // A row emitted after a branch of this split latched is dropped, as a
        // `flat_map`'s later records are once one `SinkHandoff` latches. The
        // handle resolves above this, so a mismatched one still panics.
        if self.latched {
            return;
        }
        let flow = branch.push(Record {
            payload: row,
            meta: self.meta,
            ack: self.ack.clone(),
        });
        self.latched = branch.fatal.0.is_some();
        self.emitted += 1;
        if self.flow != Flow::Blocked {
            self.flow = flow;
        }
    }

    /// The parent record's metadata.
    #[must_use]
    pub fn meta(&self) -> RecordMeta {
        self.meta
    }
}

// ── The terminal stage ──────────────────────────────────────────────────────

/// The chain's split terminal. Runs the route closure over each record and
/// aggregates the branches' lifecycle (relieve/flush/fatal). See the
/// [module docs](crate::ops).
pub struct SplitTerminal<SrcF: RecFamily, G> {
    route: G,
    branches: Vec<Box<dyn ErasedBranch>>,
    unmatched: ErrorPolicy,
    meter: OpMeterSlot,
    fatal: FatalSlot,
    /// Whether a branch holds a fatal, so the rest of the payload is dropped
    /// without running the route closure. Cleared with the branches' slots.
    latched: bool,
    component: Arc<str>,
    _family: PhantomData<fn() -> SrcF>,
}

impl<SrcF: RecFamily, G> SplitTerminal<SrcF, G> {
    pub(crate) fn new(
        route: G,
        branches: Vec<Box<dyn ErasedBranch>>,
        unmatched: ErrorPolicy,
        meter: OpMeterSlot,
        component: Arc<str>,
    ) -> Self {
        SplitTerminal {
            route,
            branches,
            unmatched,
            meter,
            fatal: FatalSlot(None),
            latched: false,
            component,
            _family: PhantomData,
        }
    }
}

impl<SrcF: RecFamily, G> std::fmt::Debug for SplitTerminal<SrcF, G> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("SplitTerminal")
            .field("branches", &self.branches.len())
            .field("unmatched", &self.unmatched)
            .finish_non_exhaustive()
    }
}

impl<'buf, SrcF, G> Collector<<SrcF as RecFamily>::Rec<'buf>> for SplitTerminal<SrcF, G>
where
    SrcF: RecFamily,
    G: for<'b> FnMut(SrcF::Rec<'b>, &mut SplitEmitter<'_>),
{
    fn push(&mut self, rec: Record<SrcF::Rec<'buf>>) -> Flow {
        self.meter.0.seen();
        // Records after a fatal are dropped until `take_fatal` drains the
        // slots at the end of the payload, this stage's own and every branch's.
        // Nothing past this return runs the route closure, so no branch is
        // written after the pipeline has decided to stop.
        if self.fatal.0.is_some() || self.latched {
            return Flow::Continue;
        }
        let Record {
            payload, meta, ack, ..
        } = rec;
        let mut em = SplitEmitter {
            branches: &mut self.branches,
            meta,
            ack: &ack,
            emitted: 0,
            flow: Flow::Continue,
            latched: false,
        };
        (self.route)(payload, &mut em);
        let (emitted, flow) = (em.emitted, em.flow);
        self.latched = em.latched;
        if emitted == 0 {
            match self.unmatched {
                // The record's ack share releases as success when `ack` drops
                // here, exactly like a `filter` drop.
                ErrorPolicy::Skip => self.meter.0.unrouted(),
                // The driver fails the batch's ack, stopping the pipeline.
                _ => {
                    self.meter.0.fatal_error();
                    self.fatal.0 = Some(FatalError {
                        component: self.component.to_string(),
                        reason: "record matched no split branch".into(),
                    });
                }
            }
        } else {
            self.meter.0.out_n(u64::from(emitted));
        }
        flow
    }
}

impl<SrcF: RecFamily, G> StageLifecycle for SplitTerminal<SrcF, G> {
    fn on_batch_end(&mut self, elapsed: Duration) {
        self.meter.0.flush(elapsed);
        for branch in &mut self.branches {
            branch.on_batch_end(elapsed);
        }
    }

    fn take_fatal(&mut self) -> Option<FatalError> {
        // Drain every slot, keeping the lowest index, which is
        // `flush_terminal`'s seal order. A branch outranks this stage's own
        // slot, since this stage swallows the rest of the payload once a
        // branch latches.
        self.latched = false;
        let mut earliest = None;
        for branch in &mut self.branches {
            let fatal = branch.take_fatal();
            if earliest.is_none() {
                earliest = fatal;
            }
        }
        earliest.or_else(|| self.fatal.0.take())
    }

    fn relieve(&mut self) -> Flow {
        // Make progress on every branch. One branch that stays blocked keeps
        // the chain from taking new payloads.
        let mut flow = Flow::Continue;
        for branch in &mut self.branches {
            if branch.relieve() == Flow::Blocked {
                flow = Flow::Blocked;
            }
        }
        flow
    }

    fn flush_terminal(&mut self) -> Flow {
        let mut flow = Flow::Continue;
        for branch in &mut self.branches {
            if branch.flush_terminal() == Flow::Blocked {
                flow = Flow::Blocked;
            }
        }
        flow
    }
}