Sink sharding

The sink is partitioned into shards, and the chain's terminal stage sends every record to exactly one of them. Sharding is how a single pipeline process fans writes across a cluster: more shards means more concurrent insert streams, and — when it matters — a record's shard can be made to match where the destination itself expects that record to live.

What a shard is

A shard is one lane of the sink's write topology, identified by an index in 0..num_shards. That index is a single identity threaded end to end:

 shard i  =  the i-th entry in the sink config  (config order is the contract)
          =  one shard worker task on the I/O runtime
          =  one endpoint (or replica set) it writes to
          =  one dedup-token namespace

The route → encode → queue path and the per-shard workers that drain those queues are the terminal stage in Architecture. Because the index means the same thing everywhere, the order you list shards in config is load-bearing: shard i is the config's i-th entry, the i-th worker, and the i-th endpoint. Reordering the list re-homes data. A connector that mirrors an external cluster's placement can verify parts of this contract at startup and warn (advisorily) on apparent mis-ordering — see the ClickHouse distributed_check — but the ordering itself stays yours to get right.

Routing happens once, at the terminal stage

Routing is not a pipeline operator. It runs exactly once, per emitted record, at the chain's terminal .sink(...) stage — after every map / filter / flat_map, and strictly before the record is encoded to wire format. The router sees the record exactly as the encoder will.

That "per emitted record, post-transform" timing is the whole reason the record-aware tier exists. A flat_map that explodes one source record into many children gives every child the same RecordMeta as its parent (the ack fan-out relies on this — see Delivery guarantees). So a router that reads only metadata necessarily colocates all children of one parent on one shard. To route those children independently, the router has to look at each child's own payload.

Two router tiers

One builder seam accepts either tier; you implement exactly one.

Meta-only — ShardRouter, routing on RecordMeta alone (key hash, source partition). The default KeyHashRouter lives here: it routes by the record's key hash and falls back to a stable per-partition hash for keyless records, keeping a partition's keyless records together and the distribution stable. This is the right default and stays front and center.
Record-aware — RecordRouter<F>, routing on the payload itself. Reach for it when shard affinity derives from a field of the terminal record type (matching a destination cluster's own sharding expression, for instance), or when you need flat_map children routed by their own fields.

The record-aware trait mirrors the row encoder's family-generic shape:

pub trait RecordRouter<F: RecFamily>: Send + Sync {
    fn route_record<'buf>(&self, rec: &Record<F::Rec<'buf>>, num_shards: usize) -> usize;
}

A minimal implementation over your own record family (here a borrowing family whose row is Event<'buf>; an owned family would read Record<Event>):

use etl::record::Record;
use etl::sink::RecordRouter;

struct BySensor;

impl RecordRouter<EventF> for BySensor {
    fn route_record<'buf>(&self, rec: &Record<Event<'buf>>, num_shards: usize) -> usize {
        // Any deterministic, stable hash of the routing field — determinism
        // across retries is a correctness requirement, not a nicety.
        (stable_hash(rec.payload.sensor) % num_shards as u64) as usize
    }
}

Every ShardRouter is automatically a RecordRouter for every family through a blanket bridge, so meta-only routers plug into the same seam unchanged. The bridge is also why you implement one tier, never both: implementing both is a coherence overlap the compiler rejects.

[!NOTE] Routers are on the per-record hot path: pure, cheap, no I/O, no blocking, and no per-call allocation. A stateful strategy (a weights table, a counter) holds its state behind &self and interior mutability.

Backpressure: one slow shard pauses the pipeline

Shards do not share a queue. Each has its own bounded chunk queue between the pipeline threads and its worker. When one shard's destination slows down, its queue fills, and the terminal stage's try_send for that shard starts failing — which pauses the source lanes rather than blocking them (the source never blocks on a channel send). So a single saturated shard back-pressures the whole pipeline.

This is deliberate: bounded memory over unbounded buffering. The alternative — letting healthy shards race ahead while one shard's backlog grows without limit — trades a bounded, observable slowdown for an unbounded memory leak and an ever-widening watermark gap. See Backpressure for the pause/resume machinery and the sizing rule.

The honest part: hot shards

[!IMPORTANT] A hot shard is your data's shape, not the router's choice. Any router that gives a key stable shard affinity — which is the entire point of payload routing — sends every record for a hot key to the same shard. If one key (or a handful) dominates your volume, that shard runs hot, and no router setting fixes it: affinity makes the skew irreducible by design.

What you can do:

Choose a high-cardinality routing key. Skew comes from few, heavy key values. A key with many roughly-even values spreads naturally; a low- cardinality key (a region code, a status flag) concentrates.
Watch the skew. It is already observable per shard via etl_sink_records_total{shard} and etl_sink_bytes_total{shard} — a persistent imbalance across shard labels is a hot shard (see docs/METRICS.md).
Reshard deliberately. Adding or removing shards changes num_shards and therefore reshuffles which shard each key maps to. It rebalances, but it is a topology change, not a live knob — it re-homes keys, with the consequences below.

CPU is not the victim here: routing runs after transform, per lane, so the hot path stays spread across pipeline threads. It is the one shard's write throughput, and the pipeline-wide backpressure it induces, that a hot shard costs you.

Deterministic routing is a correctness requirement

The router must be a pure function of the record: the same record routes to the same shard on every processing — across restarts, rebalances, and replays. Two guarantees depend on it:

Duplicates must colocate to be collapsible. At-least-once delivery means a reprocessed record lands again, and it lands with new batch boundaries and a new dedup token — no dedup window catches it (see Delivery guarantees). What absorbs those duplicates is a destination-side collapse pattern (ReplacingMergeTree-style last-write-wins), and collapse works within one shard's table: it can only absorb a duplicate that lands on the same shard as the original. A nondeterministic router scatters a key's duplicates across shards, where nothing ever collapses them.
Reads must find a key where the routing function says it lives. Anything that maps a key to a location at read time — a pruned SELECT through a parity router, a per-shard dedup query — assumes every record for that key sits on the shard the function names. One nondeterministically-placed record is silently invisible to such reads.

The dedup token's own role is narrower than either of these: a token deduplicates same-batch retries within one shard's replica set. A sealed batch is bound to one shard's worker and retried under one token — routing runs strictly before batches are sealed and never re-runs for a retry — so tokens need no help from the router. It is reprocessing, which tokens do not cover, that makes deterministic placement a correctness requirement.

The corollary: any change that re-homes keys to different shards — a resharding, a weight change, a routing-key change — is not covered by the collapse mechanism. Rows already written stay where they are; nothing merges across shards, so the old shard's copies are never absorbed: unpruned reads return both homes' rows indefinitely, while pruned reads see only the new home. Treat re-homing as a deliberate migration — backfill or rebuild the moved keys' rows, or delete the old copies — not as replay noise that settles. A plain restart with an unchanged topology is the benign case: replayed records re-route to the same shard, where the collapse pattern absorbs them.

What a shard is​

Routing happens once, at the terminal stage​

Two router tiers​

Backpressure: one slow shard pauses the pipeline​

The honest part: hot shards​

Deterministic routing is a correctness requirement​

Further reading​