Data product pipelines

Pipelines are expressions assigned at module scope—usually as constants named flow, joined, result, or arbitrary identifiers—whose values are streams consumed by .publish(...). The type checker and runtime decide which methods exist on which stream types. The tables below map surface syntax to common patterns.

Publishing

Construct	Typical pattern	Notes
.publish(Outlet)	Map or filter, then publish	Terminal side effect to a wired outlet
.publish(KeyValueApi)	Command or update stream to a KV API	Stream of commands or typed updates
Re-publish same stream	One stream, two outlets	Bind a = …, then a.publish(A), a.publish(B)

Transformations on streams

Method / call	Typical pattern	Notes
.map(closure)	One-to-one transform	Map each element
.filter(closure)	Keep matching elements	Predicate
.filterMap(closure)	Map and drop empty	Often returns Some/None via match
.filter then .map	Two-step chain in one flow	Equivalent to chained operators
.flatMap(closure)	After a join	Flatten nested collections from join results
.forEach(closure)	Side-effect per element	Often with {} body
.indexBy(key) then .map(r -> r.latestValue())	Latest (or windowed) value	earliestValue, currentValue, and previousValue may be available depending on stream metadata
.aggregateBy(key, initial, reducer)	Per-key stateful accumulation	Emits AggregateUpdate<K,V> on every event; reducer returns Option<V> (None removes the key)
innerJoin(left, right, lk, rk)	Equi-join on keys	Key extractors return lists
com.dataspine.interleave(A, B)	Merge two ingests	Namespace function on two streams
Join result .map / inner .map	Nested map on join rows	Nested jr.map on pair

Key-value commands

Construct	Typical use
KeyValueApiCommand.Upsert(key, value)	Upsert into a KV API
Ingest typed as KeyValueApiCommand<K,V>	Command stream including deletes

aggregateBy — keyed stateful accumulation

DataStream<T>.aggregateBy<K, V>(
    keyMapper:  (T) => K,
    initial:    () => V,
    reducer:    (V, T) => Option<V>,
): DataStream<AggregateUpdate<K, V>>

For each input event the operator:

1. Extracts the key with keyMapper.
2. Looks up the current accumulator in a per-key store, or falls back to initial() if the key is unseen.
3. Calls reducer(accumulator, event) to produce the next value.
4. Some(v) → store the updated accumulator and emit an update.
   None → remove the key from the store (tombstone) and emit a deletion.

Emitted shape

Struct AggregateUpdate<K, V> {
    key:   K,
    value: AggregateChange<V>,
}

Enum AggregateChange<V> {
    Added(V),                              // first event for this key
    Changed { previous: V, current: V },   // key existed and was updated
    Deleted(V),                            // reducer returned None; V is the prior value
}

Silent no-op: if reducer returns None for a key that was never added, nothing is emitted.

Accessor methods

Expression	Return type	Description
update.key()	K	The key
update.value()	AggregateChange<V>	The change variant
change.currentValue()	Option<V>	Some(v) for Added/Changed, None for Deleted
change.previousValue()	Option<V>	None for Added, Some(v) for Changed/Deleted

Example — running total per user

Struct Event { id: String, amount: Int64 }
Struct Total  { sum: Int64 }

flow = EventIngest
    .aggregateBy(
        event -> event.id,
        () -> Total { sum: 0 },
        (acc, event) -> Some(Total { sum: acc.sum + event.amount })
    )
    .filterMap(update -> update.value().currentValue())
    .publish(TotalOutlet)

Example — delete on sentinel event

flow = EventIngest
    .aggregateBy(
        event -> event.id,
        () -> Total { sum: 0 },
        (acc, event) -> if event.close then None
                        else Some(Total { sum: acc.sum + event.amount })
    )
    .filterMap(update -> update.value().currentValue())
    .publish(TotalOutlet)

When event.close is true the key is removed from the store. A subsequent event for the same key starts fresh from initial().

Why not indexBy?

indexBy stores the entire input event keyed by an ID and emits IndexOperationResult<K, T>. aggregateBy stores a derived accumulator of a potentially different type V and applies a user-provided reduction on every event — use it when you want running sums, counts, or any stateful reduction rather than a direct key-value index of raw events.

Types that often appear in pipelines

Type	Role
JoinTrigger<K,V>	Join keys and change metadata
JoinResult<T> / List<JoinResult<...>>	Rows produced by innerJoin
com.dataspine.json.Value	JSON-heavy or semi-structured payloads

Ordering and emptiness

Topic	Meaning
Multiple messages	Order of ingests affects downstream snapshots when testing step-by-step
Empty ingest	Products should define behaviour when no rows arrive

Exact generics and error messages come from dataspine check; this page focuses on syntax patterns for flows.