Batch representation

A batch is an ordered sequence of logical ticks. Internally it is stored as a flat list of segments, and each segment holds its updates in one of three layouts. Every public operation constructs, reads, or transforms a batch through its logical ticks, so the choice of layout never changes how many ticks a batch contains, their order, or which updates are simultaneous.

This page explains that representation for both directions. InputBatch and OutputBatch are separate public types, but they share the storage and iteration code in src/core/batch.rs.

Scope of the representation

Sources and runtimes produce data in different native shapes. An MQTT message carries one value, a channel source joins values that arrived together, a file yields table rows, and the dataflow runtime completes one output row per model step. The batch representation lets each producer keep its native shape while every consumer sees the same ordered ticks.

Logical and physical units

Both update types are aliases of one shared definition:

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Update<V> {
    pub variable: VarName,
    pub value: V,
}

A tick has no owned type of its own: owned code represents one as Vec<Update<V>>, and borrowed code uses the Tick view described under Iterating a batch.

A running example

The rest of this page uses one batch of four ticks:

It contains four ticks and seven updates. The example builds ticks 0 and 1 from listed ticks and ticks 2 and 3 from fixed-layout rows, then concatenates the two batches:

fn build_a_mixed_batch() -> anyhow::Result<trustworthiness_checker::OutputBatch<i32>> {
    use trustworthiness_checker::{OutputBatch, OutputUpdate, VarName};

    // Ticks 0 and 1: one independent update, then one simultaneous pair.
    let listed = OutputBatch::from_ticks(vec![
        vec![OutputUpdate::new("x".into(), 1)],
        vec![
            OutputUpdate::new("x".into(), 2),
            OutputUpdate::new("y".into(), 20),
        ],
    ])?;
    // Ticks 2 and 3: two rows sharing the layout [x, y], stored as one flat buffer.
    let layout: Vec<VarName> = vec!["x".into(), "y".into()];
    let rows = OutputBatch::packed_rows(layout, [3, 30, 4, 40])?;

    let batch = listed.concat(rows)?;

    assert_eq!(batch.tick_count(), 4);
    assert_eq!(batch.update_count(), 7);
    assert_eq!(batch.len(), 7); // `len` counts updates, not ticks
    Ok(batch)
}

len returns the update count, while is_empty asks whether the batch has any ticks. The example uses OutputBatch because packed-row construction is public only for output; InputBatch::packed_rows is available inside the crate to file and in-memory sources.

The resulting batch stores each part of the example in the layout it was built with:

Reading rule. Horizontal position is logical tick order, and a cell below a tick mark belongs to that tick. The PackedRows segment stores ticks 2 and 3 in one contiguous value buffer; each row of its layout is one tick. The bracket shows that all three segments sit side by side in one list, with no batch nested inside another.

Segments

Each segment kind groups its values into ticks differently:

SingletonTicks and Tick hold the same data, a list of updates, and differ only in how that list divides into ticks. PackedRows stores each variable name once, in its layout, instead of once per value. The input definition, which is internal to the crate, documents each kind:

/// A physical segment in an [`InputBatch`]. A segment is never itself a
/// batch: it is only one representation of an ordered range of logical ticks.
#[derive(Clone, Debug, PartialEq, Eq)]
pub(crate) enum InputSegment<V> {
    /// Each update is an independent width-one logical tick.
    SingletonTicks(Vec<InputUpdate<V>>),
    /// Exactly one simultaneous logical tick.
    Tick(Vec<InputUpdate<V>>),
    /// Fixed-width row-major ticks sharing one variable layout.
    PackedRows {
        layout: Box<[VarName]>,
        values: Vec<V>,
    },
}

The output segment has the same three kinds. Its packed layout is a ValidatedLayout, which is checked once and shared by reference between batches:

/// A physical output segment. A segment is never itself a batch: it is one
/// representation of an ordered range of logical output ticks.
#[derive(Clone, Debug, PartialEq)]
enum OutputSegment<V> {
    /// Each update is an independent width-one logical tick.
    SingletonTicks(Vec<OutputUpdate<V>>),
    /// Exactly one simultaneous logical tick.
    Tick(Vec<OutputUpdate<V>>),
    /// Fixed-width row-major ticks sharing one validated variable layout.
    /// Valid packed segments contain at least one complete row.
    PackedRows {
        layout: ValidatedLayout,
        values: Vec<V>,
    },
}

Construction validates each segment:

• A Tick must be non-empty and name each variable at most once.
• A PackedRows layout must be non-empty and name each variable at most once, and the value count must be a whole number of rows.
• A valid output PackedRows segment contains at least one row; OutputBatch::empty() represents empty output. An input PackedRows segment may have no rows: the crate-private InputBatch::packed_rows uses one to represent an empty batch with a known layout.
• SingletonTicks has no shape rule: each update is a complete tick. A variable may therefore appear in several successive singleton ticks.

Storage

A batch owns one Storage value. The single-segment case avoids allocating a list, which is the common shape for one message or one row:

#[derive(Clone, Debug, PartialEq, Eq)]
pub(crate) enum Storage<S> {
    Single(S),
    Segments(Vec<S>),
}

/// An ordered sequence of logical input ticks.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct InputBatch<V> {
    storage: InputBatchStorage<V>,
}

Constructors choose the initial layout:

When batches are combined, segments without ticks are dropped and the remaining segments keep their order. The two directions then differ:

• Input keeps adjacent segments separate. InputBatch::concat moves whole segments into one list. The input-window collector also joins segments without merging them, although it may split a segment between ticks to meet a window limit.
• Output merges adjacent SingletonTicks segments, and adjacent PackedRows segments with the same layout. OutputBatch::concat and the crate-private append used by output coalescing both apply these rules.

Neither direction merges two ticks. Two batches with the same ticks can still differ in storage, and because batch equality compares storage, such batches do not compare equal.

Iterating a batch

A batch can be read in three ways. They differ in whether tick boundaries survive and whether packed rows are expanded:

Their costs differ:

• Borrowed ticks (InputTicks, OutputTicks) allocate nothing. A packed row becomes a tick that refers into the value buffer.
• Borrowed updates (InputUpdates, OutputUpdates) allocate nothing, and list the updates of every tick one after another. A packed value takes its variable name from its layout column.
• Owned ticks (OwnedInputTicks, OwnedOutputTicks) consume the batch and build one Vec per tick. This is the only place packed rows are expanded.

All three yield items in logical order: segment order, then position within a segment. Each is an ExactSizeIterator whose length comes from the batch's tick or update count, and none is double-ended.

A borrowed tick is a small Copy handle. It refers either to a slice of updates or to one packed row and its layout, and UpdateRef borrows one variable and value from either:

#[derive(Clone, Copy, Debug, PartialEq)]
pub struct UpdateRef<'a, V> {
    pub variable: &'a VarName,
    pub value: &'a V,
}

#[derive(Clone, Copy, Debug)]
enum TickRepresentation<'a, V> {
    Slice(&'a [Update<V>]),
    Packed {
        layout: &'a [VarName],
        values: &'a [V],
    },
}

#[derive(Clone, Copy, Debug)]
pub struct Tick<'a, V> {
    representation: TickRepresentation<'a, V>,
}

tick.updates() (or tick.iter()) yields the tick's UpdateRefs, tick.len() is its width, and tick.to_updates() clones it into a Vec<Update<V>>.

Iterating borrowed ticks gives the same result whichever segment stores a tick:

fn iterate_borrowed_ticks(batch: &trustworthiness_checker::OutputBatch<i32>) {
    use trustworthiness_checker::VarName;

    // Each item is one logical tick, whatever segment stores it.
    let widths: Vec<usize> = batch.ticks().map(|tick| tick.len()).collect();
    assert_eq!(widths, [1, 2, 2, 2]);

    // `&OutputBatch` iterates the same borrowed ticks.
    let mut updates = 0;
    for tick in batch {
        updates += tick.updates().count();
    }
    assert_eq!(updates, 7);

    // Tick 3 is read directly from the packed value buffer.
    let (x, y) = (VarName::from("x"), VarName::from("y"));
    let last = batch.ticks().last().expect("four ticks");
    let values: Vec<(&VarName, i32)> = last
        .updates()
        .map(|update| (update.variable, *update.value))
        .collect();
    assert_eq!(values, [(&x, 4), (&y, 40)]);
}

updates() suits code that needs every value but not their grouping, such as checking which variables a batch names. It cannot show which values were simultaneous, so evaluation code iterates ticks instead:

fn iterate_flat_updates(batch: &trustworthiness_checker::OutputBatch<i32>) {
    // `updates()` concatenates every tick, so it cannot show which values were simultaneous.
    let names: Vec<String> = batch
        .updates()
        .map(|update| update.variable.name())
        .collect();
    let values: Vec<i32> = batch.updates().map(|update| *update.value).collect();
    assert_eq!(names, ["x", "x", "y", "x", "y", "x", "y"]);
    assert_eq!(values, [1, 2, 20, 3, 30, 4, 40]);
}

Consuming a batch yields owned ticks; tick 2, stored as a packed row, becomes an ordinary list of updates:

fn take_owned_ticks(batch: trustworthiness_checker::OutputBatch<i32>) {
    use trustworthiness_checker::OutputUpdate;

    // Consuming the batch builds one `Vec` per tick; packed rows are expanded here.
    let ticks: Vec<Vec<OutputUpdate<i32>>> = batch.into_ticks().collect();
    assert_eq!(ticks.len(), 4);
    assert_eq!(
        ticks[2],
        [
            OutputUpdate::new("x".into(), 3),
            OutputUpdate::new("y".into(), 30),
        ]
    );
}

The two directions expose iteration differently. OutputBatch implements IntoIterator by value (owned ticks) and by reference (borrowed ticks), and into_ticks is public. InputBatch publicly offers the borrowed ticks() and updates() views:

fn iterate_input_batch() -> anyhow::Result<()> {
    use trustworthiness_checker::{InputBatch, InputUpdate, VarName};

    let batch = InputBatch::update("x", 1).concat(InputBatch::tick(vec![
        InputUpdate::new("x".into(), 2),
        InputUpdate::new("y".into(), 20),
    ])?)?;

    // An input batch offers the same borrowed tick and update views.
    let (x, y) = (VarName::from("x"), VarName::from("y"));
    let ticks: Vec<Vec<(&VarName, i32)>> = batch
        .ticks()
        .map(|tick| {
            tick.updates()
                .map(|update| (update.variable, *update.value))
                .collect()
        })
        .collect();
    assert_eq!(ticks, [vec![(&x, 1)], vec![(&x, 2), (&y, 20)]]);
    Ok(())
}

Its into_ticks is crate-private. Runtimes expand input ticks at their own boundary through it: borrowed_tick_stream turns the semi-synchronous runtime's input into one item per tick, into_tick_stream does the same for the distributed runtime, and the asynchronous input fanout and reconfigurable semi-synchronous runtime call into_ticks directly.

The dataflow runtime adapter does not iterate ticks when a batch is a single packed segment. It reads the layout and value buffer through the crate-private packed_rows_segment and evaluates each row in place, falling back to ticks() for every other shape; see Runtime adapter.

Operations that preserve ticks

Most operations change storage, if anything, but leave the ticks alone. One input policy deliberately creates a new tick:

Selecting a variable removes only the ticks that would become empty:

fn select_one_variable(batch: trustworthiness_checker::OutputBatch<i32>) -> anyhow::Result<()> {
    use std::collections::BTreeSet;

    // Tick 0 has no `y`, so it disappears; the other ticks keep only their `y` value.
    let only_y = batch.select_variables(&BTreeSet::from(["y".into()]))?;
    let widths: Vec<usize> = only_y.ticks().map(|tick| tick.len()).collect();
    assert_eq!(widths, [1, 1, 1]);
    let values: Vec<i32> = only_y.updates().map(|update| *update.value).collect();
    assert_eq!(values, [20, 30, 40]);
    Ok(())
}

The input architecture describes windows, and the output architecture describes destination selection and coalescing.

Implementation mapping

• Shared update, storage, borrowed tick and update iterators, and owned tick iterator: Update, Storage, normalize, Tick, UpdateRef, Ticks, Updates, and OwnedTicks in src/core/batch.rs.
• Input segments, validation, construction, selection, mapping, and tick streams: InputSegment, InputBatch, into_tick_stream, and borrowed_tick_stream in src/core/input.rs.
• Output segments, validated layouts, merging, and IntoIterator: OutputSegment, OutputBatch, and storage_from_segments in src/core/output.rs; ValidatedLayout in src/core/layout.rs.
• Packed-row sources: src/io/file/input_stream.rs and src/io/map/input_stream.rs.
• Input windows, including their segment-preserving batching and tick-creating reduction: drive_window in src/io/aggregation.rs. Output coalescing: src/io/output/delivery.rs.
• Runtime producers and consumers: consume_singleton_streams and consume_row_streams in src/runtime/output_utils.rs; the packed-row path in src/runtime/dataflow.rs.
• Examples on this page: tests/docs_examples.rs.