Datum

Appearance

Substreams ​

Substream operators partition a single stream into a stream of streams. Each inner Source<Out> can be processed independently — folded, filtered, mapped — and then merged back into the outer stream.

flat_map_concat ​

flat_map_concat(f) maps each element to a sub-source and concatenates them strictly in order. The second sub-source does not start until the first has completed:

rust
use datum::{Sink, Source};

// flat_map_concat maps each element to a sub-source and concatenates them in order.
// The second sub-source only starts after the first has completed.
let items: Vec<u64> = Source::from_iter(1_u64..=3)
    .flat_map_concat(|x| Source::from_iter([x * 10, x * 10 + 1]))
    .run_with(Sink::collect())
    .unwrap()
    .wait()
    .unwrap();

This mirrors Akka's flatMapConcat. Use it when ordering matters and sub-sources are bounded.

flat_map_merge ​

flat_map_merge(breadth, f) runs up to breadth sub-sources concurrently and emits elements from whichever sub-source has output ready:

rust
use datum::{Sink, Source};

// flat_map_merge(breadth, f) runs up to `breadth` sub-sources concurrently.
// Output order across active sub-sources is not guaranteed; sort before asserting.
let mut items: Vec<u64> = Source::from_iter(1_u64..=3)
    .flat_map_merge(2, |x| Source::from_iter([x * 10, x * 10 + 1]))
    .run_with(Sink::collect())
    .unwrap()
    .wait()
    .unwrap();
items.sort_unstable();

Breadth 1 degenerates to flat_map_concat. Higher breadth increases parallelism at the cost of non-deterministic element ordering. The breadth parameter must be > 0.

prefix_and_tail ​

prefix_and_tail(n) emits a single (Vec<Out>, Source<Out>) pair. The vector contains the first n elements, and the tail source contains everything after that prefix:

rust
use datum::{Sink, Source};

// prefix_and_tail(n) emits one pair: the first n elements and a one-shot
// sub-source for the remaining tail.
let mut outer: Vec<(Vec<&str>, Source<&str>)> =
    Source::from_iter(["tenant-a", "csv", "row-1", "row-2"])
        .prefix_and_tail(2)
        .run_with(Sink::collect())
        .unwrap()
        .wait()
        .unwrap();

let (prefix, tail) = outer.pop().unwrap();
let rows: Vec<&str> = tail.run_with(Sink::collect()).unwrap().wait().unwrap();

assert_eq!(prefix, vec!["tenant-a", "csv"]);
assert_eq!(rows, vec!["row-1", "row-2"]);

The tail is a one-shot sub-source: materialize it once, process it like any other Source, and then let it complete. If upstream completes before n elements arrive, the prefix can be shorter and the tail is empty. If upstream fails before the prefix is ready, the outer stream fails.

flat_map_prefix ​

flat_map_prefix(n, f) consumes up to n prefix elements, calls f(prefix), and uses the returned Flow to process the remaining stream. Use it when early elements describe how the continuation should be decoded, routed, or enriched:

rust
use datum::{Flow, Sink, Source};

// flat_map_prefix(n, f) consumes the first n elements and builds the flow
// that will process the remaining elements from that prefix.
let rows: Vec<String> = Source::from_iter(["tenant-a", "csv", "row-1", "row-2"])
    .flat_map_prefix(2, |prefix| {
        let tenant = prefix[0].to_owned();
        let format = prefix[1].to_owned();
        Flow::identity().map(move |row: &str| format!("{tenant}:{format}:{row}"))
    })
    .run_with(Sink::collect())
    .unwrap()
    .wait()
    .unwrap();

assert_eq!(
    rows,
    vec![
        "tenant-a:csv:row-1".to_owned(),
        "tenant-a:csv:row-2".to_owned(),
    ]
);

The prefix is not emitted unless the returned flow emits it explicitly, for example by prepending a source. Because the continuation is built from captured prefix values, flat_map_prefix requires the input element type to be Clone.

split_when and split_after ​

split_when(pred) starts a new sub-source each time the predicate returns true. The element that triggers the split begins the new segment (the old segment closes before that element):

rust
use datum::{Sink, Source};

// split_when(pred) starts a new sub-source when pred returns true.
// The triggering element begins the new segment (it is NOT included in the old one).
// flat_map_concat(|sub| sub) flattens segments back in arrival order.
let items: Vec<u64> = Source::from_iter(1_u64..=6)
    .split_when(|&x| x == 4) // split before 4: segments [1,2,3] and [4,5,6]
    .flat_map_concat(|sub| sub)
    .run_with(Sink::collect())
    .unwrap()
    .wait()
    .unwrap();

split_after(pred) is the complement: the triggering element ends the current segment (inclusive), and the next element opens a new one. Use split_when when the boundary marker logically belongs to the next segment, split_after when it logically closes the current one.

Sub-sources from split_when/split_after are consumed with flat_map_concat(|sub| sub) to preserve segment ordering, or flat_map_merge(n, |sub| sub) to process segments concurrently. Apply Source::fold to each sub-source to reduce a segment to a single value:

rust
source
    .split_when(|&x| x == boundary)
    .flat_map_concat(|sub| sub.fold(0_u64, |acc, x| acc + x))

group_by ​

group_by(max_substreams, key_fn, allow_closed_substream_recreation) partitions the stream by a key function. Elements with the same key flow into the same sub-source:

rust
use datum::{Sink, Source};

// group_by(max_substreams, key_fn, allow_recreation) partitions elements by key.
// Each group's sub-source can be folded or transformed independently.
let mut sums: Vec<u64> = Source::from_iter(1_u64..=6)
    .group_by(2, |x| *x % 2, false) // two keys: 0 (even), 1 (odd)
    .flat_map_merge(2, |sub| sub.fold(0_u64, |acc, x| acc + x))
    .run_with(Sink::collect())
    .unwrap()
    .wait()
    .unwrap();
sums.sort_unstable();

Parameters:

  • max_substreams: the maximum number of concurrent open sub-sources. Exceeding this limit fails the stream.
  • key_fn: Fn(&Out) -> Key — must return a Clone + Eq + Hash + Send + 'static key.
  • allow_closed_substream_recreation: if true, a key that already had a sub-source that closed can open a new one when it reappears. If false, reappearance of a closed key fails the stream.

Sub-sources from group_by behave as live streams: they receive elements as the outer stream progresses and complete when no more elements for that key will arrive (i.e. when the outer stream completes). Use flat_map_merge with a concurrency equal to max_substreams to process all groups simultaneously.

Performance ​

All substream operators are at parity or better vs Akka. The substream perf gate closed in v0.2.0; subsequent inline-path optimization (WP-19c) widened the lead further. Notable observations:

  • split_when/split_after and flat_map_concat are significantly faster.
  • flat_map_merge at moderate breadth is competitive.
  • group_by is significantly faster on single-key workloads.

Full per-scenario numbers are maintained in roadmap/benchmarks/substreams.md.

Next steps ​